fix(connector): clamp exponential backoff shift to avoid uint32 wraparound

pulsar/connector/backoff_test.go

@@ -225,6 +225,15 @@ func TestNackBackoffPolicyExponential(t *testing.T) {
 			expectedDelay:         256 * time.Second, // 2^10 = 1024, but capped at 256
 			expectedValidationErr: false,
 		},
+		{
+			name:                  "Exponential: redelivery count 32 should still hit the cap, not reset to zero",
+			minMultiplier:         1,
+			maxMultiplier:         256,
+			baseDelay:             1 * time.Second,
+			redeliveryCount:       32,
+			expectedDelay:         256 * time.Second,
+			expectedValidationErr: false,
+		},
 	}
 
 	for _, tt := range tests {

pulsar/connector/dlq.go

@@ -89,7 +89,14 @@ func (nbp *NackBackoffPolicy) calculateIncrementalDelay(redeliveryCount uint32)
 
 	if useExponential {
 		// Use exponential backoff: base * 2^redeliveryCount
-		exponentialMultiplier := uint32(1 << redeliveryCount) // 2^redeliveryCount
+		// 1 << n on a uint32 wraps to 0 once n >= 32, so a message stuck in
+		// redelivery for that long would otherwise get an instant redelivery
+		// instead of the (capped) maximum delay. Clamp the shift instead.
+		shift := redeliveryCount
+		if shift >= 32 {
+			shift = 31
+		}
+		exponentialMultiplier := uint32(1) << shift
 		if nbp.maxRedeliveryDelayMultiplier > 0 && exponentialMultiplier > nbp.maxRedeliveryDelayMultiplier {
 			exponentialMultiplier = nbp.maxRedeliveryDelayMultiplier
 		}