Repository: apache/pulsar-client-go
Pull request #661 (2021-11-05 18:09:02)

Remove unused dependency in `oauth2` module
### Modifications

Remove unused dependency in `oauth2` module.

### Verifying this change

- [x] Make sure that the change passes the CI checks.


This change is a trivial rework / code cleanup without any test coverage.

### Documentation

  - Does this pull request introduce a new feature? no



Patch:
diff --git a/oauth2/go.mod b/oauth2/go.mod
index 89d5015..091477d 100644
--- a/oauth2/go.mod
+++ b/oauth2/go.mod
@@ -4,7 +4,6 @@ go 1.13
 
 require (
 	github.com/99designs/keyring v1.1.6
-	github.com/dgrijalva/jwt-go v3.2.0+incompatible
 	github.com/form3tech-oss/jwt-go v3.2.3+incompatible
 	github.com/onsi/ginkgo v1.14.0
 	github.com/onsi/gomega v1.10.1
diff --git a/oauth2/go.sum b/oauth2/go.sum
index 48ffe44..dad3d35 100644
--- a/oauth2/go.sum
+++ b/oauth2/go.sum
@@ -6,8 +6,6 @@ github.com/danieljoos/wincred v1.0.2/go.mod h1:SnuYRW9lp1oJrZX/dXJqr0cPK5gYXqx3E
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/dgrijalva/jwt-go v3.2.0+incompatible h1:7qlOGliEKZXTDg6OTjfoBKDXWrumCAMpl/TFQ4/5kLM=
-github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ=
 github.com/dvsekhvalnov/jose2go v0.0.0-20200901110807-248326c1351b h1:HBah4D48ypg3J7Np4N+HY/ZR76fx3HEUGxDU6Uk39oQ=
 github.com/dvsekhvalnov/jose2go v0.0.0-20200901110807-248326c1351b/go.mod h1:7BvyPhdbLxMXIYTFPLsyJRFMsKmOZnQmzh6Gb+uquuM=
 github.com/form3tech-oss/jwt-go v3.2.3+incompatible h1:7ZaBxOI7TMoYBfyA3cQHErNNyAWIKUMIwqxEtgHOs5c=
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)))
+}


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)
+}