KAFKA-16832: LeaveGroup API for upgrading ConsumerGroup

checkstyle/suppressions.xml

@@ -347,7 +347,7 @@
     <suppress checks="ClassDataAbstractionCouplingCheck"
               files="(RecordHelpersTest|GroupMetadataManager|GroupMetadataManagerTest|OffsetMetadataManagerTest|GroupCoordinatorServiceTest|GroupCoordinatorShardTest).java"/>
     <suppress checks="JavaNCSS"
-              files="GroupMetadataManagerTest.java"/>
+              files="(GroupMetadataManager|GroupMetadataManagerTest).java"/>
 
     <!-- storage -->
     <suppress checks="CyclomaticComplexity"

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java

@@ -2015,21 +2015,7 @@ private <T> CoordinatorResult<T, CoordinatorRecord> consumerGroupFenceMember(
         if (validateOnlineDowngrade(group, member.memberId())) {
             return convertToClassicGroup(group, member.memberId(), response);
         } else {
-            List<CoordinatorRecord> records = new ArrayList<>();
-            removeMember(records, group.groupId(), member.memberId());
-
-            // We update the subscription metadata without the leaving member.
-            Map<String, TopicMetadata> subscriptionMetadata = group.computeSubscriptionMetadata(
-                group.computeSubscribedTopicNames(member, null),
-                metadataImage.topics(),
-                metadataImage.cluster()
-            );
-
-            if (!subscriptionMetadata.equals(group.subscriptionMetadata())) {
-                log.info("[GroupId {}] Computed new subscription metadata: {}.",
-                    group.groupId(), subscriptionMetadata);
-                records.add(newGroupSubscriptionMetadataRecord(group.groupId(), subscriptionMetadata));
-            }
+            List<CoordinatorRecord> records = removeMemberAndMaybeUpdateSubscriptionMetadata(group, member);
 
             // We bump the group epoch.
             int groupEpoch = group.groupEpoch() + 1;
@@ -2041,6 +2027,36 @@ private <T> CoordinatorResult<T, CoordinatorRecord> consumerGroupFenceMember(
         }
     }
 
+    /**
+     * Remove the member and maybe update the subscription metadata without the removed member.
+     *
+     * @param group     The ConsumerGroup.
+     * @param member    The ConsumerGroupMember.
+     * @return The list of CoordinatorRecord.
+     */
+    private List<CoordinatorRecord> removeMemberAndMaybeUpdateSubscriptionMetadata(
+        ConsumerGroup group,
+        ConsumerGroupMember member
+    ) {
+        List<CoordinatorRecord> records = new ArrayList<>();
+        removeMember(records, group.groupId(), member.memberId());
+
+        // We update the subscription metadata without the leaving member.
+        Map<String, TopicMetadata> subscriptionMetadata = group.computeSubscriptionMetadata(
+            group.computeSubscribedTopicNames(member, null),
+            metadataImage.topics(),
+            metadataImage.cluster()
+        );
+
+        if (!subscriptionMetadata.equals(group.subscriptionMetadata())) {
+            log.info("[GroupId {}] Computed new subscription metadata: {}.",
+                group.groupId(), subscriptionMetadata);
+            records.add(newGroupSubscriptionMetadataRecord(group.groupId(), subscriptionMetadata));
+        }
+
+        return records;
+    }
+
     /**
      * Write tombstones for the member. The order matters here.
      *
@@ -2063,6 +2079,7 @@ private void removeMember(List<CoordinatorRecord> records, String groupId, Strin
     private void cancelTimers(String groupId, String memberId) {
         cancelConsumerGroupSessionTimeout(groupId, memberId);
         cancelConsumerGroupRebalanceTimeout(groupId, memberId);
+        cancelConsumerGroupJoinTimeout(groupId, memberId);
         cancelConsumerGroupSyncTimeout(groupId, memberId);
     }
 
@@ -4424,14 +4441,113 @@ private ConsumerGroupMember validateConsumerGroupMember(
      * @param context        The request context.
      * @param request        The actual LeaveGroup request.
      *
+     * @return The LeaveGroup response and the records to append.
+     */
+    public CoordinatorResult<LeaveGroupResponseData, CoordinatorRecord> classicGroupLeave(
+        RequestContext context,
+        LeaveGroupRequestData request
+    ) throws UnknownMemberIdException, GroupIdNotFoundException {
+        Group group = groups.get(request.groupId(), Long.MAX_VALUE);
+
+        if (group == null) {
+            throw new UnknownMemberIdException(String.format("Group %s not found.", request.groupId()));
+        }
+
+        if (group.type() == CLASSIC) {
+            return classicGroupLeaveToClassicGroup((ClassicGroup) group, context, request);
+        } else {
+            return classicGroupLeaveToConsumerGroup((ConsumerGroup) group, context, request);
+        }
+    }
+
+    /**
+     * Handle a classic LeaveGroupRequest to a ConsumerGroup.
+     *
+     * @param group          The ConsumerGroup.
+     * @param context        The request context.
+     * @param request        The actual LeaveGroup request.
+     *
+     * @return The LeaveGroup response and the records to append.
+     */
+    private CoordinatorResult<LeaveGroupResponseData, CoordinatorRecord> classicGroupLeaveToConsumerGroup(
+        ConsumerGroup group,
+        RequestContext context,
+        LeaveGroupRequestData request
+    ) throws UnknownMemberIdException, GroupIdNotFoundException {
+        List<MemberResponse> memberResponses = new ArrayList<>();
+        List<CoordinatorRecord> records = new ArrayList<>();
+        boolean hasValidLeaveGroupMember = false;
+
+        for (MemberIdentity memberIdentity: request.members()) {
+            String memberId = memberIdentity.memberId();
+            String instanceId = memberIdentity.groupInstanceId();
+            String reason = memberIdentity.reason() != null ? memberIdentity.reason() : "not provided";
+
+            ConsumerGroupMember member;
+            try {
+                if (instanceId == null) {
+                    member = group.getOrMaybeCreateMember(memberId, false);
+                    throwIfMemberDoesNotUseClassicProtocol(member);
+
+                    log.info("[Group {}] Static Member {} has left group " +
+                            "through explicit `LeaveGroup` request; client reason: {}",
+                        group.groupId(), memberId, reason);
+                } else {
+                    member = group.staticMember(instanceId);
+                    throwIfStaticMemberIsUnknown(member, memberId);
+                    // The LeaveGroup API allows administrative removal of members by GroupInstanceId
+                    // in which case we expect the MemberId to be undefined.
+                    if (!UNKNOWN_MEMBER_ID.equals(memberId)) {
+                        throwIfInstanceIdIsFenced(member, group.groupId(), memberId, instanceId);
+                    }
+                    throwIfMemberDoesNotUseClassicProtocol(member);
+
+                    log.info("[Group {}] Static Member {} with instance id {} has left group " +
+                            "through explicit `LeaveGroup` request; client reason: {}",
+                        group.groupId(), memberId, instanceId, reason);
+                }
+
+                records.addAll(removeMemberAndMaybeUpdateSubscriptionMetadata(group, member));
+                cancelTimers(group.groupId(), member.memberId());
+                memberResponses.add(
+                    new MemberResponse()
+                        .setMemberId(member.memberId())
+                        .setGroupInstanceId(instanceId)
+                );
+                hasValidLeaveGroupMember = true;
+            } catch (KafkaException e) {
+                memberResponses.add(
+                    new MemberResponse()
+                        .setMemberId(memberId)
+                        .setGroupInstanceId(instanceId)
+                        .setErrorCode(Errors.forException(e).code())
+                );
+            }
+        }
+
+        if (hasValidLeaveGroupMember) {
+            // Bump the group epoch.
+            records.add(newGroupEpochRecord(group.groupId(), group.groupEpoch() + 1));
+        }
+
+        return new CoordinatorResult<>(records, new LeaveGroupResponseData().setMembers(memberResponses));
+    }
+
+    /**
+     * Handle a classic LeaveGroupRequest to a ClassicGroup.
+     *
+     * @param group          The ClassicGroup.
+     * @param context        The request context.
+     * @param request        The actual LeaveGroup request.
+     *
      * @return The LeaveGroup response and the GroupMetadata record to append if the group
      *         no longer has any members.
      */
-    public CoordinatorResult<LeaveGroupResponseData, CoordinatorRecord> classicGroupLeave(
+    private CoordinatorResult<LeaveGroupResponseData, CoordinatorRecord> classicGroupLeaveToClassicGroup(
+        ClassicGroup group,
         RequestContext context,
         LeaveGroupRequestData request
     ) throws UnknownMemberIdException, GroupIdNotFoundException {
-        ClassicGroup group = getOrMaybeCreateClassicGroup(request.groupId(), false);
         if (group.isInState(DEAD)) {
             return new CoordinatorResult<>(
                 Collections.emptyList(),