Support nack backoff policy for SDK

pulsar/consumer.go

@@ -158,6 +158,17 @@ type ConsumerOptions struct {
 
 	// Decryption decryption related fields to decrypt the encrypted message
 	Decryption *MessageDecryptionInfo
+
+	// If enabled, the default implementation of NackBackoffPolicy will be used to calculate the delay time of
+	// nack backoff, Default: false.
+	EnableDefaultNackBackoffPolicy bool
+
+	// NackBackoffPolicy is a redelivery backoff mechanism which we can achieve redelivery with different
+	// delays according to the number of times the message is retried.
+	//
+	// > Notice: the NackBackoffPolicy will not work with `consumer.NackID(MessageID)`
+	// > because we are not able to get the redeliveryCount from the message ID.
+	NackBackoffPolicy NackBackoffPolicy
 }
 
 // Consumer is an interface that abstracts behavior of Pulsar's consumer

pulsar/consumer_impl.go

@@ -35,6 +35,7 @@ const defaultNackRedeliveryDelay = 1 * time.Minute
 type acker interface {
 	AckID(id trackingMessageID)
 	NackID(id trackingMessageID)
+	NackMsg(msg Message)
 }
 
 type consumer struct {
@@ -87,6 +88,10 @@ func newConsumer(client *client, options ConsumerOptions) (Consumer, error) {
 		}
 	}
 
+	if options.NackBackoffPolicy == nil && options.EnableDefaultNackBackoffPolicy {
+		options.NackBackoffPolicy = new(defaultNackBackoffPolicy)
+	}
+
 	// did the user pass in a message channel?
 	messageCh := options.MessageChannel
 	if options.MessageChannel == nil {
@@ -326,6 +331,7 @@ func (c *consumer) internalTopicSubscribeToPartitions() error {
 				partitionIdx:               idx,
 				receiverQueueSize:          receiverQueueSize,
 				nackRedeliveryDelay:        nackRedeliveryDelay,
+				nackBackoffPolicy:          c.options.NackBackoffPolicy,
 				metadata:                   metadata,
 				replicateSubscriptionState: c.options.ReplicateSubscriptionState,
 				startMessageID:             trackingMessageID{},
@@ -489,6 +495,20 @@ func (c *consumer) ReconsumeLater(msg Message, delay time.Duration) {
 }
 
 func (c *consumer) Nack(msg Message) {
+	if c.options.EnableDefaultNackBackoffPolicy || c.options.NackBackoffPolicy != nil {
+		mid, ok := c.messageID(msg.ID())
+		if !ok {
+			return
+		}
+
+		if mid.consumer != nil {
+			mid.Nack()
+			return
+		}
+		c.consumers[mid.partitionIdx].NackMsg(msg)
+		return
+	}
+
 	c.NackID(msg.ID())
 }
 

pulsar/consumer_multitopic.go

@@ -164,6 +164,22 @@ func (c *multiTopicConsumer) ReconsumeLater(msg Message, delay time.Duration) {
 }
 
 func (c *multiTopicConsumer) Nack(msg Message) {
+	if c.options.EnableDefaultNackBackoffPolicy || c.options.NackBackoffPolicy != nil {
+		msgID := msg.ID()
+		mid, ok := toTrackingMessageID(msgID)
+		if !ok {
+			c.log.Warnf("invalid message id type %T", msgID)
+			return
+		}
+
+		if mid.consumer == nil {
+			c.log.Warnf("unable to nack messageID=%+v can not determine topic", msgID)
+			return
+		}
+		mid.NackByMsg(msg)
+		return
+	}
+
 	c.NackID(msg.ID())
 }
 

pulsar/consumer_partition.go

@@ -89,6 +89,7 @@ type partitionConsumerOpts struct {
 	partitionIdx               int
 	receiverQueueSize          int
 	nackRedeliveryDelay        time.Duration
+	nackBackoffPolicy          NackBackoffPolicy
 	metadata                   map[string]string
 	replicateSubscriptionState bool
 	startMessageID             trackingMessageID
@@ -201,7 +202,7 @@ func newPartitionConsumer(parent Consumer, client *client, options *partitionCon
 
 	pc.decryptor = decryptor
 
-	pc.nackTracker = newNegativeAcksTracker(pc, options.nackRedeliveryDelay, pc.log)
+	pc.nackTracker = newNegativeAcksTracker(pc, options.nackRedeliveryDelay, options.nackBackoffPolicy, pc.log)
 
 	err := pc.grabConn()
 	if err != nil {
@@ -331,6 +332,11 @@ func (pc *partitionConsumer) NackID(msgID trackingMessageID) {
 	pc.metrics.NacksCounter.Inc()
 }
 
+func (pc *partitionConsumer) NackMsg(msg Message) {
+	pc.nackTracker.AddMessage(msg)
+	pc.metrics.NacksCounter.Inc()
+}
+
 func (pc *partitionConsumer) Redeliver(msgIds []messageID) {
 	pc.eventsCh <- &redeliveryRequest{msgIds}
 

pulsar/consumer_regex.go

@@ -183,6 +183,22 @@ func (c *regexConsumer) AckID(msgID MessageID) {
 }
 
 func (c *regexConsumer) Nack(msg Message) {
+	if c.options.EnableDefaultNackBackoffPolicy || c.options.NackBackoffPolicy != nil {
+		msgID := msg.ID()
+		mid, ok := toTrackingMessageID(msgID)
+		if !ok {
+			c.log.Warnf("invalid message id type %T", msgID)
+			return
+		}
+
+		if mid.consumer == nil {
+			c.log.Warnf("unable to nack messageID=%+v can not determine topic", msgID)
+			return
+		}
+		mid.NackByMsg(msg)
+		return
+	}
+
 	c.NackID(msg.ID())
 }
 

pulsar/impl_message.go

@@ -79,6 +79,13 @@ func (id trackingMessageID) Nack() {
 	id.consumer.NackID(id)
 }
 
+func (id trackingMessageID) NackByMsg(msg Message) {
+	if id.consumer == nil {
+		return
+	}
+	id.consumer.NackMsg(msg)
+}
+
 func (id trackingMessageID) ack() bool {
 	if id.tracker != nil && id.batchIdx > -1 {
 		return id.tracker.ack(int(id.batchIdx))

pulsar/negative_acks_tracker.go

@@ -35,22 +35,41 @@ type negativeAcksTracker struct {
 	doneOnce     sync.Once
 	negativeAcks map[messageID]time.Time
 	rc           redeliveryConsumer
-	tick         *time.Ticker
+	nackBackoff  NackBackoffPolicy
+	trackFlag    bool
 	delay        time.Duration
 	log          log.Logger
 }
 
-func newNegativeAcksTracker(rc redeliveryConsumer, delay time.Duration, logger log.Logger) *negativeAcksTracker {
-	t := &negativeAcksTracker{
-		doneCh:       make(chan interface{}),
-		negativeAcks: make(map[messageID]time.Time),
-		rc:           rc,
-		tick:         time.NewTicker(delay / 3),
-		delay:        delay,
-		log:          logger,
-	}
+func newNegativeAcksTracker(rc redeliveryConsumer, delay time.Duration,
+	nackBackoffPolicy NackBackoffPolicy, logger log.Logger) *negativeAcksTracker {
+
+	t := new(negativeAcksTracker)
+
+	// When using NackBackoffPolicy, the delay time needs to be calculated based on the RedeliveryCount field in
+	// the CommandMessage, so for the original default Nack() logic, we still keep the negativeAcksTracker created
+	// when we open a gorutine to execute the logic of `t.track()`. But for the NackBackoffPolicy method, we need
+	// to execute the logic of `t.track()` when AddMessage().
+	if nackBackoffPolicy != nil {
+		t = &negativeAcksTracker{
+			doneCh:       make(chan interface{}),
+			negativeAcks: make(map[messageID]time.Time),
+			nackBackoff:  nackBackoffPolicy,
+			rc:           rc,
+			log:          logger,
+		}
+	} else {
+		t = &negativeAcksTracker{
+			doneCh:       make(chan interface{}),
+			negativeAcks: make(map[messageID]time.Time),
+			rc:           rc,
+			nackBackoff:  nil,
+			delay:        delay,
+			log:          logger,
+		}
 
-	go t.track()
+		go t.track(time.NewTicker(t.delay / 3))
+	}
 	return t
 }
 
@@ -76,14 +95,48 @@ func (t *negativeAcksTracker) Add(msgID messageID) {
 	t.negativeAcks[batchMsgID] = targetTime
 }
 
-func (t *negativeAcksTracker) track() {
+func (t *negativeAcksTracker) AddMessage(msg Message) {
+	nackBackoffDelay := t.nackBackoff.Next(msg.RedeliveryCount())
+	t.delay = time.Duration(nackBackoffDelay)
+
+	// Use trackFlag to avoid opening a new gorutine to execute `t.track()` every AddMessage.
+	// In fact, we only need to execute it once.
+	if !t.trackFlag {
+		go t.track(time.NewTicker(t.delay / 3))
+		t.trackFlag = true
+	}
+
+	msgID := msg.ID()
+
+	// Always clear up the batch index since we want to track the nack
+	// for the entire batch
+	batchMsgID := messageID{
+		ledgerID: msgID.LedgerID(),
+		entryID:  msgID.EntryID(),
+		batchIdx: 0,
+	}
+
+	t.Lock()
+	defer t.Unlock()
+
+	_, present := t.negativeAcks[batchMsgID]
+	if present {
+		// The batch is already being tracked
+		return
+	}
+
+	targetTime := time.Now().Add(time.Duration(nackBackoffDelay))
+	t.negativeAcks[batchMsgID] = targetTime
+}
+
+func (t *negativeAcksTracker) track(ticker *time.Ticker) {
 	for {
 		select {
 		case <-t.doneCh:
 			t.log.Debug("Closing nack tracker")
 			return
 
-		case <-t.tick.C:
+		case <-ticker.C:
 			{
 				now := time.Now()
 				msgIds := make([]messageID, 0)
@@ -105,15 +158,13 @@ func (t *negativeAcksTracker) track() {
 					t.rc.Redeliver(msgIds)
 				}
 			}
-
 		}
 	}
 }
 
 func (t *negativeAcksTracker) Close() {
 	// allow Close() to be invoked multiple times by consumer_partition to avoid panic
 	t.doneOnce.Do(func() {
-		t.tick.Stop()
 		t.doneCh <- nil
 	})
 }