Enable configurable concurrency for pulsar

pubsub/pulsar/metadata.go

@@ -35,6 +35,7 @@ type pulsarMetadata struct {
 	PublicKey               string                    `mapstructure:"publicKey"`
 	PrivateKey              string                    `mapstructure:"privateKey"`
 	Keys                    string                    `mapstructure:"keys"`
+	MaxConcurrentHandlers   uint                      `mapstructure:"maxConcurrentHandlers"`
 
 	Token                            string `mapstructure:"token"`
 	oauth2.ClientCredentialsMetadata `mapstructure:",squash"`

pubsub/pulsar/metadata.yaml

@@ -170,4 +170,10 @@ metadata:
           {"name": "ID","type": "int"},
           {"name": "Name","type": "string"}
         ]
-      }
+      }
+  - name: maxConcurrentHandlers
+    type: number
+    description: |
+      Sets the maximum number of concurrent messages sent to the application. Default is 100.
+    default: '"100"'
+    example: '"100"'

pubsub/pulsar/pulsar.go

@@ -78,6 +78,8 @@ const (
 	defaultMaxBatchSize = 128 * 1024
 	// defaultRedeliveryDelay init default for redelivery delay.
 	defaultRedeliveryDelay = 30 * time.Second
+	// defaultConcurrency controls the number of concurrent messages sent to the app.
+	defaultConcurrency = 100
 
 	subscribeTypeKey = "subscribeType"
 
@@ -122,6 +124,7 @@ func parsePulsarMetadata(meta pubsub.Metadata) (*pulsarMetadata, error) {
 		BatchingMaxMessages:     defaultMaxMessages,
 		BatchingMaxSize:         defaultMaxBatchSize,
 		RedeliveryDelay:         defaultRedeliveryDelay,
+		MaxConcurrentHandlers:   defaultConcurrency,
 	}
 
 	if err := kitmd.DecodeMetadata(meta.Properties, &m); err != nil {
@@ -392,7 +395,7 @@ func (p *Pulsar) Subscribe(ctx context.Context, req pubsub.SubscribeRequest, han
 		return errors.New("component is closed")
 	}
 
-	channel := make(chan pulsar.ConsumerMessage, 100)
+	channel := make(chan pulsar.ConsumerMessage, p.metadata.MaxConcurrentHandlers)
 
 	topic := p.formatTopic(req.Topic)
 

pubsub/pulsar/pulsar_test.go

@@ -33,6 +33,7 @@ func TestParsePulsarMetadata(t *testing.T) {
 		"batchingMaxPublishDelay": "5s",
 		"batchingMaxSize":         "100",
 		"batchingMaxMessages":     "200",
+		"maxConcurrentHandlers":   "333",
 	}
 	meta, err := parsePulsarMetadata(m)
 
@@ -45,6 +46,7 @@ func TestParsePulsarMetadata(t *testing.T) {
 	assert.Equal(t, 5*time.Second, meta.BatchingMaxPublishDelay)
 	assert.Equal(t, uint(100), meta.BatchingMaxSize)
 	assert.Equal(t, uint(200), meta.BatchingMaxMessages)
+	assert.Equal(t, uint(333), meta.MaxConcurrentHandlers)
 	assert.Empty(t, meta.internalTopicSchemas)
 }