private CoordinatorResult<LeaveGroupResponseData, CoordinatorRecord> classicGroupLeaveToConsumerGroup(
        ConsumerGroup group,
        RequestContext context,
        LeaveGroupRequestData request
    ) throws UnknownMemberIdException {
        String groupId = group.groupId();
        List<MemberResponse> memberResponses = new ArrayList<>();
        Set<ConsumerGroupMember> validLeaveGroupMembers = new HashSet<>();
        List<CoordinatorRecord> records = new ArrayList<>();

        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 {}] Dynamic Member {} has left group " +
                            "through explicit `LeaveGroup` request; client reason: {}",
                        groupId, memberId, reason);
                } else {
                    member = group.staticMember(instanceId);
                    throwIfStaticMemberIsUnknown(member, instanceId);
                    // 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, groupId, memberId, instanceId);
                    }
                    throwIfMemberDoesNotUseClassicProtocol(member);

                    memberId = member.memberId();
                    log.info("[Group {}] Static Member {} with instance id {} has left group " +
                            "through explicit `LeaveGroup` request; client reason: {}",
                        groupId, memberId, instanceId, reason);
                }

                removeMember(records, groupId, memberId);
                cancelTimers(groupId, memberId);
                memberResponses.add(
                    new MemberResponse()
                        .setMemberId(memberId)
                        .setGroupInstanceId(instanceId)
                );
                validLeaveGroupMembers.add(member);
            } catch (KafkaException e) {
                memberResponses.add(
                    new MemberResponse()
                        .setMemberId(memberId)
                        .setGroupInstanceId(instanceId)
                        .setErrorCode(Errors.forException(e).code())
                );
            }
        }

        if (!records.isEmpty()) {
            // Maybe update the subscription metadata.
            Map<String, TopicMetadata> subscriptionMetadata = group.computeSubscriptionMetadata(
                group.computeSubscribedTopicNames(validLeaveGroupMembers),
                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));
            }

            // Bump the group epoch.
            records.add(newGroupEpochRecord(groupId, group.groupEpoch() + 1));
        }

        return new CoordinatorResult<>(records, new LeaveGroupResponseData().setMembers(memberResponses));
    }
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.
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java