KAFKA-16554: Online downgrade triggering and group type conversion

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

@@ -16,6 +16,7 @@
  */
 package org.apache.kafka.coordinator.group;
 
+import org.apache.kafka.clients.consumer.internals.ConsumerProtocol;
 import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.errors.ApiException;
@@ -777,6 +778,59 @@ public ClassicGroup classicGroup(
         }
     }
 
+    public boolean validateOnlineDowngrade(ConsumerGroup consumerGroup, String memberId) {
+        if (!consumerGroupMigrationPolicy.isDowngradeEnabled()) {
+            log.info("Cannot downgrade consumer group {} to classic group because the online downgrade is disabled.",
+                consumerGroup.groupId());
+            return false;
+        } else if (!consumerGroup.allMembersUseClassicProtocolExcept(memberId)) {
+            log.debug("Cannot downgrade consumer group {} to classic group because not all its members use the classic protocol.",
+                consumerGroup.groupId());
+            return false;
+        } else if (consumerGroup.numMembers() <= 1) {
+            log.info("Skip downgrading the consumer group {} to classic group because it's empty.",
+                consumerGroup.groupId());
+            return false;
+        } else if (consumerGroup.numMembers() - 1 > classicGroupMaxSize) {
+            log.info("Cannot downgrade consumer group {} to classic group because its group size is greater than classic group max size.",
+                consumerGroup.groupId());
+        }
+        return true;
+    }
+
+    public CompletableFuture<Void> convertToClassicGroup(ConsumerGroup consumerGroup, String leavingMemberId, List<Record> records) {
+        consumerGroup.createGroupTombstoneRecords(records);
+        ClassicGroup classicGroup;
+        try {
+            classicGroup = consumerGroup.toClassicGroup(
+                leavingMemberId,
+                logContext,
+                time,
+                consumerGroupSessionTimeoutMs,
+                metadataImage,
+                records
+            );
+        } catch (SchemaException e) {
+            log.warn("Cannot downgrade the consumer group " + consumerGroup.groupId() + ": fail to parse " +
+                "the Consumer Protocol " + ConsumerProtocol.PROTOCOL_TYPE + ".", e);
+
+            throw new GroupIdNotFoundException(String.format("Cannot downgrade the classic group %s: %s.",
+                consumerGroup.groupId(), e.getMessage()));
+        }
+
+        groups.put(consumerGroup.groupId(), classicGroup);
+        metrics.onClassicGroupStateTransition(null, classicGroup.currentState());
+
+        CompletableFuture<Void> appendFuture = new CompletableFuture<>();
+        appendFuture.whenComplete((__, t) -> {
+            if (t == null) {
+                classicGroup.allMembers().forEach(member -> rescheduleClassicGroupMemberHeartbeat(classicGroup, member));
+                prepareRebalance(classicGroup, String.format("Downgrade group %s.", classicGroup.groupId()));
+            }
+        });
+        return appendFuture;
+    }
+
     /**
      * Validates the online upgrade if the Classic Group receives a ConsumerGroupHeartbeat request.
      *
@@ -1422,10 +1476,12 @@ private CoordinatorResult<ConsumerGroupHeartbeatResponseData, Record> consumerGr
     ) throws ApiException {
         ConsumerGroup group = consumerGroup(groupId);
         List<Record> records;
+        CompletableFuture<Void> appendFuture = null;
         if (instanceId == null) {
             ConsumerGroupMember member = group.getOrMaybeCreateMember(memberId, false);
             log.info("[GroupId {}] Member {} left the consumer group.", groupId, memberId);
             records = consumerGroupFenceMember(group, member);
+            if (validateOnlineDowngrade(group, memberId)) appendFuture = convertToClassicGroup(group, memberId, records);
         } else {
             ConsumerGroupMember member = group.staticMember(instanceId);
             throwIfStaticMemberIsUnknown(member, instanceId);
@@ -1438,11 +1494,16 @@ private CoordinatorResult<ConsumerGroupHeartbeatResponseData, Record> consumerGr
                 log.info("[GroupId {}] Static Member {} with instance id {} left the consumer group.",
                     group.groupId(), memberId, instanceId);
                 records = consumerGroupFenceMember(group, member);
+                if (validateOnlineDowngrade(group, memberId)) appendFuture = convertToClassicGroup(group, memberId, records);
             }
         }
-        return new CoordinatorResult<>(records, new ConsumerGroupHeartbeatResponseData()
-            .setMemberId(memberId)
-            .setMemberEpoch(memberEpoch));
+        return new CoordinatorResult<>(
+            records,
+            new ConsumerGroupHeartbeatResponseData()
+                .setMemberId(memberId)
+                .setMemberEpoch(memberEpoch),
+            appendFuture
+        );
     }
 
     /**
@@ -1549,7 +1610,10 @@ private void scheduleConsumerGroupSessionTimeout(
                 ConsumerGroupMember member = group.getOrMaybeCreateMember(memberId, false);
                 log.info("[GroupId {}] Member {} fenced from the group because its session expired.",
                     groupId, memberId);
-                return new CoordinatorResult<>(consumerGroupFenceMember(group, member));
+                CompletableFuture<Void> appendFuture = null;
+                List<Record> records = consumerGroupFenceMember(group, member);
+                if (validateOnlineDowngrade(group, memberId)) appendFuture = convertToClassicGroup(group, memberId, records);
+                return new CoordinatorResult<>(records, appendFuture);
             } catch (GroupIdNotFoundException ex) {
                 log.debug("[GroupId {}] Could not fence {} because the group does not exist.",
                     groupId, memberId);
@@ -1599,7 +1663,10 @@ private void scheduleConsumerGroupRebalanceTimeout(
                     log.info("[GroupId {}] Member {} fenced from the group because " +
                             "it failed to transition from epoch {} within {}ms.",
                         groupId, memberId, memberEpoch, rebalanceTimeoutMs);
-                    return new CoordinatorResult<>(consumerGroupFenceMember(group, member));
+                    CompletableFuture<Void> appendFuture = null;
+                    List<Record> records = consumerGroupFenceMember(group, member);
+                    if (validateOnlineDowngrade(group, memberId)) appendFuture = convertToClassicGroup(group, memberId, records);
+                    return new CoordinatorResult<>(records, appendFuture);
                 } else {
                     log.debug("[GroupId {}] Ignoring rebalance timeout for {} because the member " +
                         "left the epoch {}.", groupId, memberId, memberEpoch);

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java

@@ -26,14 +26,20 @@
 import org.apache.kafka.common.message.ConsumerGroupDescribeResponseData;
 import org.apache.kafka.common.message.ListGroupsResponseData;
 import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.Utils;
 import org.apache.kafka.coordinator.group.Group;
 import org.apache.kafka.coordinator.group.OffsetExpirationCondition;
 import org.apache.kafka.coordinator.group.OffsetExpirationConditionImpl;
 import org.apache.kafka.coordinator.group.Record;
 import org.apache.kafka.coordinator.group.RecordHelpers;
 import org.apache.kafka.coordinator.group.classic.ClassicGroup;
+import org.apache.kafka.coordinator.group.classic.ClassicGroupMember;
+import org.apache.kafka.coordinator.group.classic.ClassicGroupState;
 import org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetricsShard;
 import org.apache.kafka.image.ClusterImage;
+import org.apache.kafka.image.MetadataImage;
 import org.apache.kafka.image.TopicImage;
 import org.apache.kafka.image.TopicsImage;
 import org.apache.kafka.timeline.SnapshotRegistry;
@@ -42,6 +48,7 @@
 import org.apache.kafka.timeline.TimelineObject;
 
 import java.nio.ByteBuffer;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -1190,6 +1197,101 @@ private static Map<Uuid, Set<Integer>> topicPartitionMapFromList(
         return topicPartitionMap;
     }
 
+    /**
+     * Create a corresponding ClassicGroup and append the record for the creation for group downgrade.
+     * The member with leavingMemberId will not be converted to the new ClassicGroup as it's the last
+     * member using new consumer protocol that left and triggered the downgrade.
+     *
+     * @param leavingMemberId               The member that will not be converted in the ClassicGroup.
+     * @param logContext                    The logContext to create the ClassicGroup.
+     * @param time                          The time to create the ClassicGroup.
+     * @param consumerGroupSessionTimeoutMs The consumerGroupSessionTimeoutMs.
+     * @param metadataImage                 The metadataImage.
+     * @param records                       The record list.
+     * @return  The created ClassicGroup.
+     */
+    public ClassicGroup toClassicGroup(
+        String leavingMemberId,
+        LogContext logContext,
+        Time time,
+        int consumerGroupSessionTimeoutMs,
+        MetadataImage metadataImage,
+        List<Record> records
+    ) {
+        ClassicGroup classicGroup = new ClassicGroup(
+            logContext,
+            groupId(),
+            ClassicGroupState.STABLE,
+            time,
+            metrics,
+            groupEpoch(),
+            Optional.ofNullable(ConsumerProtocol.PROTOCOL_TYPE),
+            Optional.empty(),
+            members().keySet().stream().filter(member -> !member.equals(leavingMemberId)).findAny(),
+            Optional.of(time.milliseconds())
+        );
+
+        members().forEach((memberId, member) -> {
+            if (!memberId.equals(leavingMemberId)) {
+                classicGroup.add(
+                    new ClassicGroupMember(
+                        memberId,
+                        Optional.ofNullable(member.instanceId()),
+                        member.clientId(),
+                        member.clientHost(),
+                        member.rebalanceTimeoutMs(),
+                        consumerGroupSessionTimeoutMs,
+                        ConsumerProtocol.PROTOCOL_TYPE,
+                        member.supportedJoinGroupRequestProtocols(),
+                        null
+                    )
+                );
+            }
+        });
+
+        classicGroup.setProtocolName(Optional.of(classicGroup.selectProtocol()));
+        classicGroup.setSubscribedTopics(classicGroup.computeSubscribedTopics());
+
+        Map<String, byte[]> assignments = new HashMap<>();
+
+        classicGroup.allMembers().forEach(classicGroupMember -> {
+            byte[] assignment = Utils.toArray(ConsumerProtocol.serializeAssignment(
+                new ConsumerPartitionAssignor.Assignment(ConsumerGroup.topicPartitionListFromMap(
+                    targetAssignment().get(classicGroupMember.memberId()).partitions(),
+                    metadataImage.topics()
+                )),
+                ConsumerProtocol.deserializeVersion(
+                    ByteBuffer.wrap(classicGroupMember.metadata(classicGroup.protocolName().orElse("")))
+                )
+            ));
+
+            classicGroupMember.setAssignment(assignment);
+            assignments.put(classicGroupMember.memberId(), assignment);
+        });
+
+        records.add(RecordHelpers.newGroupMetadataRecord(
+            classicGroup, assignments, metadataImage.features().metadataVersion()));
+
+        return classicGroup;
+    }
+
+    /**
+     * @return The list of TopicPartition converted from the map of topic id and partition set.
+     */
+    private static List<TopicPartition> topicPartitionListFromMap(
+        Map<Uuid, Set<Integer>> topicPartitions,
+        TopicsImage topicsImage
+    ) {
+        List<TopicPartition> topicPartitionList = new ArrayList<>();
+        topicPartitions.forEach((topicId, partitionSet) -> {
+            TopicImage topicImage = topicsImage.getTopic(topicId);
+            if (topicImage != null) {
+                partitionSet.forEach(partition -> topicPartitionList.add(new TopicPartition(topicImage.name(), partition)));
+            }
+        });
+        return topicPartitionList;
+    }
+
     /**
      * 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
@@ -1212,9 +1314,12 @@ public boolean supportsClassicProtocols(String memberProtocolType, Set<String> m
     }
 
     /**
+     * @param memberId The member to remove.
+     *
      * @return A boolean indicating whether all the members use the classic protocol.
      */
-    public boolean allMembersUseClassicProtocol() {
-        return numClassicProtocolMembers() == members().size();
+    public boolean allMembersUseClassicProtocolExcept(String memberId) {
+        return numClassicProtocolMembers() == members().size() - 1 &&
+            !getOrMaybeCreateMember(memberId, false).useClassicProtocol();
     }
 }

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupMember.java

@@ -479,6 +479,22 @@ public Map<Uuid, Set<Integer>> partitionsPendingRevocation() {
         return partitionsPendingRevocation;
     }
 
+    /**
+     * @return The supported classic protocols converted to JoinGroupRequestProtocolCollection.
+     */
+    public JoinGroupRequestData.JoinGroupRequestProtocolCollection supportedJoinGroupRequestProtocols() {
+        JoinGroupRequestData.JoinGroupRequestProtocolCollection protocols =
+            new JoinGroupRequestData.JoinGroupRequestProtocolCollection();
+        supportedClassicProtocols().ifPresent(classicProtocols -> classicProtocols.forEach(protocol ->
+            protocols.add(
+                new JoinGroupRequestData.JoinGroupRequestProtocol()
+                    .setName(protocol.name())
+                    .setMetadata(protocol.metadata())
+            )
+        ));
+        return protocols;
+    }
+
     /**
      * @return The classicMemberMetadata if the consumer uses the classic protocol.
      */

group-coordinator/src/test/java/org/apache/kafka/coordinator/group/Assertions.java

@@ -16,19 +16,25 @@
  */
 package org.apache.kafka.coordinator.group;
 
+import org.apache.kafka.clients.consumer.ConsumerPartitionAssignor;
+import org.apache.kafka.clients.consumer.internals.ConsumerProtocol;
 import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.message.ConsumerGroupHeartbeatResponseData;
 import org.apache.kafka.coordinator.group.generated.ConsumerGroupCurrentMemberAssignmentValue;
 import org.apache.kafka.coordinator.group.generated.ConsumerGroupPartitionMetadataValue;
+import org.apache.kafka.coordinator.group.generated.GroupMetadataValue;
 import org.apache.kafka.server.common.ApiMessageAndVersion;
 import org.opentest4j.AssertionFailedError;
 
+import java.nio.ByteBuffer;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Collectors;
 
 import static org.junit.jupiter.api.AssertionFailureBuilder.assertionFailure;
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -198,6 +204,62 @@ private static void assertApiMessageAndVersionEquals(
                     }
                 }
             }
+        } else if (actual.message() instanceof GroupMetadataValue) {
+            GroupMetadataValue expectedValue = (GroupMetadataValue) expected.message();
+            GroupMetadataValue actualValue = (GroupMetadataValue) actual.message();
+
+            assertEquals(expectedValue.protocolType(), actualValue.protocolType());
+            assertEquals(expectedValue.generation(), actualValue.generation());
+            assertEquals(expectedValue.protocol(), actualValue.protocol());
+            assertEquals(expectedValue.leader(), actualValue.leader());
+            assertEquals(expectedValue.currentStateTimestamp(), actualValue.currentStateTimestamp());
+
+            Map<String, GroupMetadataValue.MemberMetadata> expectedMemberMap =
+                expectedValue.members().stream()
+                    .collect(Collectors.toMap(GroupMetadataValue.MemberMetadata::memberId, Function.identity()));
+            Map<String, GroupMetadataValue.MemberMetadata> actualMemberMap =
+                actualValue.members().stream()
+                    .collect(Collectors.toMap(GroupMetadataValue.MemberMetadata::memberId, Function.identity()));
+
+            if (expectedMemberMap.size() != actualMemberMap.size()) {
+                fail("Member metadata maps have different sizes");
+            } else {
+                expectedMemberMap.forEach((memberId, expectedMemberMetadata) -> {
+                    GroupMetadataValue.MemberMetadata actualMemberMetadata = actualMemberMap.get(memberId);
+                    if (actualMemberMetadata == null) {
+                        fail("Member metadata maps have different values.");
+                    }
+                    assertEquals(expectedMemberMetadata.groupInstanceId(), actualMemberMetadata.groupInstanceId());
+                    assertEquals(expectedMemberMetadata.clientId(), actualMemberMetadata.clientId());
+                    assertEquals(expectedMemberMetadata.clientHost(), actualMemberMetadata.clientHost());
+                    assertEquals(expectedMemberMetadata.rebalanceTimeout(), actualMemberMetadata.rebalanceTimeout());
+                    assertEquals(expectedMemberMetadata.sessionTimeout(), actualMemberMetadata.sessionTimeout());
+
+                    // If the subscriptions or the assignments don't match, deserialize them and do the comparison.
+                    if (!expectedMemberMetadata.subscription().equals(actualMemberMetadata.subscription())) {
+                        ConsumerPartitionAssignor.Subscription expectedSubscription =
+                            ConsumerProtocol.deserializeSubscription(ByteBuffer.wrap(expectedMemberMetadata.subscription()));
+                        ConsumerPartitionAssignor.Subscription actualSubscription =
+                            ConsumerProtocol.deserializeSubscription(ByteBuffer.wrap(actualMemberMetadata.subscription()));
+
+                        assertUnorderedListEquals(expectedSubscription.topics(), actualSubscription.topics());
+                        assertEquals(expectedSubscription.userData(), actualSubscription.userData());
+                        assertUnorderedListEquals(expectedSubscription.ownedPartitions(), actualSubscription.ownedPartitions());
+                        assertEquals(expectedSubscription.rackId(), actualSubscription.rackId());
+                        assertEquals(expectedSubscription.groupInstanceId(), actualSubscription.groupInstanceId());
+                        assertEquals(expectedSubscription.generationId(), actualSubscription.generationId());
+                    }
+                    if (!expectedMemberMetadata.assignment().equals(actualMemberMetadata.assignment())) {
+                        ConsumerPartitionAssignor.Assignment expectedAssignment =
+                            ConsumerProtocol.deserializeAssignment(ByteBuffer.wrap(expectedMemberMetadata.assignment()));
+                        ConsumerPartitionAssignor.Assignment actualAssignment =
+                            ConsumerProtocol.deserializeAssignment(ByteBuffer.wrap(actualMemberMetadata.assignment()));
+
+                        assertUnorderedListEquals(expectedAssignment.partitions(), actualAssignment.partitions());
+                        assertEquals(expectedAssignment.userData(), actualAssignment.userData());
+                    }
+                });
+            }
         } else {
             assertEquals(expected.message(), actual.message());
         }