/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.kafka.coordinator.group.modern.consumer;

import org.apache.kafka.clients.consumer.internals.ConsumerProtocol;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.errors.ApiException;
import org.apache.kafka.common.errors.IllegalGenerationException;
import org.apache.kafka.common.errors.StaleMemberEpochException;
import org.apache.kafka.common.errors.UnknownMemberIdException;
import org.apache.kafka.common.errors.UnsupportedVersionException;
import org.apache.kafka.common.message.ConsumerGroupDescribeResponseData;
import org.apache.kafka.common.message.ConsumerProtocolSubscription;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.coordinator.group.CoordinatorRecord;
import org.apache.kafka.coordinator.group.CoordinatorRecordHelpers;
import org.apache.kafka.coordinator.group.OffsetExpirationCondition;
import org.apache.kafka.coordinator.group.OffsetExpirationConditionImpl;
import org.apache.kafka.coordinator.group.Utils;
import org.apache.kafka.coordinator.group.classic.ClassicGroup;
import org.apache.kafka.coordinator.group.generated.ConsumerGroupMemberMetadataValue;
import org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetricsShard;
import org.apache.kafka.coordinator.group.modern.Assignment;
import org.apache.kafka.coordinator.group.modern.MemberState;
import org.apache.kafka.coordinator.group.modern.ModernGroup;
import org.apache.kafka.coordinator.group.modern.ModernGroupMember;
import org.apache.kafka.image.TopicsImage;
import org.apache.kafka.timeline.SnapshotRegistry;
import org.apache.kafka.timeline.TimelineHashMap;
import org.apache.kafka.timeline.TimelineInteger;
import org.apache.kafka.timeline.TimelineObject;

import java.nio.ByteBuffer;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;

import static org.apache.kafka.coordinator.group.Utils.toOptional;
import static org.apache.kafka.coordinator.group.Utils.toTopicPartitionMap;
import static org.apache.kafka.coordinator.group.modern.consumer.ConsumerGroup.ConsumerGroupState.ASSIGNING;
import static org.apache.kafka.coordinator.group.modern.consumer.ConsumerGroup.ConsumerGroupState.EMPTY;
import static org.apache.kafka.coordinator.group.modern.consumer.ConsumerGroup.ConsumerGroupState.RECONCILING;
import static org.apache.kafka.coordinator.group.modern.consumer.ConsumerGroup.ConsumerGroupState.STABLE;

/**
 * A Consumer Group. All the metadata in this class are backed by
 * records in the __consumer_offsets partitions.
 */
public class ConsumerGroup extends ModernGroup<ConsumerGroupMember> {

    public enum ConsumerGroupState {
        EMPTY("Empty"),
        ASSIGNING("Assigning"),
        RECONCILING("Reconciling"),
        STABLE("Stable"),
        DEAD("Dead");

        private final String name;

        private final String lowerCaseName;

        ConsumerGroupState(String name) {
            this.name = name;
            this.lowerCaseName = name.toLowerCase(Locale.ROOT);
        }

        @Override
        public String toString() {
            return name;
        }

        public String toLowerCaseString() {
            return lowerCaseName;
        }
    }

    /**
     * The group state.
     */
    private final TimelineObject<ConsumerGroupState> state;

    /**
     * The static group members.
     */
    private final TimelineHashMap<String, String> staticMembers;

    /**
     * The number of members supporting each server assignor name.
     */
    private final TimelineHashMap<String, Integer> serverAssignors;

    /**
     * The coordinator metrics.
     */
    private final GroupCoordinatorMetricsShard metrics;

    /**
     * The number of members that use the classic protocol.
     */
    private final TimelineInteger numClassicProtocolMembers;

    /**
     * Map of protocol names to the number of members that use classic protocol and support them.
     */
    private final TimelineHashMap<String, Integer> classicProtocolMembersSupportedProtocols;

    public ConsumerGroup(
        SnapshotRegistry snapshotRegistry,
        String groupId,
        GroupCoordinatorMetricsShard metrics
    ) {
        super(snapshotRegistry, groupId);
        this.state = new TimelineObject<>(snapshotRegistry, EMPTY);
        this.staticMembers = new TimelineHashMap<>(snapshotRegistry, 0);
        this.serverAssignors = new TimelineHashMap<>(snapshotRegistry, 0);
        this.metrics = Objects.requireNonNull(metrics);
        this.numClassicProtocolMembers = new TimelineInteger(snapshotRegistry);
        this.classicProtocolMembersSupportedProtocols = new TimelineHashMap<>(snapshotRegistry, 0);
    }

    /**
     * @return The group type (Consumer).
     */
    @Override
    public GroupType type() {
        return GroupType.CONSUMER;
    }

    /**
     * @return The group protocol type (consumer).
     */
    @Override
    public String protocolType() {
        return ConsumerProtocol.PROTOCOL_TYPE;
    }

    /**
     * @return The current state as a String.
     */
    @Override
    public String stateAsString() {
        return state.get().toString();
    }

    /**
     * @return The current state as a String with given committedOffset.
     */
    public String stateAsString(long committedOffset) {
        return state.get(committedOffset).toString();
    }

    /**
     * @return The current state.
     */
    public ConsumerGroupState state() {
        return state.get();
    }

    /**
     * @return The current state based on committed offset.
     */
    public ConsumerGroupState state(long committedOffset) {
        return state.get(committedOffset);
    }

    /**
     * Sets the number of members using the classic protocol.
     *
     * @param numClassicProtocolMembers The new NumClassicProtocolMembers.
     */
    public void setNumClassicProtocolMembers(int numClassicProtocolMembers) {
        this.numClassicProtocolMembers.set(numClassicProtocolMembers);
    }

    /**
     * Get member id of a static member that matches the given group
     * instance id.
     *
     * @param groupInstanceId The group instance id.
     *
     * @return The member id corresponding to the given instance id or null if it does not exist
     */
    public String staticMemberId(String groupInstanceId) {
        if (groupInstanceId == null) return null;
        return staticMembers.get(groupInstanceId);
    }

    /**
     * Gets or creates a new member but without adding it to the group. Adding a member
     * is done via the {@link ConsumerGroup#updateMember(ConsumerGroupMember)} method.
     *
     * @param memberId          The member id.
     * @param createIfNotExists Booleans indicating whether the member must be
     *                          created if it does not exist.
     *
     * @return A ConsumerGroupMember.
     */
    public ConsumerGroupMember getOrMaybeCreateMember(
        String memberId,
        boolean createIfNotExists
    ) {
        ConsumerGroupMember member = members.get(memberId);
        if (member != null) return member;

        if (!createIfNotExists) {
            throw new UnknownMemberIdException(
                String.format("Member %s is not a member of group %s.", memberId, groupId)
            );
        }

        return new ConsumerGroupMember.Builder(memberId).build();
    }

    /**
     * Gets a static member.
     *
     * @param instanceId The group instance id.
     *
     * @return The member corresponding to the given instance id or null if it does not exist
     */
    public ConsumerGroupMember staticMember(String instanceId) {
        String existingMemberId = staticMemberId(instanceId);
        return existingMemberId == null ? null : getOrMaybeCreateMember(existingMemberId, false);
    }

    /**
     * Returns true if the static member exists.
     *
     * @param instanceId The instance id.
     *
     * @return A boolean indicating whether the member exists or not.
     */
    public boolean hasStaticMember(String instanceId) {
        if (instanceId == null) return false;
        return staticMembers.containsKey(instanceId);
    }

    /**
     * Returns the target assignment associated to the provided member id if
     * the instance id is null; otherwise returns the target assignment associated
     * to the instance id.
     *
     * @param memberId      The member id.
     * @param instanceId    The instance id.
     *
     * @return The Assignment or EMPTY if it does not exist.
     */
    public Assignment targetAssignment(String memberId, String instanceId) {
        if (instanceId == null) {
            return targetAssignment(memberId);
        } else {
            String previousMemberId = staticMemberId(instanceId);
            if (previousMemberId != null) {
                return targetAssignment(previousMemberId);
            }
        }
        return Assignment.EMPTY;
    }

    @Override
    public void updateMember(ConsumerGroupMember newMember) {
        if (newMember == null) {
            throw new IllegalArgumentException("newMember cannot be null.");
        }
        ConsumerGroupMember oldMember = members.put(newMember.memberId(), newMember);
        maybeUpdateSubscribedTopicNamesAndGroupSubscriptionType(oldMember, newMember);
        maybeUpdateServerAssignors(oldMember, newMember);
        maybeUpdatePartitionEpoch(oldMember, newMember);
        updateStaticMember(newMember);
        maybeUpdateGroupState();
        maybeUpdateNumClassicProtocolMembers(oldMember, newMember);
        maybeUpdateClassicProtocolMembersSupportedProtocols(oldMember, newMember);
    }

    /**
     * Updates the member id stored against the instance id if the member is a static member.
     *
     * @param newMember The new member state.
     */
    private void updateStaticMember(ConsumerGroupMember newMember) {
        if (newMember.instanceId() != null) {
            staticMembers.put(newMember.instanceId(), newMember.memberId());
        }
    }

    @Override
    public void removeMember(String memberId) {
        ConsumerGroupMember oldMember = members.remove(memberId);
        maybeUpdateSubscribedTopicNamesAndGroupSubscriptionType(oldMember, null);
        maybeUpdateServerAssignors(oldMember, null);
        maybeRemovePartitionEpoch(oldMember);
        removeStaticMember(oldMember);
        maybeUpdateGroupState();
        maybeUpdateNumClassicProtocolMembers(oldMember, null);
        maybeUpdateClassicProtocolMembersSupportedProtocols(oldMember, null);
    }

    /**
     * Remove the static member mapping if the removed member is static.
     *
     * @param oldMember The member to remove.
     */
    private void removeStaticMember(ConsumerGroupMember oldMember) {
        if (oldMember.instanceId() != null) {
            staticMembers.remove(oldMember.instanceId());
        }
    }

    /**
     * @return The number of members that use the classic protocol.
     */
    public int numClassicProtocolMembers() {
        return numClassicProtocolMembers.get();
    }

    /**
     * @return The map of the protocol name and the number of members using the classic protocol that support it.
     */
    public Map<String, Integer> classicMembersSupportedProtocols() {
        return Collections.unmodifiableMap(classicProtocolMembersSupportedProtocols);
    }

    /**
     * @return An immutable Map containing all the static members keyed by instance id.
     */
    public Map<String, String> staticMembers() {
        return Collections.unmodifiableMap(staticMembers);
    }

    /**
     * Compute the preferred (server side) assignor for the group while
     * taking into account the updated member. The computation relies
     * on {{@link ConsumerGroup#serverAssignors}} persisted structure
     * but it does not update it.
     *
     * @param oldMember The old member.
     * @param newMember The new member.
     *
     * @return An Optional containing the preferred assignor.
     */
    public Optional<String> computePreferredServerAssignor(
        ConsumerGroupMember oldMember,
        ConsumerGroupMember newMember
    ) {
        // Copy the current count and update it.
        Map<String, Integer> counts = new HashMap<>(this.serverAssignors);
        maybeUpdateServerAssignors(counts, oldMember, newMember);

        return counts.entrySet().stream()
            .max(Map.Entry.comparingByValue())
            .map(Map.Entry::getKey);
    }

    /**
     * @return The preferred assignor for the group.
     */
    public Optional<String> preferredServerAssignor() {
        return preferredServerAssignor(Long.MAX_VALUE);
    }

    /**
     * @return The preferred assignor for the group with given offset.
     */
    public Optional<String> preferredServerAssignor(long committedOffset) {
        return serverAssignors.entrySet(committedOffset).stream()
            .max(Map.Entry.comparingByValue())
            .map(Map.Entry::getKey);
    }

    /**
     * Validates the OffsetCommit request.
     *
     * @param memberId          The member id.
     * @param groupInstanceId   The group instance id.
     * @param memberEpoch       The member epoch.
     * @param isTransactional   Whether the offset commit is transactional or not. It has no
     *                          impact when a consumer group is used.
     * @param apiVersion        The api version.
     * @throws UnknownMemberIdException     If the member is not found.
     * @throws StaleMemberEpochException    If the member uses the consumer protocol and the provided
     *                                      member epoch doesn't match the actual member epoch.
     * @throws IllegalGenerationException   If the member uses the classic protocol and the provided
     *                                      generation id is not equal to the member epoch.
     */
    @Override
    public void validateOffsetCommit(
        String memberId,
        String groupInstanceId,
        int memberEpoch,
        boolean isTransactional,
        short apiVersion
    ) throws UnknownMemberIdException, StaleMemberEpochException, IllegalGenerationException {
        // When the member epoch is -1, the request comes from either the admin client
        // or a consumer which does not use the group management facility. In this case,
        // the request can commit offsets if the group is empty.
        if (memberEpoch < 0 && members().isEmpty()) return;

        final ConsumerGroupMember member = getOrMaybeCreateMember(memberId, false);

        // If the commit is not transactional and the member uses the new consumer protocol (KIP-848),
        // the member should be using the OffsetCommit API version >= 9.
        if (!isTransactional && !member.useClassicProtocol() && apiVersion < 9) {
            throw new UnsupportedVersionException("OffsetCommit version 9 or above must be used " +
                "by members using the modern group protocol");
        }

        validateMemberEpoch(memberEpoch, member.memberEpoch(), member.useClassicProtocol());
    }

    /**
     * Validates the OffsetFetch request.
     *
     * @param memberId              The member id for consumer groups.
     * @param memberEpoch           The member epoch for consumer groups.
     * @param lastCommittedOffset   The last committed offsets in the timeline.
     * @throws UnknownMemberIdException     If the member is not found.
     * @throws StaleMemberEpochException    If the member uses the consumer protocol and the provided
     *                                      member epoch doesn't match the actual member epoch.
     * @throws IllegalGenerationException   If the member uses the classic protocol and the provided
     *                                      generation id is not equal to the member epoch.
     */
    @Override
    public void validateOffsetFetch(
        String memberId,
        int memberEpoch,
        long lastCommittedOffset
    ) throws UnknownMemberIdException, StaleMemberEpochException, IllegalGenerationException {
        // When the member id is null and the member epoch is -1, the request either comes
        // from the admin client or from a client which does not provide them. In this case,
        // the fetch request is accepted.
        if (memberId == null && memberEpoch < 0) return;

        final ConsumerGroupMember member = members.get(memberId, lastCommittedOffset);
        if (member == null) {
            throw new UnknownMemberIdException(String.format("Member %s is not a member of group %s.",
                memberId, groupId));
        }
        validateMemberEpoch(memberEpoch, member.memberEpoch(), member.useClassicProtocol());
    }

    /**
     * Validates the OffsetDelete request.
     */
    @Override
    public void validateOffsetDelete() {
        // Do nothing.
    }

    /**
     * Validates the DeleteGroups request.
     */
    @Override
    public void validateDeleteGroup() throws ApiException {
        if (state() != ConsumerGroupState.EMPTY) {
            throw Errors.NON_EMPTY_GROUP.exception();
        }
    }

    /**
     * Populates the list of records with tombstone(s) for deleting the group.
     *
     * @param records The list of records.
     */
    @Override
    public void createGroupTombstoneRecords(List<CoordinatorRecord> records) {
        members().forEach((memberId, member) ->
            records.add(CoordinatorRecordHelpers.newCurrentAssignmentTombstoneRecord(groupId(), memberId))
        );

        members().forEach((memberId, member) ->
            records.add(CoordinatorRecordHelpers.newTargetAssignmentTombstoneRecord(groupId(), memberId))
        );
        records.add(CoordinatorRecordHelpers.newTargetAssignmentEpochTombstoneRecord(groupId()));

        members().forEach((memberId, member) ->
            records.add(CoordinatorRecordHelpers.newMemberSubscriptionTombstoneRecord(groupId(), memberId))
        );

        records.add(CoordinatorRecordHelpers.newGroupSubscriptionMetadataTombstoneRecord(groupId()));
        records.add(CoordinatorRecordHelpers.newGroupEpochTombstoneRecord(groupId()));
    }

    @Override
    public boolean isEmpty() {
        return state() == ConsumerGroupState.EMPTY;
    }

    /**
     * See {@link org.apache.kafka.coordinator.group.OffsetExpirationCondition}
     *
     * @return The offset expiration condition for the group or Empty if no such condition exists.
     */
    @Override
    public Optional<OffsetExpirationCondition> offsetExpirationCondition() {
        return Optional.of(new OffsetExpirationConditionImpl(offsetAndMetadata -> offsetAndMetadata.commitTimestampMs));
    }

    @Override
    public boolean isInStates(Set<String> statesFilter, long committedOffset) {
        return statesFilter.contains(state.get(committedOffset).toLowerCaseString());
    }

    /**
     * Throws an exception if the received member epoch does not match the expected member epoch.
     *
     * @param receivedMemberEpoch   The received member epoch or generation id.
     * @param expectedMemberEpoch   The expected member epoch.
     * @param useClassicProtocol    The boolean indicating whether the checked member uses the classic protocol.
     * @throws StaleMemberEpochException    if the member with unmatched member epoch uses the consumer protocol.
     * @throws IllegalGenerationException   if the member with unmatched generation id uses the classic protocol.
     */
    private void validateMemberEpoch(
        int receivedMemberEpoch,
        int expectedMemberEpoch,
        boolean useClassicProtocol
    ) throws StaleMemberEpochException, IllegalGenerationException {
        if (receivedMemberEpoch != expectedMemberEpoch) {
            if (useClassicProtocol) {
                throw new IllegalGenerationException(String.format("The received generation id %d does not match " +
                    "the expected member epoch %d.", receivedMemberEpoch, expectedMemberEpoch));
            } else {
                throw new StaleMemberEpochException(String.format("The received member epoch %d does not match "
                    + "the expected member epoch %d.", receivedMemberEpoch, expectedMemberEpoch));
            }
        }
    }

    @Override
    protected void maybeUpdateGroupState() {
        ConsumerGroupState previousState = state.get();
        ConsumerGroupState newState = STABLE;
        if (members.isEmpty()) {
            newState = EMPTY;
        } else if (groupEpoch.get() > targetAssignmentEpoch.get()) {
            newState = ASSIGNING;
        } else {
            for (ModernGroupMember member : members.values()) {
                if (!member.isReconciledTo(targetAssignmentEpoch.get())) {
                    newState = RECONCILING;
                    break;
                }
            }
        }

        state.set(newState);
        metrics.onConsumerGroupStateTransition(previousState, newState);
    }

    /**
     * Updates the server assignors count.
     *
     * @param oldMember The old member.
     * @param newMember The new member.
     */
    private void maybeUpdateServerAssignors(
        ConsumerGroupMember oldMember,
        ConsumerGroupMember newMember
    ) {
        maybeUpdateServerAssignors(serverAssignors, oldMember, newMember);
    }

    /**
     * Updates the server assignors count.
     *
     * @param serverAssignorCount   The count to update.
     * @param oldMember             The old member.
     * @param newMember             The new member.
     */
    private static void maybeUpdateServerAssignors(
        Map<String, Integer> serverAssignorCount,
        ConsumerGroupMember oldMember,
        ConsumerGroupMember newMember
    ) {
        if (oldMember != null) {
            oldMember.serverAssignorName().ifPresent(name ->
                serverAssignorCount.compute(name, Utils::decValue)
            );
        }
        if (newMember != null) {
            newMember.serverAssignorName().ifPresent(name ->
                serverAssignorCount.compute(name, Utils::incValue)
            );
        }
    }

    /**
     * Updates the number of the members that use the classic protocol.
     *
     * @param oldMember The old member.
     * @param newMember The new member.
     */
    private void maybeUpdateNumClassicProtocolMembers(
        ConsumerGroupMember oldMember,
        ConsumerGroupMember newMember
    ) {
        int delta = 0;
        if (oldMember != null && oldMember.useClassicProtocol()) {
            delta--;
        }
        if (newMember != null && newMember.useClassicProtocol()) {
            delta++;
        }
        setNumClassicProtocolMembers(numClassicProtocolMembers() + delta);
    }

    /**
     * Updates the supported protocol count of the members that use the classic protocol.
     *
     * @param oldMember The old member.
     * @param newMember The new member.
     */
    private void maybeUpdateClassicProtocolMembersSupportedProtocols(
        ConsumerGroupMember oldMember,
        ConsumerGroupMember newMember
    ) {
        if (oldMember != null) {
            oldMember.supportedClassicProtocols().ifPresent(protocols ->
                protocols.forEach(protocol ->
                    classicProtocolMembersSupportedProtocols.compute(protocol.name(), Utils::decValue)
                )
            );
        }
        if (newMember != null) {
            newMember.supportedClassicProtocols().ifPresent(protocols ->
                protocols.forEach(protocol ->
                    classicProtocolMembersSupportedProtocols.compute(protocol.name(), Utils::incValue)
                )
            );
        }
    }

    /**
     * Updates the partition epochs based on the old and the new member.
     *
     * @param oldMember The old member.
     * @param newMember The new member.
     */
    private void maybeUpdatePartitionEpoch(
        ConsumerGroupMember oldMember,
        ConsumerGroupMember newMember
    ) {
        maybeRemovePartitionEpoch(oldMember);
        addPartitionEpochs(newMember.assignedPartitions(), newMember.memberEpoch());
        addPartitionEpochs(newMember.partitionsPendingRevocation(), newMember.memberEpoch());
    }

    /**
     * Removes the partition epochs for the provided member.
     *
     * @param oldMember The old member.
     */
    private void maybeRemovePartitionEpoch(
        ConsumerGroupMember oldMember
    ) {
        if (oldMember != null) {
            removePartitionEpochs(oldMember.assignedPartitions(), oldMember.memberEpoch());
            removePartitionEpochs(oldMember.partitionsPendingRevocation(), oldMember.memberEpoch());
        }
    }

    public ConsumerGroupDescribeResponseData.DescribedGroup asDescribedGroup(
        long committedOffset,
        String defaultAssignor,
        TopicsImage topicsImage
    ) {
        ConsumerGroupDescribeResponseData.DescribedGroup describedGroup = new ConsumerGroupDescribeResponseData.DescribedGroup()
            .setGroupId(groupId)
            .setAssignorName(preferredServerAssignor(committedOffset).orElse(defaultAssignor))
            .setGroupEpoch(groupEpoch.get(committedOffset))
            .setGroupState(state.get(committedOffset).toString())
            .setAssignmentEpoch(targetAssignmentEpoch.get(committedOffset));
        members.entrySet(committedOffset).forEach(
            entry -> describedGroup.members().add(
                entry.getValue().asConsumerGroupDescribeMember(
                    targetAssignment.get(entry.getValue().memberId(), committedOffset),
                    topicsImage
                )
            )
        );
        return describedGroup;
    }

    /**
     * Create a new consumer group according to the given classic group.
     *
     * @param snapshotRegistry  The SnapshotRegistry.
     * @param metrics           The GroupCoordinatorMetricsShard.
     * @param classicGroup      The converted classic group.
     * @param topicsImage       The TopicsImage for topic id and topic name conversion.
     * @return  The created ConsumerGruop.
     */
    public static ConsumerGroup fromClassicGroup(
        SnapshotRegistry snapshotRegistry,
        GroupCoordinatorMetricsShard metrics,
        ClassicGroup classicGroup,
        TopicsImage topicsImage
    ) {
        String groupId = classicGroup.groupId();
        ConsumerGroup consumerGroup = new ConsumerGroup(snapshotRegistry, groupId, metrics);
        consumerGroup.setGroupEpoch(classicGroup.generationId());
        consumerGroup.setTargetAssignmentEpoch(classicGroup.generationId());

        classicGroup.allMembers().forEach(classicGroupMember -> {
            Map<Uuid, Set<Integer>> assignedPartitions = toTopicPartitionMap(
                ConsumerProtocol.deserializeConsumerProtocolAssignment(
                    ByteBuffer.wrap(classicGroupMember.assignment())
                ),
                topicsImage
            );
            ConsumerProtocolSubscription subscription = ConsumerProtocol.deserializeConsumerProtocolSubscription(
                ByteBuffer.wrap(classicGroupMember.metadata(classicGroup.protocolName().get()))
            );

            // The target assignment and the assigned partitions of each member are set based on the last
            // assignment of the classic group. All the members are put in the Stable state. If the classic
            // group was in Preparing Rebalance or Completing Rebalance states, the classic members are
            // asked to rejoin the group to re-trigger a rebalance or collect their assignments.
            ConsumerGroupMember newMember = new ConsumerGroupMember.Builder(classicGroupMember.memberId())
                .setMemberEpoch(classicGroup.generationId())
                .setState(MemberState.STABLE)
                .setPreviousMemberEpoch(classicGroup.generationId())
                .setInstanceId(classicGroupMember.groupInstanceId().orElse(null))
                .setRackId(toOptional(subscription.rackId()).orElse(null))
                .setRebalanceTimeoutMs(classicGroupMember.rebalanceTimeoutMs())
                .setClientId(classicGroupMember.clientId())
                .setClientHost(classicGroupMember.clientHost())
                .setSubscribedTopicNames(subscription.topics())
                .setAssignedPartitions(assignedPartitions)
                .setClassicMemberMetadata(
                    new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
                        .setSessionTimeoutMs(classicGroupMember.sessionTimeoutMs())
                        .setSupportedProtocols(ConsumerGroupMember.classicProtocolListFromJoinRequestProtocolCollection(
                            classicGroupMember.supportedProtocols()
                        ))
                )
                .build();
            consumerGroup.updateTargetAssignment(newMember.memberId(), new Assignment(assignedPartitions));
            consumerGroup.updateMember(newMember);
        });

        return consumerGroup;
    }

    /**
     * Populate the record list with the records needed to create the given consumer group.
     *
     * @param records The list to which the new records are added.
     */
    public void createConsumerGroupRecords(
        List<CoordinatorRecord> records
    ) {
        members().forEach((__, consumerGroupMember) ->
            records.add(CoordinatorRecordHelpers.newMemberSubscriptionRecord(groupId(), consumerGroupMember))
        );

        records.add(CoordinatorRecordHelpers.newGroupEpochRecord(groupId(), groupEpoch()));

        members().forEach((consumerGroupMemberId, consumerGroupMember) ->
            records.add(CoordinatorRecordHelpers.newTargetAssignmentRecord(
                groupId(),
                consumerGroupMemberId,
                targetAssignment(consumerGroupMemberId).partitions()
            ))
        );

        records.add(CoordinatorRecordHelpers.newTargetAssignmentEpochRecord(groupId(), groupEpoch()));

        members().forEach((__, consumerGroupMember) ->
            records.add(CoordinatorRecordHelpers.newCurrentAssignmentRecord(groupId(), consumerGroupMember))
        );
    }

    /**
     * Checks whether at least one of the given protocols can be supported. A
     * protocol can be supported if it is supported by all members that use the
     * classic protocol.
     *
     * @param memberProtocolType  The member protocol type.
     * @param memberProtocols     The set of protocol names.
     *
     * @return A boolean based on the condition mentioned above.
     */
    public boolean supportsClassicProtocols(String memberProtocolType, Set<String> memberProtocols) {
        if (ConsumerProtocol.PROTOCOL_TYPE.equals(memberProtocolType)) {
            if (isEmpty()) {
                return !memberProtocols.isEmpty();
            } else {
                return memberProtocols.stream().anyMatch(
                    name -> classicProtocolMembersSupportedProtocols.getOrDefault(name, 0) == numClassicProtocolMembers()
                );
            }
        }
        return false;
    }

    /**
     * Checks whether all the members use the classic protocol except the given member.
     *
     * @param memberId The member to remove.
     * @return A boolean indicating whether all the members use the classic protocol.
     */
    public boolean allMembersUseClassicProtocolExcept(String memberId) {
        return numClassicProtocolMembers() == members().size() - 1 &&
            !getOrMaybeCreateMember(memberId, false).useClassicProtocol();
    }

    /**
     * Checks whether the member has any unreleased partition.
     *
     * @param member The member to check.
     * @return A boolean indicating whether the member has partitions in the target
     *         assignment that hasn't been revoked by other members.
     */
    public boolean waitingOnUnreleasedPartition(ConsumerGroupMember member) {
        if (member.state() == MemberState.UNRELEASED_PARTITIONS) {
            for (Map.Entry<Uuid, Set<Integer>> entry : targetAssignment().get(member.memberId()).partitions().entrySet()) {
                Uuid topicId = entry.getKey();
                Set<Integer> assignedPartitions = member.assignedPartitions().getOrDefault(topicId, Collections.emptySet());

                for (int partition : entry.getValue()) {
                    if (!assignedPartitions.contains(partition) && currentPartitionEpoch(topicId, partition) != -1) {
                        return true;
                    }
                }
            }
        }
        return false;
    }
}
