KAFKA-19813: Incorrect jitter value in StreamsGroupHeartbeatRequestManager and AbstractHeartbeatRequestManager

clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractHeartbeatRequestManager.java

@@ -36,6 +36,7 @@
 import java.util.Collections;
 
 import static org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.PollResult.EMPTY;
+import static org.apache.kafka.clients.consumer.internals.RequestState.RETRY_BACKOFF_JITTER;
 
 /**
  * <p>Manages the request creation and response handling for the heartbeat. The module creates a
@@ -113,7 +114,7 @@ public abstract class AbstractHeartbeatRequestManager<R extends AbstractResponse
         long retryBackoffMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MS_CONFIG);
         long retryBackoffMaxMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MAX_MS_CONFIG);
         this.heartbeatRequestState = new HeartbeatRequestState(logContext, time, 0, retryBackoffMs,
-                retryBackoffMaxMs, maxPollIntervalMs);
+                retryBackoffMaxMs, RETRY_BACKOFF_JITTER);
         this.pollTimer = time.timer(maxPollIntervalMs);
         this.metricsManager = metricsManager;
     }

clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java

@@ -50,6 +50,7 @@
 import java.util.stream.IntStream;
 
 import static org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.PollResult.EMPTY;
+import static org.apache.kafka.clients.consumer.internals.RequestState.RETRY_BACKOFF_JITTER;
 
 /**
  * <p>Manages the request creation and response handling for the streams group heartbeat. The class creates a
@@ -330,7 +331,7 @@ public StreamsGroupHeartbeatRequestManager(final LogContext logContext,
             0,
             retryBackoffMs,
             retryBackoffMaxMs,
-            maxPollIntervalMs
+            RETRY_BACKOFF_JITTER
         );
         this.pollTimer = time.timer(maxPollIntervalMs);
     }

clients/src/main/java/org/apache/kafka/common/utils/ExponentialBackoff.java

@@ -19,6 +19,8 @@
 
 import java.util.concurrent.ThreadLocalRandom;
 
+import static java.lang.String.format;
+
 /**
  * A utility class for keeping the parameters and providing the value of exponential
  * retry backoff, exponential reconnect backoff, exponential timeout, etc.
@@ -42,6 +44,9 @@ public ExponentialBackoff(long initialInterval, int multiplier, long maxInterval
         this.initialInterval = Math.min(maxInterval, initialInterval);
         this.multiplier = multiplier;
         this.maxInterval = maxInterval;
+        if (jitter < 0 || jitter > 1) {
+            throw new IllegalArgumentException(format("jitter must be between 0 and 1, but got %s", jitter));
+        }
         this.jitter = jitter;
         this.expMax = maxInterval > initialInterval ?
                 Math.log(maxInterval / (double) Math.max(initialInterval, 1)) / Math.log(multiplier) : 0;

clients/src/test/java/org/apache/kafka/common/utils/ExponentialBackoffTest.java

@@ -20,6 +20,7 @@
 import org.junit.jupiter.api.Test;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 public class ExponentialBackoffTest {
@@ -54,4 +55,14 @@ public void testExponentialBackoffWithoutJitter() {
         assertEquals(400, exponentialBackoff.backoff(2));
         assertEquals(400, exponentialBackoff.backoff(3));
     }
+
+    @Test
+    public void testExponentialBackoffWithInvalidJitter() {
+        assertEquals("jitter must be between 0 and 1, but got -1.0",
+                assertThrows(IllegalArgumentException.class,
+                        () -> new ExponentialBackoff(100, 2, 400, -1)).getMessage());
+        assertEquals("jitter must be between 0 and 1, but got 3000.0",
+                assertThrows(IllegalArgumentException.class,
+                        () -> new ExponentialBackoff(100, 2, 400, 3000)).getMessage());
+    }
 }