Add metadata property to configure Batching in Pulsar

pubsub/pulsar/metadata.go

@@ -13,13 +13,18 @@ limitations under the License.
 
 package pulsar
 
+import "time"
+
 type pulsarMetadata struct {
-	Host            string `json:"host"`
-	ConsumerID      string `json:"consumerID"`
-	EnableTLS       bool   `json:"enableTLS"`
-	DisableBatching bool   `json:"disableBatching"`
-	Tenant          string `json:"tenant"`
-	Namespace       string `json:"namespace"`
-	Persistent      bool   `json:"persistent"`
-	Token           string `json:"token"`
+	Host                    string        `json:"host"`
+	ConsumerID              string        `json:"consumerID"`
+	EnableTLS               bool          `json:"enableTLS"`
+	DisableBatching         bool          `json:"disableBatching"`
+	BatchingMaxPublishDelay time.Duration `json:"batchingMaxPublishDelay"`
+	BatchingMaxSize         uint          `json:"batchingMaxSize"`
+	BatchingMaxMessages     uint          `json:"batchingMaxMessages"`
+	Tenant                  string        `json:"tenant"`
+	Namespace               string        `json:"namespace"`
+	Persistent              bool          `json:"persistent"`
+	Token                   string        `json:"token"`
 }

pubsub/pulsar/pulsar.go

@@ -30,15 +30,18 @@ import (
 )
 
 const (
-	host            = "host"
-	consumerID      = "consumerID"
-	enableTLS       = "enableTLS"
-	deliverAt       = "deliverAt"
-	deliverAfter    = "deliverAfter"
-	disableBatching = "disableBatching"
-	tenant          = "tenant"
-	namespace       = "namespace"
-	persistent      = "persistent"
+	host                    = "host"
+	consumerID              = "consumerID"
+	enableTLS               = "enableTLS"
+	deliverAt               = "deliverAt"
+	deliverAfter            = "deliverAfter"
+	disableBatching         = "disableBatching"
+	batchingMaxPublishDelay = "batchingMaxPublishDelay"
+	batchingMaxSize         = "batchingMaxSize"
+	batchingMaxMessages     = "batchingMaxMessages"
+	tenant                  = "tenant"
+	namespace               = "namespace"
+	persistent              = "persistent"
 
 	defaultTenant     = "public"
 	defaultNamespace  = "default"
@@ -50,6 +53,13 @@ const (
 	topicFormat      = "%s://%s/%s/%s"
 	persistentStr    = "persistent"
 	nonPersistentStr = "non-persistent"
+
+	// defaultBatchingMaxPublishDelay init default for maximum delay to batch messages.
+	defaultBatchingMaxPublishDelay = 10 * time.Millisecond
+	// defaultMaxMessages init default num of entries in per batch.
+	defaultMaxMessages = 1000
+	// defaultMaxBatchSize init default for maximum number of bytes per batch.
+	defaultMaxBatchSize = 128 * 1024
 )
 
 type Pulsar struct {
@@ -92,7 +102,30 @@ func parsePulsarMetadata(meta pubsub.Metadata) (*pulsarMetadata, error) {
 		}
 		m.DisableBatching = disableBatching
 	}
-
+	m.BatchingMaxPublishDelay = defaultBatchingMaxPublishDelay
+	if val, ok := meta.Properties[batchingMaxPublishDelay]; ok {
+		batchingMaxPublishDelay, err := formatDuration(val)
+		if err != nil {
+			return nil, errors.New("pulsar error: invalid value for batchingMaxPublishDelay")
+		}
+		m.BatchingMaxPublishDelay = batchingMaxPublishDelay
+	}
+	m.BatchingMaxMessages = defaultMaxMessages
+	if val, ok := meta.Properties[batchingMaxMessages]; ok {
+		batchingMaxMessages, err := strconv.ParseUint(val, 10, 64)
+		if err != nil {
+			return nil, errors.New("pulsar error: invalid value for batchingMaxMessages")
+		}
+		m.BatchingMaxMessages = uint(batchingMaxMessages)
+	}
+	m.BatchingMaxSize = defaultMaxBatchSize
+	if val, ok := meta.Properties[batchingMaxSize]; ok {
+		batchingMaxSize, err := strconv.ParseUint(val, 10, 64)
+		if err != nil {
+			return nil, errors.New("pulsar error: invalid value for batchingMaxSize")
+		}
+		m.BatchingMaxSize = uint(batchingMaxSize)
+	}
 	if val, ok := meta.Properties[persistent]; ok && val != "" {
 		per, err := strconv.ParseBool(val)
 		if err != nil {
@@ -179,8 +212,11 @@ func (p *Pulsar) Publish(req *pubsub.PublishRequest) error {
 	if cache == nil {
 		p.logger.Debugf("creating producer for topic %s, full topic name in pulsar is %s", req.Topic, topic)
 		producer, err = p.client.CreateProducer(pulsar.ProducerOptions{
-			Topic:           topic,
-			DisableBatching: p.metadata.DisableBatching,
+			Topic:                   topic,
+			DisableBatching:         p.metadata.DisableBatching,
+			BatchingMaxPublishDelay: p.metadata.BatchingMaxPublishDelay,
+			BatchingMaxMessages:     p.metadata.BatchingMaxMessages,
+			BatchingMaxSize:         p.metadata.BatchingMaxSize,
 		})
 		if err != nil {
 			return err
@@ -317,3 +353,14 @@ func (p *Pulsar) formatTopic(topic string) string {
 	}
 	return fmt.Sprintf(topicFormat, persist, p.metadata.Tenant, p.metadata.Namespace, topic)
 }
+
+func formatDuration(durationString string) (time.Duration, error) {
+	if val, err := strconv.Atoi(durationString); err == nil {
+		return time.Duration(val) * time.Millisecond, nil
+	}
+
+	// Convert it by parsing
+	d, err := time.ParseDuration(durationString)
+
+	return d, err
+}

pubsub/pulsar/pulsar_test.go

@@ -25,9 +25,12 @@ import (
 func TestParsePulsarMetadata(t *testing.T) {
 	m := pubsub.Metadata{}
 	m.Properties = map[string]string{
-		"host":            "a",
-		"enableTLS":       "false",
-		"disableBatching": "true",
+		"host":                    "a",
+		"enableTLS":               "false",
+		"disableBatching":         "true",
+		"batchingMaxPublishDelay": "5s",
+		"batchingMaxSize":         "100",
+		"batchingMaxMessages":     "200",
 	}
 	meta, err := parsePulsarMetadata(m)
 
@@ -37,6 +40,9 @@ func TestParsePulsarMetadata(t *testing.T) {
 	assert.Equal(t, true, meta.DisableBatching)
 	assert.Equal(t, defaultTenant, meta.Tenant)
 	assert.Equal(t, defaultNamespace, meta.Namespace)
+	assert.Equal(t, 5*time.Second, meta.BatchingMaxPublishDelay)
+	assert.Equal(t, uint(100), meta.BatchingMaxSize)
+	assert.Equal(t, uint(200), meta.BatchingMaxMessages)
 }
 
 func TestParsePublishMetadata(t *testing.T) {