Added redeliveryDelay to pubsub.pulsar

Signed-off-by: ItalyPaleAle <43508+ItalyPaleAle@users.noreply.github.com>

pubsub/pulsar/metadata.go

@@ -27,4 +27,5 @@ type pulsarMetadata struct {
 	Namespace               string        `json:"namespace"`
 	Persistent              bool          `json:"persistent"`
 	Token                   string        `json:"token"`
+	RedeliveryDelay         time.Duration `json:"redeliveryDelay"`
 }

pubsub/pulsar/pulsar.go

@@ -41,6 +41,7 @@ const (
 	tenant                  = "tenant"
 	namespace               = "namespace"
 	persistent              = "persistent"
+	redeliveryDelay         = "redeliveryDelay"
 
 	defaultTenant     = "public"
 	defaultNamespace  = "default"
@@ -59,6 +60,8 @@ const (
 	defaultMaxMessages = 1000
 	// defaultMaxBatchSize init default for maximum number of bytes per batch.
 	defaultMaxBatchSize = 128 * 1024
+	// defaultRedeliveryDelay init default for redelivery delay.
+	defaultRedeliveryDelay = 30 * time.Second
 )
 
 type Pulsar struct {
@@ -124,6 +127,14 @@ func parsePulsarMetadata(meta pubsub.Metadata) (*pulsarMetadata, error) {
 		}
 		m.BatchingMaxSize = uint(batchingMaxSize)
 	}
+	m.RedeliveryDelay = defaultRedeliveryDelay
+	if val, ok := meta.Properties[redeliveryDelay]; ok {
+		redeliveryDelay, err := formatDuration(val)
+		if err != nil {
+			return nil, errors.New("pulsar error: invalid value for redeliveryDelay")
+		}
+		m.RedeliveryDelay = redeliveryDelay
+	}
 	if val, ok := meta.Properties[persistent]; ok && val != "" {
 		per, err := strconv.ParseBool(val)
 		if err != nil {
@@ -255,10 +266,11 @@ func (p *Pulsar) Subscribe(ctx context.Context, req pubsub.SubscribeRequest, han
 
 	topic := p.formatTopic(req.Topic)
 	options := pulsar.ConsumerOptions{
-		Topic:            topic,
-		SubscriptionName: p.metadata.ConsumerID,
-		Type:             pulsar.Shared,
-		MessageChannel:   channel,
+		Topic:               topic,
+		SubscriptionName:    p.metadata.ConsumerID,
+		Type:                pulsar.Shared,
+		MessageChannel:      channel,
+		NackRedeliveryDelay: p.metadata.RedeliveryDelay,
 	}
 
 	consumer, err := p.client.Subscribe(options)

tests/config/pubsub/pulsar/pubsub.yml

@@ -9,4 +9,6 @@ spec:
   - name: host
     value: "localhost:6650"
   - name: consumerID
-    value: myConsumerID
+    value: myConsumerID
+  - name: redeliveryDelay
+    value: 1s