[Issue 1384][consumer] Fix the default nack backoff policy (#1385)

Co-authored-by: Copilot <[email protected]>
Co-authored-by: crossoverJie <[email protected]>

Author: 3 people

pulsar/negative_backoff_policy.go

@@ -41,9 +41,11 @@ func (nbp *defaultNackBackoffPolicy) Next(redeliveryCount uint32) time.Duration
 	minNackTime := 1 * time.Second  // 1sec
 	maxNackTime := 10 * time.Minute // 10min
 
-	if redeliveryCount < 0 {
-		return minNackTime
+	backoff := float64(minNackTime << redeliveryCount)
+	if backoff == 0 {
+		// Overflow so we assign the maximum value of the backoff.
+		backoff = float64(maxNackTime)
 	}
 
-	return time.Duration(math.Min(math.Abs(float64(minNackTime<<redeliveryCount)), float64(maxNackTime)))
+	return time.Duration(math.Min(backoff, float64(maxNackTime)))
 }

pulsar/negative_backoff_policy_test.go

@@ -18,6 +18,7 @@
 package pulsar
 
 import (
+	"math"
 	"testing"
 	"time"
 
@@ -27,9 +28,10 @@ import (
 func TestDefaultNackBackoffPolicy_Next(t *testing.T) {
 	defaultNackBackoff := new(defaultNackBackoffPolicy)
 
-	res0 := defaultNackBackoff.Next(0)
-	assert.Equal(t, 1*time.Second, res0)
-
-	res5 := defaultNackBackoff.Next(5)
-	assert.Equal(t, 32*time.Second, res5)
+	assert.Equal(t, 1*time.Second, defaultNackBackoff.Next(0))
+	assert.Equal(t, 32*time.Second, defaultNackBackoff.Next(5))
+	assert.Equal(t, 10*time.Minute, defaultNackBackoff.Next(1000))
+	assert.Equal(t, 10*time.Minute, defaultNackBackoff.Next(math.MaxUint8))
+	assert.Equal(t, 10*time.Minute, defaultNackBackoff.Next(math.MaxUint16))
+	assert.Equal(t, 10*time.Minute, defaultNackBackoff.Next(math.MaxUint32))
 }