Repository: apache/pulsar-client-go
Pull request #660 (2021-11-03 10:06:11)

Support nack backoff policy for SDK
Signed-off-by: xiaolongran <xiaolongran@tencent.com>


Master Issue: #658

### Motivation

Support nack backoff policy

### Modifications

- Add `EnableDefaultNackBackoffPolicy` and `NackBackoffPolicy` options in ConsumerOptions.
- Modify the original implementation of the `Nack(Message)` interface
- Add test case
- Add defaultNackBackoffPolicy 




Patch:
diff --git a/pulsar/consumer.go b/pulsar/consumer.go
index c9fbc0d..5127955 100644
--- a/pulsar/consumer.go
+++ b/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
diff --git a/pulsar/consumer_impl.go b/pulsar/consumer_impl.go
index 6aef497..1bf75b5 100644
--- a/pulsar/consumer_impl.go
+++ b/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())
 }
 
diff --git a/pulsar/consumer_multitopic.go b/pulsar/consumer_multitopic.go
index f689fae..c1cb3d8 100644
--- a/pulsar/consumer_multitopic.go
+++ b/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())
 }
 
diff --git a/pulsar/consumer_partition.go b/pulsar/consumer_partition.go
index 30639bd..c8e5d9f 100644
--- a/pulsar/consumer_partition.go
+++ b/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}
 
diff --git a/pulsar/consumer_regex.go b/pulsar/consumer_regex.go
index 2f46c48..ed2ae1a 100644
--- a/pulsar/consumer_regex.go
+++ b/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())
 }
 
diff --git a/pulsar/impl_message.go b/pulsar/impl_message.go
index 19fa6d8..a9809aa 100644
--- a/pulsar/impl_message.go
+++ b/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))
diff --git a/pulsar/negative_acks_tracker.go b/pulsar/negative_acks_tracker.go
index e10ab49..3485e1b 100644
--- a/pulsar/negative_acks_tracker.go
+++ b/pulsar/negative_acks_tracker.go
@@ -35,21 +35,32 @@ type negativeAcksTracker struct {
 	doneOnce     sync.Once
 	negativeAcks map[messageID]time.Time
 	rc           redeliveryConsumer
+	nackBackoff  NackBackoffPolicy
 	tick         *time.Ticker
 	delay        time.Duration
 	log          log.Logger
 }
 
-func newNegativeAcksTracker(rc redeliveryConsumer, delay time.Duration, logger log.Logger) *negativeAcksTracker {
+func newNegativeAcksTracker(rc redeliveryConsumer, delay time.Duration,
+	nackBackoffPolicy NackBackoffPolicy, 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,
+		nackBackoff:  nackBackoffPolicy,
 		log:          logger,
 	}
 
+	if nackBackoffPolicy != nil {
+		firstDelayForNackBackoff := nackBackoffPolicy.Next(1)
+		t.delay = time.Duration(firstDelayForNackBackoff)
+	} else {
+		t.delay = delay
+	}
+
+	t.tick = time.NewTicker(t.delay / 3)
+
 	go t.track()
 	return t
 }
@@ -76,6 +87,32 @@ func (t *negativeAcksTracker) Add(msgID messageID) {
 	t.negativeAcks[batchMsgID] = targetTime
 }
 
+func (t *negativeAcksTracker) AddMessage(msg Message) {
+	nackBackoffDelay := t.nackBackoff.Next(msg.RedeliveryCount())
+
+	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() {
 	for {
 		select {
@@ -105,7 +142,6 @@ func (t *negativeAcksTracker) track() {
 					t.rc.Redeliver(msgIds)
 				}
 			}
-
 		}
 	}
 }
diff --git a/pulsar/negative_backoff_policy.go b/pulsar/negative_backoff_policy.go
new file mode 100644
index 0000000..cf080ad
--- /dev/null
+++ b/pulsar/negative_backoff_policy.go
@@ -0,0 +1,46 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package pulsar
+
+import "math"
+
+// NackBackoffPolicy is a interface for custom message negativeAcked policy, users can specify a NackBackoffPolicy
+// for a consumer.
+//
+// > Notice: the consumer crashes will trigger the redelivery of the unacked message, this case will not respect the
+// > NackBackoffPolicy, which means the message might get redelivered earlier than the delay time
+// > from the backoff.
+type NackBackoffPolicy interface {
+	// The redeliveryCount indicates the number of times the message was redelivered.
+	// We can get the redeliveryCount from the CommandMessage.
+	Next(redeliveryCount uint32) int64
+}
+
+// defaultNackBackoffPolicy is default impl for NackBackoffPolicy.
+type defaultNackBackoffPolicy struct{}
+
+func (nbp *defaultNackBackoffPolicy) Next(redeliveryCount uint32) int64 {
+	minNackTimeMs := int64(1000 * 30) // 30sec
+	maxNackTimeMs := 1000 * 60 * 10   // 10min
+
+	if redeliveryCount < 0 {
+		return minNackTimeMs
+	}
+
+	return int64(math.Min(math.Abs(float64(minNackTimeMs<<redeliveryCount)), float64(maxNackTimeMs)))
+}
diff --git a/pulsar/producer_partition.go b/pulsar/producer_partition.go
index b2b9273..d67c0c0 100644
--- a/pulsar/producer_partition.go
+++ b/pulsar/producer_partition.go
@@ -784,7 +784,7 @@ func (p *partitionProducer) ReceivedSendReceipt(response *pb.CommandSendReceipt)
 		return
 	}
 
-	if pi.sequenceID != response.GetSequenceId() {
+	if pi.sequenceID < response.GetSequenceId() {
 		// if we receive a receipt that is not the one expected, the state of the broker and the producer differs.
 		// At that point, it is better to close the connection to the broker to reconnect to a broker hopping it solves
 		// the state discrepancy.
@@ -792,49 +792,49 @@ func (p *partitionProducer) ReceivedSendReceipt(response *pb.CommandSendReceipt)
 			response.GetSequenceId(), pi.sequenceID)
 		p.cnx.Close()
 		return
-	}
-
-	// The ack was indeed for the expected item in the queue, we can remove it and trigger the callback
-	p.pendingQueue.Poll()
-
-	now := time.Now().UnixNano()
-
-	// lock the pending item while sending the requests
-	pi.Lock()
-	defer pi.Unlock()
-	p.metrics.PublishRPCLatency.Observe(float64(now-pi.sentAt.UnixNano()) / 1.0e9)
-	for idx, i := range pi.sendRequests {
-		sr := i.(*sendRequest)
-		if sr.msg != nil {
-			atomic.StoreInt64(&p.lastSequenceID, int64(pi.sequenceID))
-			p.publishSemaphore.Release()
+	} else if pi.sequenceID == response.GetSequenceId() {
+		// The ack was indeed for the expected item in the queue, we can remove it and trigger the callback
+		p.pendingQueue.Poll()
+
+		now := time.Now().UnixNano()
+
+		// lock the pending item while sending the requests
+		pi.Lock()
+		defer pi.Unlock()
+		p.metrics.PublishRPCLatency.Observe(float64(now-pi.sentAt.UnixNano()) / 1.0e9)
+		for idx, i := range pi.sendRequests {
+			sr := i.(*sendRequest)
+			if sr.msg != nil {
+				atomic.StoreInt64(&p.lastSequenceID, int64(pi.sequenceID))
+				p.publishSemaphore.Release()
+
+				p.metrics.PublishLatency.Observe(float64(now-sr.publishTime.UnixNano()) / 1.0e9)
+				p.metrics.MessagesPublished.Inc()
+				p.metrics.MessagesPending.Dec()
+				payloadSize := float64(len(sr.msg.Payload))
+				p.metrics.BytesPublished.Add(payloadSize)
+				p.metrics.BytesPending.Sub(payloadSize)
+			}
 
-			p.metrics.PublishLatency.Observe(float64(now-sr.publishTime.UnixNano()) / 1.0e9)
-			p.metrics.MessagesPublished.Inc()
-			p.metrics.MessagesPending.Dec()
-			payloadSize := float64(len(sr.msg.Payload))
-			p.metrics.BytesPublished.Add(payloadSize)
-			p.metrics.BytesPending.Sub(payloadSize)
-		}
+			if sr.callback != nil || len(p.options.Interceptors) > 0 {
+				msgID := newMessageID(
+					int64(response.MessageId.GetLedgerId()),
+					int64(response.MessageId.GetEntryId()),
+					int32(idx),
+					p.partitionIdx,
+				)
 
-		if sr.callback != nil || len(p.options.Interceptors) > 0 {
-			msgID := newMessageID(
-				int64(response.MessageId.GetLedgerId()),
-				int64(response.MessageId.GetEntryId()),
-				int32(idx),
-				p.partitionIdx,
-			)
+				if sr.callback != nil {
+					sr.callback(msgID, sr.msg, nil)
+				}
 
-			if sr.callback != nil {
-				sr.callback(msgID, sr.msg, nil)
+				p.options.Interceptors.OnSendAcknowledgement(p, sr.msg, msgID)
 			}
-
-			p.options.Interceptors.OnSendAcknowledgement(p, sr.msg, msgID)
 		}
-	}
 
-	// Mark this pending item as done
-	pi.Complete()
+		// Mark this pending item as done
+		pi.Complete()
+	}
 }
 
 func (p *partitionProducer) internalClose(req *closeProducer) {


Test patch:
diff --git a/pulsar/negative_acks_tracker_test.go b/pulsar/negative_acks_tracker_test.go
index e587f3f..51965ea 100644
--- a/pulsar/negative_acks_tracker_test.go
+++ b/pulsar/negative_acks_tracker_test.go
@@ -75,7 +75,7 @@ func (nmc *nackMockedConsumer) Wait() <-chan messageID {
 
 func TestNacksTracker(t *testing.T) {
 	nmc := newNackMockedConsumer()
-	nacks := newNegativeAcksTracker(nmc, testNackDelay, log.DefaultNopLogger())
+	nacks := newNegativeAcksTracker(nmc, testNackDelay, nil, log.DefaultNopLogger())
 
 	nacks.Add(messageID{
 		ledgerID: 1,
@@ -108,7 +108,7 @@ func TestNacksTracker(t *testing.T) {
 
 func TestNacksWithBatchesTracker(t *testing.T) {
 	nmc := newNackMockedConsumer()
-	nacks := newNegativeAcksTracker(nmc, testNackDelay, log.DefaultNopLogger())
+	nacks := newNegativeAcksTracker(nmc, testNackDelay, nil, log.DefaultNopLogger())
 
 	nacks.Add(messageID{
 		ledgerID: 1,
@@ -148,3 +148,155 @@ func TestNacksWithBatchesTracker(t *testing.T) {
 
 	nacks.Close()
 }
+
+func TestNackBackoffTracker(t *testing.T) {
+	nmc := newNackMockedConsumer()
+	nacks := newNegativeAcksTracker(nmc, testNackDelay, new(defaultNackBackoffPolicy), log.DefaultNopLogger())
+
+	nacks.AddMessage(new(mockMessage1))
+	nacks.AddMessage(new(mockMessage2))
+
+	msgIds := make([]messageID, 0)
+	for id := range nmc.Wait() {
+		msgIds = append(msgIds, id)
+	}
+	msgIds = sortMessageIds(msgIds)
+
+	assert.Equal(t, 2, len(msgIds))
+	assert.Equal(t, int64(1), msgIds[0].ledgerID)
+	assert.Equal(t, int64(1), msgIds[0].entryID)
+	assert.Equal(t, int64(2), msgIds[1].ledgerID)
+	assert.Equal(t, int64(2), msgIds[1].entryID)
+
+	nacks.Close()
+	// allow multiple Close without panicing
+	nacks.Close()
+}
+
+type mockMessage1 struct {
+	properties map[string]string
+}
+
+func (msg *mockMessage1) Topic() string {
+	return ""
+}
+
+func (msg *mockMessage1) Properties() map[string]string {
+	return msg.properties
+}
+
+func (msg *mockMessage1) Payload() []byte {
+	return nil
+}
+
+func (msg *mockMessage1) ID() MessageID {
+	return messageID{
+		ledgerID: 1,
+		entryID:  1,
+		batchIdx: 1,
+	}
+}
+
+func (msg *mockMessage1) PublishTime() time.Time {
+	return time.Time{}
+}
+
+func (msg *mockMessage1) EventTime() time.Time {
+	return time.Time{}
+}
+
+func (msg *mockMessage1) Key() string {
+	return ""
+}
+
+func (msg *mockMessage1) OrderingKey() string {
+	return ""
+}
+
+func (msg *mockMessage1) RedeliveryCount() uint32 {
+	return 0
+}
+
+func (msg *mockMessage1) IsReplicated() bool {
+	return false
+}
+
+func (msg *mockMessage1) GetReplicatedFrom() string {
+	return ""
+}
+
+func (msg *mockMessage1) GetSchemaValue(v interface{}) error {
+	return nil
+}
+
+func (msg *mockMessage1) ProducerName() string {
+	return ""
+}
+
+func (msg *mockMessage1) GetEncryptionContext() *EncryptionContext {
+	return &EncryptionContext{}
+}
+
+type mockMessage2 struct {
+	properties map[string]string
+}
+
+func (msg *mockMessage2) Topic() string {
+	return ""
+}
+
+func (msg *mockMessage2) Properties() map[string]string {
+	return msg.properties
+}
+
+func (msg *mockMessage2) Payload() []byte {
+	return nil
+}
+
+func (msg *mockMessage2) ID() MessageID {
+	return messageID{
+		ledgerID: 2,
+		entryID:  2,
+		batchIdx: 1,
+	}
+}
+
+func (msg *mockMessage2) PublishTime() time.Time {
+	return time.Time{}
+}
+
+func (msg *mockMessage2) EventTime() time.Time {
+	return time.Time{}
+}
+
+func (msg *mockMessage2) Key() string {
+	return ""
+}
+
+func (msg *mockMessage2) OrderingKey() string {
+	return ""
+}
+
+func (msg *mockMessage2) RedeliveryCount() uint32 {
+	return 0
+}
+
+func (msg *mockMessage2) IsReplicated() bool {
+	return false
+}
+
+func (msg *mockMessage2) GetReplicatedFrom() string {
+	return ""
+}
+
+func (msg *mockMessage2) GetSchemaValue(v interface{}) error {
+	return nil
+}
+
+func (msg *mockMessage2) ProducerName() string {
+	return ""
+}
+
+func (msg *mockMessage2) GetEncryptionContext() *EncryptionContext {
+	return &EncryptionContext{}
+}
diff --git a/pulsar/negative_backoff_policy_test.go b/pulsar/negative_backoff_policy_test.go
new file mode 100644
index 0000000..bfbb6a8
--- /dev/null
+++ b/pulsar/negative_backoff_policy_test.go
@@ -0,0 +1,34 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package pulsar
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestDefaultNackBackoffPolicy_Next(t *testing.T) {
+	defaultNackBackoff := new(defaultNackBackoffPolicy)
+
+	res0 := defaultNackBackoff.Next(0)
+	assert.Equal(t, int64(1000*30), res0)
+
+	res5 := defaultNackBackoff.Next(5)
+	assert.Equal(t, int64(600000), res5)
+}