diff --git a/clients/src/main/java/org/apache/kafka/common/requests/JoinGroupRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/JoinGroupRequest.java index a6ad17e5b30a5..95d125d312a17 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/JoinGroupRequest.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/JoinGroupRequest.java @@ -54,7 +54,6 @@ public String toString() { } private final JoinGroupRequestData data; - private final short version; public static final String UNKNOWN_MEMBER_ID = ""; @@ -95,13 +94,21 @@ static boolean containsValidPattern(String topic) { public JoinGroupRequest(JoinGroupRequestData data, short version) { super(ApiKeys.JOIN_GROUP, version); this.data = data; - this.version = version; + maybeOverrideRebalanceTimeout(version); } public JoinGroupRequest(Struct struct, short version) { super(ApiKeys.JOIN_GROUP, version); this.data = new JoinGroupRequestData(struct, version); - this.version = version; + maybeOverrideRebalanceTimeout(version); + } + + private void maybeOverrideRebalanceTimeout(short version) { + if (version == 0) { + // Version 0 has no rebalance timeout, so we use the session timeout + // to be consistent with the original behavior of the API. + data.setRebalanceTimeoutMs(data.sessionTimeoutMs()); + } } public JoinGroupRequestData data() { @@ -149,6 +156,6 @@ public static JoinGroupRequest parse(ByteBuffer buffer, short version) { @Override protected Struct toStruct() { - return data.toStruct(version); + return data.toStruct(version()); } } diff --git a/clients/src/test/java/org/apache/kafka/common/requests/JoinGroupRequestTest.java b/clients/src/test/java/org/apache/kafka/common/requests/JoinGroupRequestTest.java index 644af9ae4f325..9d8031cd30074 100644 --- a/clients/src/test/java/org/apache/kafka/common/requests/JoinGroupRequestTest.java +++ b/clients/src/test/java/org/apache/kafka/common/requests/JoinGroupRequestTest.java @@ -19,11 +19,14 @@ import org.apache.kafka.common.errors.InvalidConfigurationException; import org.apache.kafka.common.errors.UnsupportedVersionException; import org.apache.kafka.common.message.JoinGroupRequestData; +import org.apache.kafka.common.protocol.types.Struct; import org.apache.kafka.test.TestUtils; import org.junit.Test; +import java.nio.ByteBuffer; import java.util.Arrays; +import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.fail; @@ -75,4 +78,25 @@ public void testRequestVersionCompatibilityFailBuild() { .setProtocolType("consumer") ).build((short) 4); } + + @Test + public void testRebalanceTimeoutDefaultsToSessionTimeoutV0() { + int sessionTimeoutMs = 30000; + + Struct struct = new JoinGroupRequestData() + .setGroupId("groupId") + .setMemberId("consumerId") + .setGroupInstanceId("groupInstanceId") + .setProtocolType("consumer") + .setSessionTimeoutMs(sessionTimeoutMs) + .toStruct((short) 0); + + ByteBuffer buffer = ByteBuffer.allocate(struct.sizeOf()); + struct.writeTo(buffer); + buffer.flip(); + + JoinGroupRequest request = JoinGroupRequest.parse(buffer, (short) 0); + assertEquals(sessionTimeoutMs, request.data().sessionTimeoutMs()); + assertEquals(sessionTimeoutMs, request.data().rebalanceTimeoutMs()); + } } diff --git a/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java b/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java index 11c19b90cbbef..f8d45abe3fc0c 100644 --- a/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java +++ b/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java @@ -819,17 +819,7 @@ private JoinGroupRequest createJoinGroupRequest(int version) { .setName("consumer-range") .setMetadata(new byte[0])).iterator() ); - if (version == 0) { - return new JoinGroupRequest.Builder( - new JoinGroupRequestData() - .setGroupId("group1") - .setSessionTimeoutMs(30000) - .setMemberId("consumer1") - .setGroupInstanceId(null) - .setProtocolType("consumer") - .setProtocols(protocols)) - .build((short) version); - } else if (version <= 4) { + if (version <= 4) { return new JoinGroupRequest.Builder( new JoinGroupRequestData() .setGroupId("group1")