KAFKA-9437: KIP-559: Make the Kafka Protocol Friendlier with L7 Proxies

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

@@ -436,7 +436,7 @@ boolean joinGroupIfNeeded(final Timer timer) {
                     // Duplicate the buffer in case `onJoinComplete` does not complete and needs to be retried.
                     ByteBuffer memberAssignment = future.value().duplicate();
 
-                    onJoinComplete(generationSnapshot.generationId, generationSnapshot.memberId, generationSnapshot.protocol, memberAssignment);
+                    onJoinComplete(generationSnapshot.generationId, generationSnapshot.memberId, generationSnapshot.protocolName, memberAssignment);
 
                     // Generally speaking we should always resetJoinGroupFuture once the future is done, but here
                     // we can only reset the join group future after the completion callback returns. This ensures
@@ -574,7 +574,11 @@ private class JoinGroupResponseHandler extends CoordinatorResponseHandler<JoinGr
         @Override
         public void handle(JoinGroupResponse joinResponse, RequestFuture<ByteBuffer> future) {
             Errors error = joinResponse.error();
-            if (error == Errors.NONE) {
+            if (isProtocolTypeInconsistent(joinResponse.data().protocolType())) {
+                log.debug("JoinGroup failed: Received inconsistent ProtocolType ({})",
+                    joinResponse.data().protocolType());
+                future.raise(Errors.INCONSISTENT_GROUP_PROTOCOL);
+            } else if (error == Errors.NONE) {
                 log.debug("Received successful JoinGroup response: {}", joinResponse);
                 sensors.joinSensor.record(response.requestLatencyMs());
 
@@ -654,6 +658,8 @@ private RequestFuture<ByteBuffer> onJoinFollower() {
                         new SyncGroupRequestData()
                                 .setGroupId(rebalanceConfig.groupId)
                                 .setMemberId(generation.memberId)
+                                .setProtocolType(protocolType())
+                                .setProtocolName(generation.protocolName)
                                 .setGroupInstanceId(this.rebalanceConfig.groupInstanceId.orElse(null))
                                 .setGenerationId(generation.generationId)
                                 .setAssignments(Collections.emptyList())
@@ -681,6 +687,8 @@ private RequestFuture<ByteBuffer> onJoinLeader(JoinGroupResponse joinResponse) {
                             new SyncGroupRequestData()
                                     .setGroupId(rebalanceConfig.groupId)
                                     .setMemberId(generation.memberId)
+                                    .setProtocolType(protocolType())
+                                    .setProtocolName(generation.protocolName)
                                     .setGroupInstanceId(this.rebalanceConfig.groupInstanceId.orElse(null))
                                     .setGenerationId(generation.generationId)
                                     .setAssignments(groupAssignmentList)
@@ -704,7 +712,12 @@ private class SyncGroupResponseHandler extends CoordinatorResponseHandler<SyncGr
         public void handle(SyncGroupResponse syncResponse,
                            RequestFuture<ByteBuffer> future) {
             Errors error = syncResponse.error();
-            if (error == Errors.NONE) {
+            if (isProtocolTypeInconsistent(syncResponse.data.protocolType())
+                || isProtocolNameInconsistent(syncResponse.data.protocolName())) {
+                log.debug("SyngGroup failed: Received inconsistent ProtocolType ({}) and/or ProtocolName ({})",
+                    syncResponse.data.protocolType(), syncResponse.data.protocolName());
+                future.raise(Errors.INCONSISTENT_GROUP_PROTOCOL);
+            } else if (error == Errors.NONE) {
                 sensors.syncSensor.record(response.requestLatencyMs());
                 future.complete(ByteBuffer.wrap(syncResponse.data.assignment()));
             } else {
@@ -887,6 +900,14 @@ protected synchronized void requestRejoin() {
         this.rejoinNeeded = true;
     }
 
+    private boolean isProtocolTypeInconsistent(String protocolType) {
+        return protocolType != null && !protocolType.equals(protocolType());
+    }
+
+    private boolean isProtocolNameInconsistent(String protocolName) {
+        return protocolName != null && !protocolName.equals(generation().protocolName);
+    }
+
     /**
      * Close the coordinator, waiting if needed to send LeaveGroup.
      */
@@ -1320,12 +1341,12 @@ protected static class Generation {
 
         public final int generationId;
         public final String memberId;
-        public final String protocol;
+        public final String protocolName;
 
-        public Generation(int generationId, String memberId, String protocol) {
+        public Generation(int generationId, String memberId, String protocolName) {
             this.generationId = generationId;
             this.memberId = memberId;
-            this.protocol = protocol;
+            this.protocolName = protocolName;
         }
 
         /**
@@ -1343,20 +1364,20 @@ public boolean equals(final Object o) {
             final Generation that = (Generation) o;
             return generationId == that.generationId &&
                     Objects.equals(memberId, that.memberId) &&
-                    Objects.equals(protocol, that.protocol);
+                    Objects.equals(protocolName, that.protocolName);
         }
 
         @Override
         public int hashCode() {
-            return Objects.hash(generationId, memberId, protocol);
+            return Objects.hash(generationId, memberId, protocolName);
         }
 
         @Override
         public String toString() {
             return "Generation{" +
                     "generationId=" + generationId +
                     ", memberId='" + memberId + '\'' +
-                    ", protocol='" + protocol + '\'' +
+                    ", protocol='" + protocolName + '\'' +
                     '}';
         }
     }

clients/src/main/java/org/apache/kafka/common/requests/JoinGroupRequest.java

@@ -58,6 +58,7 @@ public String toString() {
     public static final String UNKNOWN_MEMBER_ID = "";
     public static final int UNKNOWN_GENERATION_ID = -1;
     public static final String UNKNOWN_PROTOCOL = "";
+    public static final String UNKNOWN_PROTOCOL_TYPE = null;
 
     private static final int MAX_GROUP_INSTANCE_ID_LENGTH = 249;
 

clients/src/main/java/org/apache/kafka/common/requests/SyncGroupRequest.java

@@ -81,6 +81,17 @@ public Map<String, ByteBuffer> groupAssignments() {
         return groupAssignments;
     }
 
+    /**
+     * ProtocolType and ProtocolName are mandatory since version 5. This methods verifies that
+     * they are defined for version 5 or higher, or returns true otherwise for older versions.
+     */
+    public boolean areMandatoryProtocolTypeAndNamePresent() {
+        if (version() >= 5)
+            return data.protocolType() != null && data.protocolName() != null;
+        else
+            return true;
+    }
+
     public static SyncGroupRequest parse(ByteBuffer buffer, short version) {
         return new SyncGroupRequest(ApiKeys.SYNC_GROUP.parseRequest(version, buffer), version);
     }

clients/src/main/java/org/apache/kafka/common/requests/SyncGroupResponse.java

@@ -62,7 +62,7 @@ protected Struct toStruct(short version) {
     }
 
     public static SyncGroupResponse parse(ByteBuffer buffer, short version) {
-        return new SyncGroupResponse(ApiKeys.SYNC_GROUP.parseResponse(version, buffer));
+        return new SyncGroupResponse(ApiKeys.SYNC_GROUP.parseResponse(version, buffer), version);
     }
 
     @Override

clients/src/main/resources/common/message/JoinGroupRequest.json

@@ -27,7 +27,9 @@
   // with assigned id.
   //
   // Version 6 is the first flexible version.
-  "validVersions": "0-6",
+  //
+  // Version 7 is the same as version 6.
+  "validVersions": "0-7",
   "flexibleVersions": "6+",
   "fields": [
     { "name": "GroupId", "type": "string", "versions": "0+", "entityType": "groupId",

clients/src/main/resources/common/message/JoinGroupResponse.json

@@ -29,7 +29,9 @@
   // Version 5 is bumped to apply group.instance.id to identify member across restarts.
   //
   // Version 6 is the first flexible version.
-  "validVersions": "0-6",
+  //
+  // Starting from version 7, the broker sends back the Protocol Type to the client (KIP-599).
+  "validVersions": "0-7",
   "flexibleVersions": "6+",
   "fields": [
     { "name": "ThrottleTimeMs", "type": "int32", "versions": "2+", "ignorable": true,
@@ -38,6 +40,9 @@
       "about": "The error code, or 0 if there was no error." },
     { "name": "GenerationId", "type": "int32", "versions": "0+", "default": "-1",
       "about": "The generation ID of the group." },
+    { "name": "ProtocolType", "type": "string", "versions": "7+",
+      "nullableVersions": "7+", "default": "null", "ignorable": true,
+      "about": "The group protocol name." },
     { "name": "ProtocolName", "type": "string", "versions": "0+",
       "about": "The group protocol selected by the coordinator." },
     { "name": "Leader", "type": "string", "versions": "0+",

clients/src/main/resources/common/message/SyncGroupRequest.json

@@ -22,7 +22,11 @@
   // Starting from version 3, we add a new field called groupInstanceId to indicate member identity across restarts.
   //
   // Version 4 is the first flexible version.
-  "validVersions": "0-4",
+  //
+  // Starting from version 5, the client sends the Protocol Type and the Protocol Name
+  // to the broker (KIP-599). The broker will reject the request if they are inconsistent
+  // with the Type and Name known by the broker.
+  "validVersions": "0-5",
   "flexibleVersions": "4+",
   "fields": [
     { "name": "GroupId", "type": "string", "versions": "0+", "entityType": "groupId",
@@ -34,6 +38,12 @@
     { "name": "GroupInstanceId", "type": "string", "versions": "3+", 
       "nullableVersions": "3+", "default": "null",
       "about": "The unique identifier of the consumer instance provided by end user." },
+    { "name": "ProtocolType", "type": "string", "versions": "5+",
+      "nullableVersions": "5+", "default": "null", "ignorable": true,
+      "about": "The group protocol type." },
+    { "name": "ProtocolName", "type": "string", "versions": "5+",
+      "nullableVersions": "5+", "default": "null", "ignorable": true,
+      "about": "The group protocol name." },
     { "name": "Assignments", "type": "[]SyncGroupRequestAssignment", "versions": "0+",
       "about": "Each assignment.", "fields": [
       { "name": "MemberId", "type": "string", "versions": "0+",

clients/src/main/resources/common/message/SyncGroupResponse.json

@@ -24,13 +24,22 @@
   // Starting from version 3, syncGroupRequest supports a new field called groupInstanceId to indicate member identity across restarts.
   //
   // Version 4 is the first flexible version.
-  "validVersions": "0-4",
+  //
+  // Starting from version 5, the broker sends back the Protocol Type and the Protocol Name
+  // to the client (KIP-599).
+  "validVersions": "0-5",
   "flexibleVersions": "4+",
   "fields": [
     { "name": "ThrottleTimeMs", "type": "int32", "versions": "1+", "ignorable": true,
       "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
     { "name": "ErrorCode", "type": "int16", "versions": "0+",
       "about": "The error code, or 0 if there was no error." },
+    { "name": "ProtocolType", "type": "string", "versions": "5+",
+      "nullableVersions": "5+", "default": "null", "ignorable": true,
+      "about": "The group protocol type." },
+    { "name": "ProtocolName", "type": "string", "versions": "5+",
+      "nullableVersions": "5+", "default": "null", "ignorable": true,
+      "about": "The group protocol name." },
     { "name": "Assignment", "type": "bytes", "versions": "0+",
       "about": "The member assignment." }
   ]

clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java

@@ -22,6 +22,7 @@
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.errors.AuthenticationException;
 import org.apache.kafka.common.errors.FencedInstanceIdException;
+import org.apache.kafka.common.errors.InconsistentGroupProtocolException;
 import org.apache.kafka.common.errors.UnknownMemberIdException;
 import org.apache.kafka.common.errors.WakeupException;
 import org.apache.kafka.common.internals.ClusterResourceListeners;
@@ -85,6 +86,8 @@ public class AbstractCoordinatorTest {
     private static final int REQUEST_TIMEOUT_MS = 40000;
     private static final String GROUP_ID = "dummy-group";
     private static final String METRIC_GROUP_PREFIX = "consumer";
+    private static final String PROTOCOL_TYPE = "dummy";
+    private static final String PROTOCOL_NAME = "dummy-subprotocol";
 
     private Node node;
     private Metrics metrics;
@@ -360,6 +363,63 @@ public void testJoinGroupRequestWithFencedInstanceIdException() {
         assertFalse(future.isRetriable());
     }
 
+    @Test
+    public void testJoinGroupProtocolTypeAndName() {
+        String wrongProtocolType = "wrong-type";
+        String wrongProtocolName = "wrong-name";
+
+        // No Protocol Type in both JoinGroup and SyncGroup responses
+        assertTrue(joinGroupWithProtocolTypeAndName(null, null, null));
+
+        // Protocol Type in both JoinGroup and SyncGroup responses
+        assertTrue(joinGroupWithProtocolTypeAndName(PROTOCOL_TYPE, PROTOCOL_TYPE, PROTOCOL_NAME));
+
+        // Wrong protocol type in the JoinGroupResponse
+        assertThrows(InconsistentGroupProtocolException.class,
+            () -> joinGroupWithProtocolTypeAndName("wrong", null, null));
+
+        // Correct protocol type in the JoinGroupResponse
+        // Wrong protocol type in the SyncGroupResponse
+        // Correct protocol name in the SyncGroupResponse
+        assertThrows(InconsistentGroupProtocolException.class,
+            () -> joinGroupWithProtocolTypeAndName(PROTOCOL_TYPE, wrongProtocolType, PROTOCOL_NAME));
+
+        // Correct protocol type in the JoinGroupResponse
+        // Correct protocol type in the SyncGroupResponse
+        // Wrong protocol name in the SyncGroupResponse
+        assertThrows(InconsistentGroupProtocolException.class,
+            () -> joinGroupWithProtocolTypeAndName(PROTOCOL_TYPE, PROTOCOL_TYPE, wrongProtocolName));
+    }
+
+    private boolean joinGroupWithProtocolTypeAndName(String joinGroupResponseProtocolType,
+                                                     String syncGroupResponseProtocolType,
+                                                     String syncGroupResponseProtocolName) {
+        setupCoordinator();
+        mockClient.reset();
+        mockClient.prepareResponse(groupCoordinatorResponse(node, Errors.NONE));
+        coordinator.ensureCoordinatorReady(mockTime.timer(0));
+
+        mockClient.prepareResponse(body -> {
+            if (!(body instanceof JoinGroupRequest)) {
+                return false;
+            }
+            JoinGroupRequest joinGroupRequest = (JoinGroupRequest) body;
+            return joinGroupRequest.data().protocolType().equals(PROTOCOL_TYPE);
+        }, joinGroupFollowerResponse(defaultGeneration, memberId,
+            "memberid", Errors.NONE, joinGroupResponseProtocolType));
+
+        mockClient.prepareResponse(body -> {
+            if (!(body instanceof SyncGroupRequest)) {
+                return false;
+            }
+            SyncGroupRequest syncGroupRequest = (SyncGroupRequest) body;
+            return syncGroupRequest.data.protocolType().equals(PROTOCOL_TYPE)
+                && syncGroupRequest.data.protocolName().equals(PROTOCOL_NAME);
+        }, syncGroupResponse(Errors.NONE, syncGroupResponseProtocolType, syncGroupResponseProtocolName));
+
+        return coordinator.joinGroupIfNeeded(mockTime.timer(5000L));
+    }
+
     @Test
     public void testSyncGroupRequestWithFencedInstanceIdException() {
         setupCoordinator();
@@ -946,12 +1006,24 @@ private HeartbeatResponse heartbeatResponse(Errors error) {
         return new HeartbeatResponse(new HeartbeatResponseData().setErrorCode(error.code()));
     }
 
-    private JoinGroupResponse joinGroupFollowerResponse(int generationId, String memberId, String leaderId, Errors error) {
+    private JoinGroupResponse joinGroupFollowerResponse(int generationId,
+                                                        String memberId,
+                                                        String leaderId,
+                                                        Errors error) {
+        return joinGroupFollowerResponse(generationId, memberId, leaderId, error, null);
+    }
+
+    private JoinGroupResponse joinGroupFollowerResponse(int generationId,
+                                                        String memberId,
+                                                        String leaderId,
+                                                        Errors error,
+                                                        String protocolType) {
         return new JoinGroupResponse(
                 new JoinGroupResponseData()
                         .setErrorCode(error.code())
                         .setGenerationId(generationId)
-                        .setProtocolName("dummy-subprotocol")
+                        .setProtocolType(protocolType)
+                        .setProtocolName(PROTOCOL_NAME)
                         .setMemberId(memberId)
                         .setLeader(leaderId)
                         .setMembers(Collections.emptyList())
@@ -964,9 +1036,17 @@ private JoinGroupResponse joinGroupResponse(Errors error) {
     }
 
     private SyncGroupResponse syncGroupResponse(Errors error) {
+        return syncGroupResponse(error, null, null);
+    }
+
+    private SyncGroupResponse syncGroupResponse(Errors error,
+                                                String protocolType,
+                                                String protocolName) {
         return new SyncGroupResponse(
                 new SyncGroupResponseData()
                         .setErrorCode(error.code())
+                        .setProtocolType(protocolType)
+                        .setProtocolName(protocolName)
                         .setAssignment(new byte[0])
         );
     }
@@ -992,14 +1072,14 @@ public DummyCoordinator(GroupRebalanceConfig rebalanceConfig,
 
         @Override
         protected String protocolType() {
-            return "dummy";
+            return PROTOCOL_TYPE;
         }
 
         @Override
         protected JoinGroupRequestData.JoinGroupRequestProtocolCollection metadata() {
             return new JoinGroupRequestData.JoinGroupRequestProtocolCollection(
                     Collections.singleton(new JoinGroupRequestData.JoinGroupRequestProtocol()
-                            .setName("dummy-subprotocol")
+                            .setName(PROTOCOL_NAME)
                             .setMetadata(EMPTY_DATA.array())).iterator()
             );
         }