KAFKA-8653; Default rebalance timeout to session timeout for JoinGroup v0

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());
     }
 }

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());
+    }
 }

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")