Repository: apache/pulsar-client-go (go)
Issue: [Bug] default nack backoff policy overflows to zero
#### Expected behavior

The default nack backoff policy should return the maximum backoff time once reaching an arbitrary high redelivery count.

#### Actual behavior

The default nack backoff policy returns zero once a sufficiently big redelivery count is reached.

#### Steps to reproduce

```Go
func TestDefaultNackBackoffPolicy_Next(t *testing.T) {
	defaultNackBackoff := new(defaultNackBackoffPolicy)

	res := defaultNackBackoff.Next(math.MaxUint32)
	assert.Equal(t, 10*time.Minute, res)
}
```

#### System configuration
**Pulsar version**: v4.0.5
**Pulsar Go client**: v0.15.1

Pull request description: [Issue 1384][consumer] Fix the default nack backoff policy
<!--
### Contribution Checklist
  
  - Name the pull request in the form "[Issue XYZ][component] Title of the pull request", where *XYZ* should be replaced by the actual issue number.
    Skip *Issue XYZ* if there is no associated github issue for this pull request.
    Skip *component* if you are unsure about which is the best component. E.g. `[docs] Fix typo in produce method`.

  - Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
  
  - Each pull request should address only one issue, not mix up code from multiple issues.
  
  - Each commit in the pull request has a meaningful commit message

  - Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.

**(The sections below can be removed for hotfixes of typos)**
-->

Fixes #1384 

### Motivation


Once the redelivery count is high enough, the backoff time overflows to zero causing immediate redelivery.

### Modifications

Check for the overflow to assign the max backoff time in that case.

### Verifying this change

- [x] Make sure that the change passes the CI checks.


This change is already covered by existing tests, such as:
- Extend `TestDefaultNackBackoffPolicy_Next` with failing cases added.

### Does this pull request potentially affect one of the following parts:

*If `yes` was chosen, please highlight the changes*

  - Dependencies (does it add or upgrade a dependency): no
  - The public API: no
  - The schema: no
  - The default values of configurations: no
  - The wire protocol: no

### Documentation

  - Does this pull request introduce a new feature? no


Patch:
diff --git a/pulsar/negative_backoff_policy.go b/pulsar/negative_backoff_policy.go
index 5cd35bc..937c210 100644
--- a/pulsar/negative_backoff_policy.go
+++ b/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)))
 }


Test patch:
diff --git a/pulsar/negative_backoff_policy_test.go b/pulsar/negative_backoff_policy_test.go
index f37cf7c..7083f74 100644
--- a/pulsar/negative_backoff_policy_test.go
+++ b/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))
 }