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
@@ -575,21 +575,28 @@ private class JoinGroupResponseHandler extends CoordinatorResponseHandler<JoinGr
         public void handle(JoinGroupResponse joinResponse, RequestFuture<ByteBuffer> future) {
             Errors error = joinResponse.error();
             if (error == Errors.NONE) {
-                log.debug("Received successful JoinGroup response: {}", joinResponse);
-                sensors.joinSensor.record(response.requestLatencyMs());
+                if (isProtocolTypeInconsistent(joinResponse.data().protocolType())) {
+                    log.debug("JoinGroup failed due to inconsistent Protocol Type, received {} but expected {}",
+                        joinResponse.data().protocolType(), protocolType());
+                    future.raise(Errors.INCONSISTENT_GROUP_PROTOCOL);
+                } else {
+                    log.debug("Received successful JoinGroup response: {}", joinResponse);
+                    sensors.joinSensor.record(response.requestLatencyMs());
 
-                synchronized (AbstractCoordinator.this) {
-                    if (state != MemberState.REBALANCING) {
-                        // if the consumer was woken up before a rebalance completes, we may have already left
-                        // the group. In this case, we do not want to continue with the sync group.
-                        future.raise(new UnjoinedGroupException());
-                    } else {
-                        AbstractCoordinator.this.generation = new Generation(joinResponse.data().generationId(),
-                                joinResponse.data().memberId(), joinResponse.data().protocolName());
-                        if (joinResponse.isLeader()) {
-                            onJoinLeader(joinResponse).chain(future);
+                    synchronized (AbstractCoordinator.this) {
+                        if (state != MemberState.REBALANCING) {
+                            // if the consumer was woken up before a rebalance completes, we may have already left
+                            // the group. In this case, we do not want to continue with the sync group.
+                            future.raise(new UnjoinedGroupException());
                         } else {
-                            onJoinFollower().chain(future);
+                            AbstractCoordinator.this.generation = new Generation(
+                                joinResponse.data().generationId(),
+                                joinResponse.data().memberId(), joinResponse.data().protocolName());
+                            if (joinResponse.isLeader()) {
+                                onJoinLeader(joinResponse).chain(future);
+                            } else {
+                                onJoinFollower().chain(future);
+                            }
                         }
                     }
                 }
@@ -654,6 +661,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 +690,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)
@@ -705,8 +716,18 @@ public void handle(SyncGroupResponse syncResponse,
                            RequestFuture<ByteBuffer> future) {
             Errors error = syncResponse.error();
             if (error == Errors.NONE) {
-                sensors.syncSensor.record(response.requestLatencyMs());
-                future.complete(ByteBuffer.wrap(syncResponse.data.assignment()));
+                if (isProtocolTypeInconsistent(syncResponse.data.protocolType())) {
+                    log.debug("SyncGroup failed due to inconsistent Protocol Type, received {} but expected {}",
+                        syncResponse.data.protocolType(), protocolType());
+                    future.raise(Errors.INCONSISTENT_GROUP_PROTOCOL);
+                } else if (isProtocolNameInconsistent(syncResponse.data.protocolName())) {
+                    log.debug("SyncGroup failed due to inconsistent Protocol Name, received {} but expected {}",
+                        syncResponse.data.protocolName(), generation().protocolName);
+                    future.raise(Errors.INCONSISTENT_GROUP_PROTOCOL);
+                } else {
+                    sensors.syncSensor.record(response.requestLatencyMs());
+                    future.complete(ByteBuffer.wrap(syncResponse.data.assignment()));
+                }
             } else {
                 requestRejoin();
 
@@ -887,6 +908,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 +1349,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 +1372,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

@@ -57,7 +57,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_NAME = "";
 
     private static final int MAX_GROUP_INSTANCE_ID_LENGTH = 249;
 
@@ -119,14 +119,21 @@ public JoinGroupRequestData data() {
 
     @Override
     public AbstractResponse getErrorResponse(int throttleTimeMs, Throwable e) {
-        return new JoinGroupResponse(new JoinGroupResponseData()
-                .setThrottleTimeMs(throttleTimeMs)
-                .setErrorCode(Errors.forException(e).code())
-                .setGenerationId(UNKNOWN_GENERATION_ID)
-                .setProtocolName(UNKNOWN_PROTOCOL)
-                .setLeader(UNKNOWN_MEMBER_ID)
-                .setMemberId(UNKNOWN_MEMBER_ID)
-                .setMembers(Collections.emptyList()));
+        JoinGroupResponseData data = new JoinGroupResponseData()
+            .setThrottleTimeMs(throttleTimeMs)
+            .setErrorCode(Errors.forException(e).code())
+            .setGenerationId(UNKNOWN_GENERATION_ID)
+            .setProtocolName(UNKNOWN_PROTOCOL_NAME)
+            .setLeader(UNKNOWN_MEMBER_ID)
+            .setMemberId(UNKNOWN_MEMBER_ID)
+            .setMembers(Collections.emptyList());
+
+        if (version() >= 7)
+            data.setProtocolName(null);
+        else
+            data.setProtocolName(UNKNOWN_PROTOCOL_NAME);
+
+        return new JoinGroupResponse(data);
     }
 
     public static JoinGroupRequest parse(ByteBuffer buffer, short version) {

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,7 +40,10 @@
       "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": "ProtocolName", "type": "string", "versions": "0+",
+    { "name": "ProtocolType", "type": "string", "versions": "7+",
+      "nullableVersions": "7+", "default": "null", "ignorable": true,
+      "about": "The group protocol name." },
+    { "name": "ProtocolName", "type": "string", "versions": "0+", "nullableVersions": "7+",
       "about": "The group protocol selected by the coordinator." },
     { "name": "Leader", "type": "string", "versions": "0+",
       "about": "The leader of the group." },

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." }
   ]