Repository: apache/pulsar-client-go
Pull request #785 (2022-06-07 05:28:28)

Fix using closed connection in consumer
### Motivation
Related to https://github.com/apache/pulsar-client-go/pull/761

This PR fixed the issue using closed connection at consumer side.

If same issue as https://github.com/apache/pulsar-client-go/pull/761 happens at consumer side, consumer will report the following errors and can't consume any messages.
```
ERRO[0180] [Connection was closed when request flow cmd]  consumerID=1 name=wzrtq subscription=sub1 topic="persistent://public/default/topic-go-1"
ERRO[0180] [unable to send initial permits to broker]    consumerID=1 error="connection closed" name=wzrtq subscription=sub1 topic="persistent://public/default/topic-go-1"
```

### Modifications
* return an error when unable to add consumer handler.




Patch:
diff --git a/.github/workflows/go.yml b/.github/workflows/go.yml
index a3c4c60..1819b53 100644
--- a/.github/workflows/go.yml
+++ b/.github/workflows/go.yml
@@ -6,7 +6,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        go-version: [1.15, 1.16, 1.17]
+        go-version: [1.15, 1.16, 1.17, 1.18]
     steps:
       - name: clean docker cache
         run: |
diff --git a/.github/workflows/project.yml b/.github/workflows/project.yml
index 810979a..ce8545b 100644
--- a/.github/workflows/project.yml
+++ b/.github/workflows/project.yml
@@ -6,7 +6,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        go-version: [1.15, 1.16, 1.17]
+        go-version: [1.15, 1.16, 1.17, 1.18]
     steps:
       - name: Set up Go
         uses: actions/setup-go@v1
@@ -19,7 +19,7 @@ jobs:
 
       - name: InstallTool
         run: |
-          wget -O - -q https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh| sh -s v1.18.0
+          wget -O - -q https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh| sh -s v1.46.2
           ./bin/golangci-lint --version
 
       - name: Build
diff --git a/.golangci.yml b/.golangci.yml
index e87841b..525d34c 100644
--- a/.golangci.yml
+++ b/.golangci.yml
@@ -10,10 +10,9 @@ linters:
     - bodyclose
     - deadcode
     - goimports
-    - golint
+    - revive
     - gosimple
     - govet
-    - interfacer
     - misspell
     - structcheck
     - stylecheck
diff --git a/pulsar/consumer_partition.go b/pulsar/consumer_partition.go
index b7c2c8e..4cc1645 100644
--- a/pulsar/consumer_partition.go
+++ b/pulsar/consumer_partition.go
@@ -756,20 +756,20 @@ func createEncryptionContext(msgMeta *pb.MessageMetadata) *EncryptionContext {
 		encCtx.CompressionType = CompressionType(*msgMeta.Compression)
 	}
 
-	kMap := map[string]EncryptionKey{}
+	keyMap := map[string]EncryptionKey{}
 	for _, k := range msgMeta.GetEncryptionKeys() {
 		metaMap := map[string]string{}
 		for _, m := range k.GetMetadata() {
 			metaMap[*m.Key] = *m.Value
 		}
 
-		kMap[*k.Key] = EncryptionKey{
+		keyMap[*k.Key] = EncryptionKey{
 			KeyValue: k.GetValue(),
 			Metadata: metaMap,
 		}
 	}
 
-	encCtx.Keys = kMap
+	encCtx.Keys = keyMap
 	return &encCtx
 }
 
@@ -1187,7 +1187,11 @@ func (pc *partitionConsumer) grabConn() error {
 
 	pc._setConn(res.Cnx)
 	pc.log.Info("Connected consumer")
-	pc._getConn().AddConsumeHandler(pc.consumerID, pc)
+	err = pc._getConn().AddConsumeHandler(pc.consumerID, pc)
+	if err != nil {
+		pc.log.WithError(err).Error("Failed to add consumer handler")
+		return err
+	}
 
 	msgType := res.Response.GetType()
 
diff --git a/pulsar/internal/batch_builder.go b/pulsar/internal/batch_builder.go
index 9d18f26..fb7598e 100644
--- a/pulsar/internal/batch_builder.go
+++ b/pulsar/internal/batch_builder.go
@@ -157,6 +157,10 @@ func (bc *batchContainer) IsFull() bool {
 
 // hasSpace should return true if and only if the batch container can accommodate another message of length payload.
 func (bc *batchContainer) hasSpace(payload []byte) bool {
+	if bc.numMessages == 0 {
+		// allow to add at least one message when batching is disabled
+		return true
+	}
 	msgSize := uint32(len(payload))
 	return bc.numMessages+1 <= bc.maxMessages && bc.buffer.ReadableBytes()+msgSize <= uint32(bc.maxBatchSize)
 }
diff --git a/pulsar/internal/connection.go b/pulsar/internal/connection.go
index fa8d055..0249020 100644
--- a/pulsar/internal/connection.go
+++ b/pulsar/internal/connection.go
@@ -53,8 +53,9 @@ type TLSOptions struct {
 }
 
 var (
-	errConnectionClosed       = errors.New("connection closed")
-	errUnableRegisterListener = errors.New("unable register listener when con closed")
+	errConnectionClosed        = errors.New("connection closed")
+	errUnableRegisterListener  = errors.New("unable register listener when con closed")
+	errUnableAddConsumeHandler = errors.New("unable add consumer handler when con closed")
 )
 
 // ConnectionListener is a user of a connection (eg. a producer or
@@ -75,7 +76,7 @@ type Connection interface {
 	WriteData(data Buffer)
 	RegisterListener(id uint64, listener ConnectionListener) error
 	UnregisterListener(id uint64)
-	AddConsumeHandler(id uint64, handler ConsumerHandler)
+	AddConsumeHandler(id uint64, handler ConsumerHandler) error
 	DeleteConsumeHandler(id uint64)
 	ID() string
 	GetMaxMessageSize() int32
@@ -994,16 +995,17 @@ func (c *connection) getTLSConfig() (*tls.Config, error) {
 	return tlsConfig, nil
 }
 
-func (c *connection) AddConsumeHandler(id uint64, handler ConsumerHandler) {
+func (c *connection) AddConsumeHandler(id uint64, handler ConsumerHandler) error {
 	// do not add if connection is closed
 	if c.closed() {
 		c.log.Warnf("Closed connection unable add consumer with id=%+v", id)
-		return
+		return errUnableAddConsumeHandler
 	}
 
 	c.consumerHandlersLock.Lock()
 	defer c.consumerHandlersLock.Unlock()
 	c.consumerHandlers[id] = handler
+	return nil
 }
 
 func (c *connection) DeleteConsumeHandler(id uint64) {
diff --git a/pulsar/internal/key_based_batch_builder.go b/pulsar/internal/key_based_batch_builder.go
index d09138c..667e855 100644
--- a/pulsar/internal/key_based_batch_builder.go
+++ b/pulsar/internal/key_based_batch_builder.go
@@ -117,6 +117,11 @@ func (bc *keyBasedBatchContainer) IsMultiBatches() bool {
 
 // hasSpace should return true if and only if the batch container can accommodate another message of length payload.
 func (bc *keyBasedBatchContainer) hasSpace(payload []byte) bool {
+	if bc.numMessages == 0 {
+		// allow to add at least one message
+		// and a single max message size is checked in the producer partition, therefore no need to validate batch size
+		return true
+	}
 	msgSize := uint32(len(payload))
 	return bc.numMessages+1 <= bc.maxMessages && bc.buffer.ReadableBytes()+msgSize <= uint32(bc.maxBatchSize)
 }
diff --git a/pulsar/schema.go b/pulsar/schema.go
index 8021ffe..42499ad 100644
--- a/pulsar/schema.go
+++ b/pulsar/schema.go
@@ -91,18 +91,29 @@ type JSONSchema struct {
 	SchemaInfo
 }
 
+// NewJSONSchema creates a new JSONSchema
+// Note: the function will panic if creation of codec fails
 func NewJSONSchema(jsonAvroSchemaDef string, properties map[string]string) *JSONSchema {
+	js, err := NewJSONSchemaWithValidation(jsonAvroSchemaDef, properties)
+	if err != nil {
+		log.Fatalf("JSONSchema init codec error:%v", err)
+	}
+	return js
+}
+
+// NewJSONSchemaWithValidation creates a new JSONSchema and error to indicate codec failure
+func NewJSONSchemaWithValidation(jsonAvroSchemaDef string, properties map[string]string) (*JSONSchema, error) {
 	js := new(JSONSchema)
 	avroCodec, err := initAvroCodec(jsonAvroSchemaDef)
 	if err != nil {
-		log.Fatalf("init codec error:%v", err)
+		return nil, err
 	}
 	schemaDef := NewSchemaDefinition(avroCodec)
 	js.SchemaInfo.Schema = schemaDef.Codec.Schema()
 	js.SchemaInfo.Type = JSON
 	js.SchemaInfo.Properties = properties
 	js.SchemaInfo.Name = "JSON"
-	return js
+	return js, nil
 }
 
 func (js *JSONSchema) Encode(data interface{}) ([]byte, error) {
@@ -126,11 +137,22 @@ type ProtoSchema struct {
 	SchemaInfo
 }
 
+// NewProtoSchema creates a new ProtoSchema
+// Note: the function will panic if creation of codec fails
 func NewProtoSchema(protoAvroSchemaDef string, properties map[string]string) *ProtoSchema {
+	ps, err := NewProtoSchemaWithValidation(protoAvroSchemaDef, properties)
+	if err != nil {
+		log.Fatalf("ProtoSchema init codec error:%v", err)
+	}
+	return ps
+}
+
+// NewProtoSchemaWithValidation creates a new ProtoSchema and error to indicate codec failure
+func NewProtoSchemaWithValidation(protoAvroSchemaDef string, properties map[string]string) (*ProtoSchema, error) {
 	ps := new(ProtoSchema)
 	avroCodec, err := initAvroCodec(protoAvroSchemaDef)
 	if err != nil {
-		log.Fatalf("init codec error:%v", err)
+		return nil, err
 	}
 	schemaDef := NewSchemaDefinition(avroCodec)
 	ps.AvroCodec.Codec = schemaDef.Codec
@@ -138,7 +160,7 @@ func NewProtoSchema(protoAvroSchemaDef string, properties map[string]string) *Pr
 	ps.SchemaInfo.Type = PROTOBUF
 	ps.SchemaInfo.Properties = properties
 	ps.SchemaInfo.Name = "Proto"
-	return ps
+	return ps, nil
 }
 
 func (ps *ProtoSchema) Encode(data interface{}) ([]byte, error) {
@@ -162,11 +184,22 @@ type AvroSchema struct {
 	SchemaInfo
 }
 
+// NewAvroSchema creates a new AvroSchema
+// Note: the function will panic if creation of codec fails
 func NewAvroSchema(avroSchemaDef string, properties map[string]string) *AvroSchema {
+	ps, err := NewAvroSchemaWithValidation(avroSchemaDef, properties)
+	if err != nil {
+		log.Fatalf("AvroSchema init codec error:%v", err)
+	}
+	return ps
+}
+
+// NewAvroSchemaWithValidation creates a new AvroSchema and error to indicate codec failure
+func NewAvroSchemaWithValidation(avroSchemaDef string, properties map[string]string) (*AvroSchema, error) {
 	as := new(AvroSchema)
 	avroCodec, err := initAvroCodec(avroSchemaDef)
 	if err != nil {
-		log.Fatalf("init codec error:%v", err)
+		return nil, err
 	}
 	schemaDef := NewSchemaDefinition(avroCodec)
 	as.AvroCodec.Codec = schemaDef.Codec
@@ -174,7 +207,7 @@ func NewAvroSchema(avroSchemaDef string, properties map[string]string) *AvroSche
 	as.SchemaInfo.Type = AVRO
 	as.SchemaInfo.Name = "Avro"
 	as.SchemaInfo.Properties = properties
-	return as
+	return as, nil
 }
 
 func (as *AvroSchema) Encode(data interface{}) ([]byte, error) {
@@ -235,14 +268,10 @@ func (ss *StringSchema) Encode(v interface{}) ([]byte, error) {
 	return []byte(v.(string)), nil
 }
 
+// Decode convert from byte slice to string without allocating a new string
 func (ss *StringSchema) Decode(data []byte, v interface{}) error {
-	bh := (*reflect.SliceHeader)(unsafe.Pointer(&data))
-	sh := reflect.StringHeader{
-		Data: bh.Data,
-		Len:  bh.Len,
-	}
-	shPtr := (*string)(unsafe.Pointer(&sh))
-	reflect.ValueOf(v).Elem().Set(reflect.ValueOf(shPtr))
+	strPtr := (*string)(unsafe.Pointer(&data))
+	reflect.ValueOf(v).Elem().Set(reflect.ValueOf(strPtr))
 	return nil
 }
 


Test patch:
diff --git a/pulsar/producer_test.go b/pulsar/producer_test.go
index 541c1fe..6b2b5d9 100644
--- a/pulsar/producer_test.go
+++ b/pulsar/producer_test.go
@@ -877,20 +877,44 @@ func TestMaxBatchSize(t *testing.T) {
 	assert.NotNil(t, producer)
 	defer producer.Close()
 
-	for bias := -1; bias <= 1; bias++ {
+	for bias := -1; bias <= 3; bias++ {
 		payload := make([]byte, batchMaxMessageSize+bias)
 		ID, err := producer.Send(context.Background(), &ProducerMessage{
 			Payload: payload,
 		})
-		if bias <= 0 {
-			assert.NoError(t, err)
-			assert.NotNil(t, ID)
-		} else {
-			assert.Equal(t, errFailAddToBatch, err)
-		}
+		// regardless max batch size, if the batch size limit is reached, batching is triggered to send messages
+		assert.NoError(t, err)
+		assert.NotNil(t, ID)
 	}
 }
 
+func TestBatchingDisabled(t *testing.T) {
+	defaultMaxMessageSize := 128 * 1024
+	client, err := NewClient(ClientOptions{
+		URL: serviceURL,
+	})
+	assert.NoError(t, err)
+	defer client.Close()
+
+	// when batching is disabled, the batching size has no effect
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic:           newTopicName(),
+		DisableBatching: true,
+		BatchingMaxSize: uint(defaultMaxBatchSize),
+	})
+	assert.NoError(t, err)
+	assert.NotNil(t, producer)
+	defer producer.Close()
+
+	payload := make([]byte, defaultMaxMessageSize+100)
+	ID, err := producer.Send(context.Background(), &ProducerMessage{
+		Payload: payload,
+	})
+	// regardless max batch size, if the batch size limit is reached, batching is triggered to send messages
+	assert.NoError(t, err)
+	assert.NotNil(t, ID)
+}
+
 func TestMaxMessageSize(t *testing.T) {
 	serverMaxMessageSize := 1024 * 1024
 
diff --git a/pulsar/schema_test.go b/pulsar/schema_test.go
index ce19bac..1aa35f0 100644
--- a/pulsar/schema_test.go
+++ b/pulsar/schema_test.go
@@ -82,7 +82,9 @@ func TestJsonSchema(t *testing.T) {
 	//create consumer
 	var s testJSON
 
-	consumerJS := NewJSONSchema(exampleSchemaDef, nil)
+	consumerJS, err := NewJSONSchemaWithValidation(exampleSchemaDef, nil)
+	assert.Nil(t, err)
+	consumerJS = NewJSONSchema(exampleSchemaDef, nil) // test this legacy function
 	consumer, err := client.Subscribe(ConsumerOptions{
 		Topic:                       "jsonTopic",
 		SubscriptionName:            "sub-1",
@@ -105,7 +107,9 @@ func TestProtoSchema(t *testing.T) {
 	defer client.Close()
 
 	// create producer
-	psProducer := NewProtoSchema(protoSchemaDef, nil)
+	psProducer, err := NewProtoSchemaWithValidation(protoSchemaDef, nil)
+	assert.Nil(t, err)
+	psProducer = NewProtoSchema(protoSchemaDef, nil)
 	producer, err := client.CreateProducer(ProducerOptions{
 		Topic:  "proto",
 		Schema: psProducer,
@@ -146,7 +150,9 @@ func TestAvroSchema(t *testing.T) {
 	defer client.Close()
 
 	// create producer
-	asProducer := NewAvroSchema(exampleSchemaDef, nil)
+	asProducer, err := NewAvroSchemaWithValidation(exampleSchemaDef, nil)
+	assert.Nil(t, err)
+	asProducer = NewAvroSchema(exampleSchemaDef, nil)
 	producer, err := client.CreateProducer(ProducerOptions{
 		Topic:  "avro-topic",
 		Schema: asProducer,