Add receiverQueueSize to pulsar (#3589)

Signed-off-by: yaron2 <[email protected]>

pubsub/pulsar/metadata.go

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

pubsub/pulsar/metadata.yaml

@@ -176,4 +176,11 @@ metadata:
     description: |
       Sets the maximum number of concurrent messages sent to the application. Default is 100.
     default: '"100"'
-    example: '"100"'
+    example: '"100"'
+  - name: receiverQueueSize
+    type: number
+    description: |
+      Sets the size of the consumer receive queue.
+      Controls how many messages can be accumulated by the consumer before it is explicitly called to read messages by Dapr.
+    default: '"1000"'
+    example: '"1000"'

pubsub/pulsar/pulsar.go

@@ -80,6 +80,8 @@ const (
 	defaultRedeliveryDelay = 30 * time.Second
 	// defaultConcurrency controls the number of concurrent messages sent to the app.
 	defaultConcurrency = 100
+	// defaultReceiverQueueSize controls the number of messages the pulsar sdk pulls before dapr explicitly consumes the messages.
+	defaultReceiverQueueSize = 1000
 
 	subscribeTypeKey = "subscribeType"
 
@@ -125,6 +127,7 @@ func parsePulsarMetadata(meta pubsub.Metadata) (*pulsarMetadata, error) {
 		BatchingMaxSize:         defaultMaxBatchSize,
 		RedeliveryDelay:         defaultRedeliveryDelay,
 		MaxConcurrentHandlers:   defaultConcurrency,
+		ReceiverQueueSize:       defaultReceiverQueueSize,
 	}
 
 	if err := kitmd.DecodeMetadata(meta.Properties, &m); err != nil {
@@ -403,6 +406,7 @@ func (p *Pulsar) Subscribe(ctx context.Context, req pubsub.SubscribeRequest, han
 		Type:                getSubscribeType(req.Metadata),
 		MessageChannel:      channel,
 		NackRedeliveryDelay: p.metadata.RedeliveryDelay,
+		ReceiverQueueSize:   p.metadata.ReceiverQueueSize,
 	}
 
 	if p.useConsumerEncryption() {