You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by da...@apache.org on 2024/04/17 11:57:52 UTC
(kafka) branch trunk updated: KAFKA-16436; Online upgrade triggering and group type conversion (#15662)
This is an automated email from the ASF dual-hosted git repository.
dajac pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new a9f65a5d7ff KAFKA-16436; Online upgrade triggering and group type conversion (#15662)
a9f65a5d7ff is described below
commit a9f65a5d7ff86fb31d020c7c6bb4afa48521810f
Author: Dongnuo Lyu <13...@users.noreply.github.com>
AuthorDate: Wed Apr 17 07:57:44 2024 -0400
KAFKA-16436; Online upgrade triggering and group type conversion (#15662)
This patch introduces the conversion from a classic group to a consumer group when a member joins with the new consumer group protocol (epoch is 0) but only if the conversion is enabled.
Reviewers: David Jacot <dj...@confluent.io>
---
.../group/ConsumerGroupMigrationPolicy.java | 24 +-
.../coordinator/group/GroupCoordinatorShard.java | 2 +-
.../coordinator/group/GroupMetadataManager.java | 77 ++-
.../kafka/coordinator/group/RecordHelpers.java | 3 +-
.../coordinator/group/classic/ClassicGroup.java | 51 +-
.../coordinator/group/consumer/ConsumerGroup.java | 219 +++++++
.../group/consumer/ConsumerGroupMember.java | 65 +-
.../message/ConsumerGroupMemberMetadataValue.json | 17 +-
.../apache/kafka/coordinator/group/Assertions.java | 2 +-
.../group/GroupMetadataManagerTest.java | 708 +++++++++++++++++++++
.../group/GroupMetadataManagerTestContext.java | 7 +
.../kafka/coordinator/group/RecordHelpersTest.java | 11 +-
.../group/classic/ClassicGroupTest.java | 81 +++
.../group/consumer/ConsumerGroupMemberTest.java | 73 ++-
.../group/consumer/ConsumerGroupTest.java | 102 +++
15 files changed, 1417 insertions(+), 25 deletions(-)
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/ConsumerGroupMigrationPolicy.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/ConsumerGroupMigrationPolicy.java
index c78ded13847..759e862c0b1 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/ConsumerGroupMigrationPolicy.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/ConsumerGroupMigrationPolicy.java
@@ -25,21 +25,33 @@ import java.util.stream.Collectors;
public enum ConsumerGroupMigrationPolicy {
/** Both upgrade and downgrade are enabled.*/
- BIDIRECTIONAL("bidirectional"),
+ BIDIRECTIONAL("bidirectional", true, true),
/** Only upgrade is enabled.*/
- UPGRADE("upgrade"),
+ UPGRADE("upgrade", true, false),
/** Only downgrade is enabled.*/
- DOWNGRADE("downgrade"),
+ DOWNGRADE("downgrade", false, true),
/** Neither upgrade nor downgrade is enabled.*/
- DISABLED("disabled");
+ DISABLED("disabled", false, false);
private final String name;
+ private final boolean isUpgradeEnabled;
+ private final boolean isDowngradeEnabled;
- ConsumerGroupMigrationPolicy(String name) {
- this.name = name;
+ ConsumerGroupMigrationPolicy(String config, boolean isUpgradeEnabled, boolean isDowngradeEnabled) {
+ this.name = config;
+ this.isUpgradeEnabled = isUpgradeEnabled;
+ this.isDowngradeEnabled = isDowngradeEnabled;
+ }
+
+ public boolean isUpgradeEnabled() {
+ return isUpgradeEnabled;
+ }
+
+ public boolean isDowngradeEnabled() {
+ return isDowngradeEnabled;
}
@Override
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java
index f63b1e80ecf..3c45a4f573c 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java
@@ -189,7 +189,7 @@ public class GroupCoordinatorShard implements CoordinatorShard<Record> {
.withClassicGroupNewMemberJoinTimeoutMs(config.classicGroupNewMemberJoinTimeoutMs)
.withClassicGroupMinSessionTimeoutMs(config.classicGroupMinSessionTimeoutMs)
.withClassicGroupMaxSessionTimeoutMs(config.classicGroupMaxSessionTimeoutMs)
- .withGroupProtocolMigrationPolicy(config.consumerGroupMigrationPolicy)
+ .withConsumerGroupMigrationPolicy(config.consumerGroupMigrationPolicy)
.withGroupCoordinatorMetricsShard(metricsShard)
.build();
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
index 441c7b658de..54f34d5c6ae 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
@@ -48,6 +48,7 @@ import org.apache.kafka.common.message.ListGroupsResponseData;
import org.apache.kafka.common.message.SyncGroupRequestData;
import org.apache.kafka.common.message.SyncGroupResponseData;
import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.protocol.types.SchemaException;
import org.apache.kafka.common.requests.JoinGroupRequest;
import org.apache.kafka.common.requests.RequestContext;
import org.apache.kafka.common.utils.LogContext;
@@ -238,7 +239,7 @@ public class GroupMetadataManager {
return this;
}
- Builder withGroupProtocolMigrationPolicy(ConsumerGroupMigrationPolicy consumerGroupMigrationPolicy) {
+ Builder withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy consumerGroupMigrationPolicy) {
this.consumerGroupMigrationPolicy = consumerGroupMigrationPolicy;
return this;
}
@@ -629,10 +630,11 @@ public class GroupMetadataManager {
} else {
if (group.type() == CONSUMER) {
return (ConsumerGroup) group;
+ } else if (createIfNotExists && validateOnlineUpgrade((ClassicGroup) group)) {
+ return convertToConsumerGroup((ClassicGroup) group, records);
} else {
- // We don't support upgrading/downgrading between protocols at the moment so
- // we throw an exception if a group exists with the wrong type.
- throw new GroupIdNotFoundException(String.format("Group %s is not a consumer group.", groupId));
+ throw new GroupIdNotFoundException(String.format("Group %s is not a consumer group.",
+ groupId));
}
}
}
@@ -775,6 +777,73 @@ public class GroupMetadataManager {
}
}
+ /**
+ * Validates the online upgrade if the Classic Group receives a ConsumerGroupHeartbeat request.
+ *
+ * @param classicGroup A ClassicGroup.
+ * @return The boolean indicating whether it's valid to online upgrade the classic group.
+ */
+ private boolean validateOnlineUpgrade(ClassicGroup classicGroup) {
+ if (!consumerGroupMigrationPolicy.isUpgradeEnabled()) {
+ log.info("Cannot upgrade classic group {} to consumer group because the online upgrade is disabled.",
+ classicGroup.groupId());
+ return false;
+ } else if (!classicGroup.usesConsumerGroupProtocol()) {
+ log.info("Cannot upgrade classic group {} to consumer group because the group does not use the consumer embedded protocol.",
+ classicGroup.groupId());
+ return false;
+ } else if (classicGroup.size() > consumerGroupMaxSize) {
+ log.info("Cannot upgrade classic group {} to consumer group because the group size exceeds the consumer group maximum size.",
+ classicGroup.groupId());
+ return false;
+ }
+ return true;
+ }
+
+ /**
+ * Creates a ConsumerGroup corresponding to the given classic group.
+ *
+ * @param classicGroup The ClassicGroup to convert.
+ * @param records The list of Records.
+ * @return The created ConsumerGroup.
+ */
+ ConsumerGroup convertToConsumerGroup(ClassicGroup classicGroup, List<Record> records) {
+ // The upgrade is always triggered by a new member joining the classic group, which always results in
+ // updatedMember.subscribedTopicNames changing, the group epoch being bumped, and triggering a new rebalance.
+ // If the ClassicGroup is rebalancing, inform the awaiting consumers of another ongoing rebalance
+ // so that they will rejoin for the new rebalance.
+ classicGroup.completeAllJoinFutures(Errors.REBALANCE_IN_PROGRESS);
+ classicGroup.completeAllSyncFutures(Errors.REBALANCE_IN_PROGRESS);
+
+ classicGroup.createGroupTombstoneRecords(records);
+
+ ConsumerGroup consumerGroup;
+ try {
+ consumerGroup = ConsumerGroup.fromClassicGroup(
+ snapshotRegistry,
+ metrics,
+ classicGroup,
+ metadataImage.topics()
+ );
+ } catch (SchemaException e) {
+ log.warn("Cannot upgrade the classic group " + classicGroup.groupId() +
+ " to consumer group because the embedded consumer protocol is malformed: "
+ + e.getMessage() + ".", e);
+
+ throw new GroupIdNotFoundException("Cannot upgrade the classic group " + classicGroup.groupId() +
+ " to consumer group because the embedded consumer protocol is malformed.");
+ }
+ consumerGroup.createConsumerGroupRecords(records);
+
+ // Create the session timeouts for the new members. If the conversion fails, the group will remain a
+ // classic group, thus these timers will fail the group type check and do nothing.
+ consumerGroup.members().forEach((memberId, __) ->
+ scheduleConsumerGroupSessionTimeout(consumerGroup.groupId(), memberId)
+ );
+
+ return consumerGroup;
+ }
+
/**
* Removes the group.
*
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/RecordHelpers.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/RecordHelpers.java
index c2bd093d6a8..c9d141ea6f7 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/RecordHelpers.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/RecordHelpers.java
@@ -82,7 +82,8 @@ public class RecordHelpers {
.setSubscribedTopicNames(member.subscribedTopicNames())
.setSubscribedTopicRegex(member.subscribedTopicRegex())
.setServerAssignor(member.serverAssignorName().orElse(null))
- .setRebalanceTimeoutMs(member.rebalanceTimeoutMs()),
+ .setRebalanceTimeoutMs(member.rebalanceTimeoutMs())
+ .setClassicMemberMetadata(member.classicMemberMetadata().orElse(null)),
(short) 0
)
);
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/classic/ClassicGroup.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/classic/ClassicGroup.java
index 66ee538a177..6baaed32be9 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/classic/ClassicGroup.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/classic/ClassicGroup.java
@@ -365,6 +365,13 @@ public class ClassicGroup implements Group {
return currentStateTimestamp.orElse(-1L);
}
+ /**
+ * @return the current supportedProtocols.
+ */
+ public Map<String, Integer> supportedProtocols() {
+ return supportedProtocols;
+ }
+
/**
* Sets newMemberAdded.
*
@@ -383,6 +390,15 @@ public class ClassicGroup implements Group {
this.subscribedTopics = subscribedTopics;
}
+ /**
+ * Sets protocolName.
+ *
+ * @param protocolName the value to set.
+ */
+ public void setProtocolName(Optional<String> protocolName) {
+ this.protocolName = protocolName;
+ }
+
/**
* @return whether the group is using the consumer protocol.
*/
@@ -1225,6 +1241,22 @@ public class ClassicGroup implements Group {
return false;
}
+ /**
+ * Complete all the awaiting join future with the given error.
+ *
+ * @param error the error to complete the future with.
+ */
+ public void completeAllJoinFutures(
+ Errors error
+ ) {
+ members.forEach((memberId, member) -> completeJoinFuture(
+ member,
+ new JoinGroupResponseData()
+ .setMemberId(memberId)
+ .setErrorCode(error.code())
+ ));
+ }
+
/**
* Complete a member's sync future.
*
@@ -1244,17 +1276,32 @@ public class ClassicGroup implements Group {
return false;
}
+ /**
+ * Complete all the awaiting sync future with the give error.
+ *
+ * @param error the error to complete the future with.
+ */
+ public void completeAllSyncFutures(
+ Errors error
+ ) {
+ members.forEach((__, member) -> completeSyncFuture(
+ member,
+ new SyncGroupResponseData()
+ .setErrorCode(error.code())
+ ));
+ }
+
/**
* Initiate the next generation for the group.
*/
public void initNextGeneration() {
generationId++;
if (!members.isEmpty()) {
- protocolName = Optional.of(selectProtocol());
+ setProtocolName(Optional.of(selectProtocol()));
subscribedTopics = computeSubscribedTopics();
transitionTo(COMPLETING_REBALANCE);
} else {
- protocolName = Optional.empty();
+ setProtocolName(Optional.empty());
subscribedTopics = computeSubscribedTopics();
transitionTo(EMPTY);
}
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java
index 68851b5176c..7fabc657245 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java
@@ -16,7 +16,9 @@
*/
package org.apache.kafka.coordinator.group.consumer;
+import org.apache.kafka.clients.consumer.ConsumerPartitionAssignor;
import org.apache.kafka.clients.consumer.internals.ConsumerProtocol;
+import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.errors.ApiException;
import org.apache.kafka.common.errors.StaleMemberEpochException;
@@ -29,6 +31,7 @@ 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.metrics.GroupCoordinatorMetricsShard;
import org.apache.kafka.image.ClusterImage;
import org.apache.kafka.image.TopicImage;
@@ -38,6 +41,7 @@ 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.HashSet;
@@ -180,6 +184,16 @@ public class ConsumerGroup implements Group {
*/
private DeadlineAndEpoch metadataRefreshDeadline = DeadlineAndEpoch.EMPTY;
+ /**
+ * 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,
@@ -198,6 +212,8 @@ public class ConsumerGroup implements Group {
this.targetAssignment = new TimelineHashMap<>(snapshotRegistry, 0);
this.currentPartitionEpoch = new TimelineHashMap<>(snapshotRegistry, 0);
this.metrics = Objects.requireNonNull(metrics);
+ this.numClassicProtocolMembers = new TimelineInteger(snapshotRegistry);
+ this.classicProtocolMembersSupportedProtocols = new TimelineHashMap<>(snapshotRegistry, 0);
}
/**
@@ -290,6 +306,15 @@ public class ConsumerGroup implements Group {
maybeUpdateGroupState();
}
+ /**
+ * 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.
@@ -355,6 +380,8 @@ public class ConsumerGroup implements Group {
maybeUpdatePartitionEpoch(oldMember, newMember);
updateStaticMember(newMember);
maybeUpdateGroupState();
+ maybeUpdateNumClassicProtocolMembers(oldMember, newMember);
+ maybeUpdateClassicProtocolMembersSupportedProtocols(oldMember, newMember);
}
/**
@@ -380,6 +407,8 @@ public class ConsumerGroup implements Group {
maybeRemovePartitionEpoch(oldMember);
removeStaticMember(oldMember);
maybeUpdateGroupState();
+ maybeUpdateNumClassicProtocolMembers(oldMember, null);
+ maybeUpdateClassicProtocolMembersSupportedProtocols(oldMember, null);
}
/**
@@ -411,6 +440,20 @@ public class ConsumerGroup implements Group {
return members.size();
}
+ /**
+ * @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 members keyed by their id.
*/
@@ -827,6 +870,52 @@ public class ConsumerGroup implements Group {
}
}
+ /**
+ * 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(), ConsumerGroup::decValue)
+ )
+ );
+ }
+ if (newMember != null) {
+ newMember.supportedClassicProtocols().ifPresent(protocols ->
+ protocols.forEach(protocol ->
+ classicProtocolMembersSupportedProtocols.compute(protocol.name(), ConsumerGroup::incValue)
+ )
+ );
+ }
+ }
+
/**
* Updates the subscribed topic names count.
*
@@ -998,4 +1087,134 @@ public class ConsumerGroup implements Group {
);
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 -> {
+ ConsumerPartitionAssignor.Assignment assignment = ConsumerProtocol.deserializeAssignment(
+ ByteBuffer.wrap(classicGroupMember.assignment())
+ );
+ Map<Uuid, Set<Integer>> partitions = topicPartitionMapFromList(assignment.partitions(), topicsImage);
+
+ ConsumerPartitionAssignor.Subscription subscription = ConsumerProtocol.deserializeSubscription(
+ 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(subscription.rackId().orElse(null))
+ .setRebalanceTimeoutMs(classicGroupMember.rebalanceTimeoutMs())
+ .setClientId(classicGroupMember.clientId())
+ .setClientHost(classicGroupMember.clientHost())
+ .setSubscribedTopicNames(subscription.topics())
+ .setAssignedPartitions(partitions)
+ .setSupportedClassicProtocols(classicGroupMember.supportedProtocols())
+ .build();
+ consumerGroup.updateTargetAssignment(newMember.memberId(), new Assignment(partitions));
+ 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<Record> records
+ ) {
+ members().forEach((__, consumerGroupMember) ->
+ records.add(RecordHelpers.newMemberSubscriptionRecord(groupId(), consumerGroupMember))
+ );
+
+ records.add(RecordHelpers.newGroupEpochRecord(groupId(), groupEpoch()));
+
+ members().forEach((consumerGroupMemberId, consumerGroupMember) ->
+ records.add(RecordHelpers.newTargetAssignmentRecord(
+ groupId(),
+ consumerGroupMemberId,
+ targetAssignment(consumerGroupMemberId).partitions()
+ ))
+ );
+
+ records.add(RecordHelpers.newTargetAssignmentEpochRecord(groupId(), groupEpoch()));
+
+ members().forEach((__, consumerGroupMember) ->
+ records.add(RecordHelpers.newCurrentAssignmentRecord(groupId(), consumerGroupMember))
+ );
+ }
+
+ /**
+ * @return The map of topic id and partition set converted from the list of TopicPartition.
+ */
+ private static Map<Uuid, Set<Integer>> topicPartitionMapFromList(
+ List<TopicPartition> partitions,
+ TopicsImage topicsImage
+ ) {
+ Map<Uuid, Set<Integer>> topicPartitionMap = new HashMap<>();
+ partitions.forEach(topicPartition -> {
+ TopicImage topicImage = topicsImage.getTopic(topicPartition.topic());
+ if (topicImage != null) {
+ topicPartitionMap
+ .computeIfAbsent(topicImage.id(), __ -> new HashSet<>())
+ .add(topicPartition.partition());
+ }
+ });
+ return topicPartitionMap;
+ }
+
+ /**
+ * 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 (isEmpty()) {
+ return !memberProtocolType.isEmpty() && !memberProtocols.isEmpty();
+ } else {
+ return ConsumerProtocol.PROTOCOL_TYPE.equals(memberProtocolType) &&
+ memberProtocols.stream().anyMatch(
+ name -> classicProtocolMembersSupportedProtocols.getOrDefault(name, 0) == numClassicProtocolMembers()
+ );
+ }
+ }
+
+ /**
+ * @return A boolean indicating whether all the members use the classic protocol.
+ */
+ public boolean allMembersUseClassicProtocol() {
+ return numClassicProtocolMembers() == members().size();
+ }
}
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupMember.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupMember.java
index 3bf87caef9b..bb1fab0006f 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupMember.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupMember.java
@@ -18,6 +18,7 @@ package org.apache.kafka.coordinator.group.consumer;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.message.ConsumerGroupDescribeResponseData;
+import org.apache.kafka.common.message.JoinGroupRequestData;
import org.apache.kafka.coordinator.group.generated.ConsumerGroupCurrentMemberAssignmentValue;
import org.apache.kafka.coordinator.group.generated.ConsumerGroupMemberMetadataValue;
import org.apache.kafka.image.TopicImage;
@@ -64,6 +65,7 @@ public class ConsumerGroupMember {
private String serverAssignorName = null;
private Map<Uuid, Set<Integer>> assignedPartitions = Collections.emptyMap();
private Map<Uuid, Set<Integer>> partitionsPendingRevocation = Collections.emptyMap();
+ private ConsumerGroupMemberMetadataValue.ClassicMemberMetadata classicMemberMetadata = null;
public Builder(String memberId) {
this.memberId = Objects.requireNonNull(memberId);
@@ -86,6 +88,7 @@ public class ConsumerGroupMember {
this.state = member.state;
this.assignedPartitions = member.assignedPartitions;
this.partitionsPendingRevocation = member.partitionsPendingRevocation;
+ this.classicMemberMetadata = member.classicMemberMetadata;
}
public Builder updateMemberEpoch(int memberEpoch) {
@@ -192,6 +195,25 @@ public class ConsumerGroupMember {
return this;
}
+ public Builder setClassicMemberMetadata(ConsumerGroupMemberMetadataValue.ClassicMemberMetadata classicMemberMetadata) {
+ this.classicMemberMetadata = classicMemberMetadata;
+ return this;
+ }
+
+ public Builder setSupportedClassicProtocols(JoinGroupRequestData.JoinGroupRequestProtocolCollection protocols) {
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> newSupportedProtocols = new ArrayList<>();
+ protocols.forEach(protocol ->
+ newSupportedProtocols.add(
+ new ConsumerGroupMemberMetadataValue.ClassicProtocol()
+ .setName(protocol.name())
+ .setMetadata(protocol.metadata())
+ )
+ );
+ this.classicMemberMetadata = new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(newSupportedProtocols);
+ return this;
+ }
+
public Builder updateWith(ConsumerGroupMemberMetadataValue record) {
setInstanceId(record.instanceId());
setRackId(record.rackId());
@@ -201,6 +223,7 @@ public class ConsumerGroupMember {
setSubscribedTopicRegex(record.subscribedTopicRegex());
setRebalanceTimeoutMs(record.rebalanceTimeoutMs());
setServerAssignorName(record.serverAssignor());
+ setClassicMemberMetadata(record.classicMemberMetadata());
return this;
}
@@ -236,7 +259,8 @@ public class ConsumerGroupMember {
serverAssignorName,
state,
assignedPartitions,
- partitionsPendingRevocation
+ partitionsPendingRevocation,
+ classicMemberMetadata
);
}
}
@@ -311,6 +335,11 @@ public class ConsumerGroupMember {
*/
private final Map<Uuid, Set<Integer>> partitionsPendingRevocation;
+ /**
+ * The classic member metadata if the consumer uses the classic protocol.
+ */
+ private final ConsumerGroupMemberMetadataValue.ClassicMemberMetadata classicMemberMetadata;
+
private ConsumerGroupMember(
String memberId,
int memberEpoch,
@@ -325,7 +354,8 @@ public class ConsumerGroupMember {
String serverAssignorName,
MemberState state,
Map<Uuid, Set<Integer>> assignedPartitions,
- Map<Uuid, Set<Integer>> partitionsPendingRevocation
+ Map<Uuid, Set<Integer>> partitionsPendingRevocation,
+ ConsumerGroupMemberMetadataValue.ClassicMemberMetadata classicMemberMetadata
) {
this.memberId = memberId;
this.memberEpoch = memberEpoch;
@@ -341,6 +371,7 @@ public class ConsumerGroupMember {
this.serverAssignorName = serverAssignorName;
this.assignedPartitions = assignedPartitions;
this.partitionsPendingRevocation = partitionsPendingRevocation;
+ this.classicMemberMetadata = classicMemberMetadata;
}
/**
@@ -448,6 +479,24 @@ public class ConsumerGroupMember {
return partitionsPendingRevocation;
}
+ /**
+ * @return The classicMemberMetadata if the consumer uses the classic protocol.
+ */
+ public Optional<ConsumerGroupMemberMetadataValue.ClassicMemberMetadata> classicMemberMetadata() {
+ return Optional.ofNullable(classicMemberMetadata);
+ }
+
+ /**
+ * @return The list of protocols if the consumer uses the classic protocol.
+ */
+ public Optional<List<ConsumerGroupMemberMetadataValue.ClassicProtocol>> supportedClassicProtocols() {
+ if (useClassicProtocol()) {
+ return Optional.ofNullable(classicMemberMetadata.supportedProtocols());
+ } else {
+ return Optional.empty();
+ }
+ }
+
/**
* @param targetAssignment The target assignment of this member in the corresponding group.
*
@@ -504,6 +553,13 @@ public class ConsumerGroupMember {
}
}
+ /**
+ * @return The boolean indicating whether the member uses the classic protocol.
+ */
+ public boolean useClassicProtocol() {
+ return classicMemberMetadata != null;
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) return true;
@@ -522,7 +578,8 @@ public class ConsumerGroupMember {
&& Objects.equals(subscribedTopicRegex, that.subscribedTopicRegex)
&& Objects.equals(serverAssignorName, that.serverAssignorName)
&& Objects.equals(assignedPartitions, that.assignedPartitions)
- && Objects.equals(partitionsPendingRevocation, that.partitionsPendingRevocation);
+ && Objects.equals(partitionsPendingRevocation, that.partitionsPendingRevocation)
+ && Objects.equals(classicMemberMetadata, that.classicMemberMetadata);
}
@Override
@@ -541,6 +598,7 @@ public class ConsumerGroupMember {
result = 31 * result + Objects.hashCode(serverAssignorName);
result = 31 * result + Objects.hashCode(assignedPartitions);
result = 31 * result + Objects.hashCode(partitionsPendingRevocation);
+ result = 31 * result + Objects.hashCode(classicMemberMetadata);
return result;
}
@@ -561,6 +619,7 @@ public class ConsumerGroupMember {
", serverAssignorName='" + serverAssignorName + '\'' +
", assignedPartitions=" + assignedPartitions +
", partitionsPendingRevocation=" + partitionsPendingRevocation +
+ ", classicMemberMetadata='" + classicMemberMetadata + '\'' +
')';
}
diff --git a/group-coordinator/src/main/resources/common/message/ConsumerGroupMemberMetadataValue.json b/group-coordinator/src/main/resources/common/message/ConsumerGroupMemberMetadataValue.json
index 4886b7ad9d3..bbefc68c4ea 100644
--- a/group-coordinator/src/main/resources/common/message/ConsumerGroupMemberMetadataValue.json
+++ b/group-coordinator/src/main/resources/common/message/ConsumerGroupMemberMetadataValue.json
@@ -35,6 +35,21 @@
{ "name": "RebalanceTimeoutMs", "type": "int32", "versions": "0+", "default": -1,
"about": "The rebalance timeout" },
{ "name": "ServerAssignor", "versions": "0+", "nullableVersions": "0+", "type": "string",
- "about": "The server assignor to use; or null if not used." }
+ "about": "The server assignor to use; or null if not used." },
+ { "name": "ClassicMemberMetadata", "versions": "0+", "nullableVersions": "0+", "type": "ClassicMemberMetadata",
+ "default": null, "taggedVersions": "0+", "tag": 0,
+ "about": "The classic member metadata which includes the supported protocols of the consumer if it uses the classic protocol. The metadata is null if the consumer uses the consumer protocol.",
+ "fields": [
+ { "name": "SupportedProtocols", "type": "[]ClassicProtocol", "versions": "0+",
+ "about": "The list of protocols that the member supports if the consumer uses the classic protocol.",
+ "fields": [
+ { "name": "Name", "type": "string", "versions": "0+",
+ "about": "The protocol name."},
+ { "name": "Metadata", "type": "bytes", "versions": "0+",
+ "about": "The protocol metadata."}
+ ]
+ }
+ ]
+ }
]
}
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/Assertions.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/Assertions.java
index b5f40240ebb..cf859d8a1a1 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/Assertions.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/Assertions.java
@@ -187,7 +187,7 @@ public class Assertions {
if (expectedPartitionMetadataList.size() != actualPartitionMetadataList.size()) {
fail("Partition metadata lists have different sizes");
} else if (!expectedPartitionMetadataList.isEmpty() && !actualPartitionMetadataList.isEmpty()) {
- for (int j = 0; j < expectedTopicMetadataList.size(); j++) {
+ for (int j = 0; j < expectedPartitionMetadataList.size(); j++) {
ConsumerGroupPartitionMetadataValue.PartitionMetadata expectedPartitionMetadata =
expectedPartitionMetadataList.get(j);
ConsumerGroupPartitionMetadataValue.PartitionMetadata actualPartitionMetadata =
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
index 81c582ed4d4..21fb536a43f 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
@@ -54,6 +54,7 @@ import org.apache.kafka.common.metadata.RemoveTopicRecord;
import org.apache.kafka.common.metadata.TopicRecord;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.coordinator.group.MockCoordinatorTimer.ExpiredTimeout;
import org.apache.kafka.coordinator.group.MockCoordinatorTimer.ScheduledTimeout;
import org.apache.kafka.coordinator.group.assignor.GroupAssignment;
@@ -84,6 +85,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -9522,6 +9524,712 @@ public class GroupMetadataManagerTest {
);
}
+ @Test
+ public void testConsumerGroupHeartbeatWithStableClassicGroup() {
+ String groupId = "group-id";
+ String memberId1 = "member-id-1";
+ String memberId2 = "member-id-2";
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+ Uuid barTopicId = Uuid.randomUuid();
+ String barTopicName = "bar";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+ assignor.prepareGroupAssignment(new GroupAssignment(
+ new HashMap<String, MemberAssignment>() {
+ {
+ put(memberId1, new MemberAssignment(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0)
+ )));
+ put(memberId2, new MemberAssignment(mkAssignment(
+ mkTopicAssignment(barTopicId, 0)
+ )));
+ }
+ }
+ ));
+
+ MetadataImage metadataImage = new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 1)
+ .addTopic(barTopicId, barTopicName, 1)
+ .addRacks()
+ .build();
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.UPGRADE)
+ .withAssignors(Collections.singletonList(assignor))
+ .withMetadataImage(metadataImage)
+ .build();
+
+ JoinGroupRequestData.JoinGroupRequestProtocolCollection protocols = new JoinGroupRequestData.JoinGroupRequestProtocolCollection(1);
+ protocols.add(new JoinGroupRequestData.JoinGroupRequestProtocol()
+ .setName("range")
+ .setMetadata(Utils.toArray(ConsumerProtocol.serializeSubscription(new ConsumerPartitionAssignor.Subscription(
+ Arrays.asList(fooTopicName, barTopicName),
+ null,
+ Arrays.asList(
+ new TopicPartition(fooTopicName, 0),
+ new TopicPartition(barTopicName, 0)
+ )
+ ))))
+ );
+
+ Map<String, byte[]> assignments = new HashMap<String, byte[]>() {
+ {
+ put(
+ memberId1,
+ Utils.toArray(ConsumerProtocol.serializeAssignment(new ConsumerPartitionAssignor.Assignment(Arrays.asList(
+ new TopicPartition(fooTopicName, 0),
+ new TopicPartition(barTopicName, 0)
+ ))))
+ );
+ }
+ };
+
+ // Create a stable classic group with member 1.
+ ClassicGroup group = context.createClassicGroup(groupId);
+ group.setProtocolName(Optional.ofNullable("range"));
+ group.add(
+ new ClassicGroupMember(
+ memberId1,
+ Optional.empty(),
+ "client-id",
+ "client-host",
+ 10000,
+ 5000,
+ "consumer",
+ protocols,
+ assignments.get(memberId1)
+ )
+ );
+
+ group.transitionTo(PREPARING_REBALANCE);
+ group.transitionTo(COMPLETING_REBALANCE);
+ group.transitionTo(STABLE);
+
+ context.replay(RecordHelpers.newGroupMetadataRecord(group, assignments, metadataImage.features().metadataVersion()));
+ context.commit();
+ group = context.groupMetadataManager.getOrMaybeCreateClassicGroup(groupId, false);
+
+ // A new member 2 with new protocol joins the classic group, triggering the upgrade.
+ CoordinatorResult<ConsumerGroupHeartbeatResponseData, Record> result = context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId2)
+ .setRebalanceTimeoutMs(5000)
+ .setServerAssignor("range")
+ .setSubscribedTopicNames(Arrays.asList(fooTopicName, barTopicName))
+ .setTopicPartitions(Collections.emptyList()));
+
+ ConsumerGroupMember expectedMember1 = new ConsumerGroupMember.Builder(memberId1)
+ .setMemberEpoch(0)
+ .setPreviousMemberEpoch(0)
+ .setClientId("client-id")
+ .setClientHost("client-host")
+ .setSubscribedTopicNames(Arrays.asList(fooTopicName, barTopicName))
+ .setRebalanceTimeoutMs(10000)
+ .setSupportedClassicProtocols(protocols)
+ .setAssignedPartitions(new HashMap<Uuid, Set<Integer>>() {
+ {
+ put(fooTopicId, new HashSet<>(Collections.singletonList(0)));
+ put(barTopicId, new HashSet<>(Collections.singletonList(0)));
+ }
+ })
+ .build();
+
+ ConsumerGroupMember expectedMember2 = new ConsumerGroupMember.Builder(memberId2)
+ .setMemberEpoch(1)
+ .setPreviousMemberEpoch(0)
+ .setState(MemberState.UNRELEASED_PARTITIONS)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ .setServerAssignorName("range")
+ .setSubscribedTopicNames(Arrays.asList(fooTopicName, barTopicName))
+ .setRebalanceTimeoutMs(5000)
+ .setAssignedPartitions(Collections.emptyMap())
+ .build();
+
+ List<Record> expectedRecords = Arrays.asList(
+ // The existing classic group tombstone.
+ RecordHelpers.newGroupMetadataTombstoneRecord(groupId),
+
+ // Create the new consumer group with member 1.
+ RecordHelpers.newMemberSubscriptionRecord(groupId, expectedMember1),
+ RecordHelpers.newGroupEpochRecord(groupId, 0),
+ RecordHelpers.newTargetAssignmentRecord(groupId, memberId1, expectedMember1.assignedPartitions()),
+ RecordHelpers.newTargetAssignmentEpochRecord(groupId, 0),
+ RecordHelpers.newCurrentAssignmentRecord(groupId, expectedMember1),
+
+ // Member 2 joins the new consumer group.
+ RecordHelpers.newMemberSubscriptionRecord(groupId, expectedMember2),
+
+ // The subscription metadata hasn't been updated during the conversion, so a new one is computed.
+ RecordHelpers.newGroupSubscriptionMetadataRecord(groupId, new HashMap<String, TopicMetadata>() {
+ {
+ put(fooTopicName, new TopicMetadata(fooTopicId, fooTopicName, 1, new HashMap<Integer, Set<String>>() {
+ {
+ put(0, new HashSet<>(Arrays.asList("rack0", "rack1")));
+ }
+ }));
+ put(barTopicName, new TopicMetadata(barTopicId, barTopicName, 1, new HashMap<Integer, Set<String>>() {
+ {
+ put(0, new HashSet<>(Arrays.asList("rack0", "rack1")));
+ }
+ }));
+ }
+ }),
+
+ // Newly joining member 2 bumps the group epoch. A new target assignment is computed.
+ RecordHelpers.newGroupEpochRecord(groupId, 1),
+ RecordHelpers.newTargetAssignmentRecord(groupId, memberId2, new HashMap<Uuid, Set<Integer>>() {
+ {
+ put(barTopicId, new HashSet<>(Collections.singletonList(0)));
+ }
+ }),
+ RecordHelpers.newTargetAssignmentRecord(groupId, memberId1, new HashMap<Uuid, Set<Integer>>() {
+ {
+ put(fooTopicId, new HashSet<>(Collections.singletonList(0)));
+ }
+ }),
+ RecordHelpers.newTargetAssignmentEpochRecord(groupId, 1),
+
+ // Member 2 has no pending revoking partition. Bump its member epoch and transition to UNRELEASED_PARTITIONS.
+ RecordHelpers.newCurrentAssignmentRecord(groupId, expectedMember2)
+ );
+
+ assertRecordsEquals(expectedRecords, result.records());
+
+ context.assertSessionTimeout(groupId, memberId1, 45000);
+ context.assertSessionTimeout(groupId, memberId2, 45000);
+
+ // Simulate a failed replay. The context is rolled back and the group is converted back to the classic group.
+ context.rollback();
+ assertEquals(group, context.groupMetadataManager.getOrMaybeCreateClassicGroup("group-id", false));
+ }
+
+ @Test
+ public void testConsumerGroupHeartbeatWithPreparingRebalanceClassicGroup() throws Exception {
+ String groupId = "group-id";
+ String memberId1 = "member-id-1";
+ String memberId2 = "member-id-2";
+ String memberId3 = "member-id-3";
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+ Uuid barTopicId = Uuid.randomUuid();
+ String barTopicName = "bar";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+ assignor.prepareGroupAssignment(new GroupAssignment(
+ new HashMap<String, MemberAssignment>() {
+ {
+ put(memberId1, new MemberAssignment(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0)
+ )));
+ put(memberId2, new MemberAssignment(mkAssignment(
+ mkTopicAssignment(barTopicId, 0)
+ )));
+ put(memberId3, new MemberAssignment(mkAssignment(
+ mkTopicAssignment(fooTopicId, 1)
+ )));
+ }
+ }
+ ));
+
+ MetadataImage metadataImage = new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 2)
+ .addTopic(barTopicId, barTopicName, 1)
+ .addRacks()
+ .build();
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.UPGRADE)
+ .withAssignors(Collections.singletonList(assignor))
+ .withMetadataImage(metadataImage)
+ .build();
+
+ JoinGroupRequestData.JoinGroupRequestProtocolCollection protocols1 = new JoinGroupRequestData.JoinGroupRequestProtocolCollection(1);
+ protocols1.add(new JoinGroupRequestData.JoinGroupRequestProtocol()
+ .setName("range")
+ .setMetadata(Utils.toArray(ConsumerProtocol.serializeSubscription(new ConsumerPartitionAssignor.Subscription(
+ Arrays.asList(fooTopicName, barTopicName),
+ null,
+ Arrays.asList(
+ new TopicPartition(fooTopicName, 0),
+ new TopicPartition(fooTopicName, 1)
+ )
+ ))))
+ );
+
+ JoinGroupRequestData.JoinGroupRequestProtocolCollection protocols2 = new JoinGroupRequestData.JoinGroupRequestProtocolCollection(1);
+ protocols2.add(new JoinGroupRequestData.JoinGroupRequestProtocol()
+ .setName("range")
+ .setMetadata(Utils.toArray(ConsumerProtocol.serializeSubscription(new ConsumerPartitionAssignor.Subscription(
+ Arrays.asList(fooTopicName, barTopicName),
+ null,
+ Arrays.asList(
+ new TopicPartition(barTopicName, 0)
+ )
+ ))))
+ );
+
+ Map<String, byte[]> assignments = new HashMap<String, byte[]>() {
+ {
+ put(
+ memberId1,
+ Utils.toArray(ConsumerProtocol.serializeAssignment(new ConsumerPartitionAssignor.Assignment(Arrays.asList(
+ new TopicPartition(fooTopicName, 0),
+ new TopicPartition(fooTopicName, 1)
+ ))))
+ );
+ put(
+ memberId2,
+ Utils.toArray(ConsumerProtocol.serializeAssignment(new ConsumerPartitionAssignor.Assignment(Arrays.asList(
+ new TopicPartition(barTopicName, 0)
+ ))))
+ );
+ }
+ };
+
+ // Construct a stable group with two members.
+ ClassicGroup group = context.createClassicGroup(groupId);
+ group.setProtocolName(Optional.ofNullable("range"));
+ group.add(
+ new ClassicGroupMember(
+ memberId1,
+ Optional.empty(),
+ "client-id",
+ "client-host",
+ 10000,
+ 5000,
+ "consumer",
+ protocols1,
+ assignments.get(memberId1)
+ )
+ );
+ group.add(
+ new ClassicGroupMember(
+ memberId2,
+ Optional.empty(),
+ "client-id",
+ "client-host",
+ 10000,
+ 5000,
+ "consumer",
+ protocols2,
+ assignments.get(memberId2)
+ )
+ );
+
+ group.transitionTo(PREPARING_REBALANCE);
+ group.transitionTo(COMPLETING_REBALANCE);
+ group.transitionTo(STABLE);
+
+ context.replay(RecordHelpers.newGroupMetadataRecord(group, assignments, metadataImage.features().metadataVersion()));
+ context.commit();
+ group = context.groupMetadataManager.getOrMaybeCreateClassicGroup(groupId, false);
+
+ // The leader rejoins, triggering a rebalance.
+ GroupMetadataManagerTestContext.JoinResult joinResult = context.sendClassicGroupJoin(
+ new GroupMetadataManagerTestContext.JoinGroupRequestBuilder()
+ .withGroupId("group-id")
+ .withMemberId(memberId1)
+ .withProtocols(protocols1)
+ .withSessionTimeoutMs(5000)
+ .withRebalanceTimeoutMs(10000)
+ .build()
+ );
+ assertTrue(group.isInState(PREPARING_REBALANCE));
+
+ // Another new member 3 joins with new protocol, triggering the upgrade.
+ CoordinatorResult<ConsumerGroupHeartbeatResponseData, Record> consumerGroupHeartbeatResult = context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId3)
+ .setRebalanceTimeoutMs(5000)
+ .setServerAssignor("range")
+ .setSubscribedTopicNames(Arrays.asList(fooTopicName, barTopicName))
+ .setTopicPartitions(Collections.emptyList()));
+
+ ConsumerGroupMember expectedMember1 = new ConsumerGroupMember.Builder(memberId1)
+ .setMemberEpoch(0)
+ .setPreviousMemberEpoch(0)
+ .setClientId("client-id")
+ .setClientHost("client-host")
+ .setSubscribedTopicNames(Arrays.asList(fooTopicName, barTopicName))
+ .setRebalanceTimeoutMs(10000)
+ .setSupportedClassicProtocols(protocols1)
+ .setAssignedPartitions(new HashMap<Uuid, Set<Integer>>() {
+ {
+ put(fooTopicId, new HashSet<>(Arrays.asList(0, 1)));
+ }
+ })
+ .build();
+
+ ConsumerGroupMember expectedMember2 = new ConsumerGroupMember.Builder(memberId2)
+ .setMemberEpoch(0)
+ .setPreviousMemberEpoch(0)
+ .setClientId("client-id")
+ .setClientHost("client-host")
+ .setSubscribedTopicNames(Arrays.asList(fooTopicName, barTopicName))
+ .setRebalanceTimeoutMs(10000)
+ .setSupportedClassicProtocols(protocols2)
+ .setAssignedPartitions(new HashMap<Uuid, Set<Integer>>() {
+ {
+ put(barTopicId, new HashSet<>(Collections.singletonList(0)));
+ }
+ })
+ .build();
+
+ ConsumerGroupMember expectedMember3 = new ConsumerGroupMember.Builder(memberId3)
+ .setMemberEpoch(1)
+ .setPreviousMemberEpoch(0)
+ .setState(MemberState.UNRELEASED_PARTITIONS)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ .setServerAssignorName("range")
+ .setSubscribedTopicNames(Arrays.asList(fooTopicName, barTopicName))
+ .setRebalanceTimeoutMs(5000)
+ .setAssignedPartitions(Collections.emptyMap())
+ .build();
+
+ List<Record> expectedRecords = Arrays.asList(
+ // The existing classic group tombstone.
+ RecordHelpers.newGroupMetadataTombstoneRecord(groupId),
+
+ // Create the new consumer group with member 1 and member 2.
+ RecordHelpers.newMemberSubscriptionRecord(groupId, expectedMember1),
+ RecordHelpers.newMemberSubscriptionRecord(groupId, expectedMember2),
+
+ RecordHelpers.newGroupEpochRecord(groupId, 0),
+ RecordHelpers.newTargetAssignmentRecord(groupId, memberId1, expectedMember1.assignedPartitions()),
+ RecordHelpers.newTargetAssignmentRecord(groupId, memberId2, expectedMember2.assignedPartitions()),
+
+ RecordHelpers.newTargetAssignmentEpochRecord(groupId, 0),
+
+ RecordHelpers.newCurrentAssignmentRecord(groupId, expectedMember1),
+ RecordHelpers.newCurrentAssignmentRecord(groupId, expectedMember2),
+
+ // Member 3 joins the new consumer group.
+ RecordHelpers.newMemberSubscriptionRecord(groupId, expectedMember3),
+
+ // The subscription metadata hasn't been updated during the conversion, so a new one is computed.
+ RecordHelpers.newGroupSubscriptionMetadataRecord(groupId, new HashMap<String, TopicMetadata>() {
+ {
+ put(fooTopicName, new TopicMetadata(fooTopicId, fooTopicName, 2, new HashMap<Integer, Set<String>>() {
+ {
+ put(0, new HashSet<>(Arrays.asList("rack0", "rack1")));
+ put(1, new HashSet<>(Arrays.asList("rack1", "rack2")));
+ }
+ }));
+ put(barTopicName, new TopicMetadata(barTopicId, barTopicName, 1, new HashMap<Integer, Set<String>>() {
+ {
+ put(0, new HashSet<>(Arrays.asList("rack0", "rack1")));
+ }
+ }));
+ }
+ }),
+
+ // Newly joining member 3 bumps the group epoch. A new target assignment is computed.
+ RecordHelpers.newGroupEpochRecord(groupId, 1),
+ RecordHelpers.newTargetAssignmentRecord(groupId, memberId1, new HashMap<Uuid, Set<Integer>>() {
+ {
+ put(fooTopicId, new HashSet<>(Collections.singletonList(0)));
+ }
+ }),
+ RecordHelpers.newTargetAssignmentRecord(groupId, memberId3, new HashMap<Uuid, Set<Integer>>() {
+ {
+ put(fooTopicId, new HashSet<>(Collections.singletonList(1)));
+ }
+ }),
+ RecordHelpers.newTargetAssignmentEpochRecord(groupId, 1),
+
+ // Member 3 has no pending revoking partition. Bump its member epoch and transition to UNRELEASED_PARTITIONS.
+ RecordHelpers.newCurrentAssignmentRecord(groupId, expectedMember3)
+ );
+
+ assertRecordsEquals(expectedRecords, consumerGroupHeartbeatResult.records());
+ assertTrue(joinResult.joinFuture.isDone());
+ assertEquals(Errors.REBALANCE_IN_PROGRESS.code(), joinResult.joinFuture.get().errorCode());
+
+ context.assertSessionTimeout(groupId, memberId1, 45000);
+ context.assertSessionTimeout(groupId, memberId2, 45000);
+ context.assertSessionTimeout(groupId, memberId3, 45000);
+
+ // Simulate a failed replay. The context is rolled back and the group is converted back to the classic group.
+ context.rollback();
+ assertEquals(group, context.groupMetadataManager.getOrMaybeCreateClassicGroup("group-id", false));
+ }
+
+ @Test
+ public void testConsumerGroupHeartbeatWithCompletingRebalanceClassicGroup() throws Exception {
+ String groupId = "group-id";
+ String memberId1 = "member-id-1";
+ String memberId2 = "member-id-2";
+ String memberId3 = "member-id-3";
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+ Uuid barTopicId = Uuid.randomUuid();
+ String barTopicName = "bar";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+ assignor.prepareGroupAssignment(new GroupAssignment(
+ new HashMap<String, MemberAssignment>() {
+ {
+ put(memberId1, new MemberAssignment(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0)
+ )));
+ put(memberId2, new MemberAssignment(mkAssignment(
+ mkTopicAssignment(barTopicId, 0)
+ )));
+ put(memberId3, new MemberAssignment(mkAssignment(
+ mkTopicAssignment(fooTopicId, 1)
+ )));
+ }
+ }
+ ));
+
+ MetadataImage metadataImage = new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 2)
+ .addTopic(barTopicId, barTopicName, 1)
+ .addRacks()
+ .build();
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.UPGRADE)
+ .withAssignors(Collections.singletonList(assignor))
+ .withMetadataImage(metadataImage)
+ .build();
+
+ JoinGroupRequestData.JoinGroupRequestProtocolCollection protocols1 = new JoinGroupRequestData.JoinGroupRequestProtocolCollection(1);
+ protocols1.add(new JoinGroupRequestData.JoinGroupRequestProtocol()
+ .setName("range")
+ .setMetadata(Utils.toArray(ConsumerProtocol.serializeSubscription(new ConsumerPartitionAssignor.Subscription(
+ Arrays.asList(fooTopicName, barTopicName),
+ null,
+ Arrays.asList(
+ new TopicPartition(fooTopicName, 0),
+ new TopicPartition(fooTopicName, 1)
+ )
+ ))))
+ );
+
+ JoinGroupRequestData.JoinGroupRequestProtocolCollection protocols2 = new JoinGroupRequestData.JoinGroupRequestProtocolCollection(1);
+ protocols2.add(new JoinGroupRequestData.JoinGroupRequestProtocol()
+ .setName("range")
+ .setMetadata(Utils.toArray(ConsumerProtocol.serializeSubscription(new ConsumerPartitionAssignor.Subscription(
+ Arrays.asList(fooTopicName, barTopicName),
+ null,
+ Arrays.asList(
+ new TopicPartition(barTopicName, 0)
+ )
+ ))))
+ );
+
+ Map<String, byte[]> assignments = new HashMap<String, byte[]>() {
+ {
+ put(
+ memberId1,
+ Utils.toArray(ConsumerProtocol.serializeAssignment(new ConsumerPartitionAssignor.Assignment(Arrays.asList(
+ new TopicPartition(fooTopicName, 0),
+ new TopicPartition(fooTopicName, 1)
+ ))))
+ );
+ put(
+ memberId2,
+ Utils.toArray(ConsumerProtocol.serializeAssignment(new ConsumerPartitionAssignor.Assignment(Arrays.asList(
+ new TopicPartition(barTopicName, 0)
+ ))))
+ );
+ }
+ };
+
+ // Construct a stable group with two members.
+ ClassicGroup group = context.createClassicGroup(groupId);
+ group.setProtocolName(Optional.ofNullable("range"));
+ group.add(
+ new ClassicGroupMember(
+ memberId1,
+ Optional.empty(),
+ "client-id",
+ "client-host",
+ 10000,
+ 5000,
+ "consumer",
+ protocols1,
+ assignments.get(memberId1)
+ )
+ );
+ group.add(
+ new ClassicGroupMember(
+ memberId2,
+ Optional.empty(),
+ "client-id",
+ "client-host",
+ 10000,
+ 5000,
+ "consumer",
+ protocols2,
+ assignments.get(memberId2)
+ )
+ );
+
+ group.transitionTo(PREPARING_REBALANCE);
+ group.transitionTo(COMPLETING_REBALANCE);
+ group.transitionTo(STABLE);
+
+ context.replay(RecordHelpers.newGroupMetadataRecord(group, assignments, metadataImage.features().metadataVersion()));
+ context.commit();
+ group = context.groupMetadataManager.getOrMaybeCreateClassicGroup(groupId, false);
+
+ // The leader rejoins, triggering a rebalance.
+ context.sendClassicGroupJoin(
+ new GroupMetadataManagerTestContext.JoinGroupRequestBuilder()
+ .withGroupId("group-id")
+ .withMemberId(memberId1)
+ .withProtocols(protocols1)
+ .withSessionTimeoutMs(5000)
+ .withRebalanceTimeoutMs(10000)
+ .build()
+ );
+
+ // The follower rejoins. All members have rejoined so the group transitions to COMPLETING_REBALANCE state.
+ context.sendClassicGroupJoin(
+ new GroupMetadataManagerTestContext.JoinGroupRequestBuilder()
+ .withGroupId("group-id")
+ .withMemberId(memberId2)
+ .withProtocols(protocols2)
+ .withSessionTimeoutMs(5000)
+ .withRebalanceTimeoutMs(10000)
+ .build()
+ );
+ assertTrue(group.isInState(COMPLETING_REBALANCE));
+
+ GroupMetadataManagerTestContext.SyncResult syncResult = context.sendClassicGroupSync(
+ new GroupMetadataManagerTestContext.SyncGroupRequestBuilder()
+ .withGroupId("group-id")
+ .withMemberId(memberId2)
+ .withGenerationId(1)
+ .build());
+
+ // Another new member 3 joins with new protocol, triggering the upgrade.
+ CoordinatorResult<ConsumerGroupHeartbeatResponseData, Record> consumerGroupHeartbeatResult = context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId3)
+ .setRebalanceTimeoutMs(5000)
+ .setServerAssignor("range")
+ .setSubscribedTopicNames(Arrays.asList(fooTopicName, barTopicName))
+ .setTopicPartitions(Collections.emptyList()));
+
+ ConsumerGroupMember expectedMember1 = new ConsumerGroupMember.Builder(memberId1)
+ .setMemberEpoch(1)
+ .setPreviousMemberEpoch(1)
+ .setClientId("client-id")
+ .setClientHost("client-host")
+ .setSubscribedTopicNames(Arrays.asList(fooTopicName, barTopicName))
+ .setRebalanceTimeoutMs(10000)
+ .setSupportedClassicProtocols(protocols1)
+ .setAssignedPartitions(new HashMap<Uuid, Set<Integer>>() {
+ {
+ put(fooTopicId, new HashSet<>(Arrays.asList(0, 1)));
+ }
+ })
+ .build();
+
+ ConsumerGroupMember expectedMember2 = new ConsumerGroupMember.Builder(memberId2)
+ .setMemberEpoch(1)
+ .setPreviousMemberEpoch(1)
+ .setClientId("client-id")
+ .setClientHost("client-host")
+ .setSubscribedTopicNames(Arrays.asList(fooTopicName, barTopicName))
+ .setRebalanceTimeoutMs(10000)
+ .setSupportedClassicProtocols(protocols2)
+ .setAssignedPartitions(new HashMap<Uuid, Set<Integer>>() {
+ {
+ put(barTopicId, new HashSet<>(Collections.singletonList(0)));
+ }
+ })
+ .build();
+
+ ConsumerGroupMember expectedMember3 = new ConsumerGroupMember.Builder(memberId3)
+ .setMemberEpoch(2)
+ .setPreviousMemberEpoch(0)
+ .setState(MemberState.UNRELEASED_PARTITIONS)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ .setServerAssignorName("range")
+ .setSubscribedTopicNames(Arrays.asList(fooTopicName, barTopicName))
+ .setRebalanceTimeoutMs(5000)
+ .setAssignedPartitions(Collections.emptyMap())
+ .build();
+
+ List<Record> expectedRecords = Arrays.asList(
+ // The existing classic group tombstone.
+ RecordHelpers.newGroupMetadataTombstoneRecord(groupId),
+
+ // Create the new consumer group with member 1 and member 2.
+ RecordHelpers.newMemberSubscriptionRecord(groupId, expectedMember1),
+ RecordHelpers.newMemberSubscriptionRecord(groupId, expectedMember2),
+
+ RecordHelpers.newGroupEpochRecord(groupId, 1),
+ RecordHelpers.newTargetAssignmentRecord(groupId, memberId1, expectedMember1.assignedPartitions()),
+ RecordHelpers.newTargetAssignmentRecord(groupId, memberId2, expectedMember2.assignedPartitions()),
+
+ RecordHelpers.newTargetAssignmentEpochRecord(groupId, 1),
+
+ RecordHelpers.newCurrentAssignmentRecord(groupId, expectedMember1),
+ RecordHelpers.newCurrentAssignmentRecord(groupId, expectedMember2),
+
+ // Member 3 joins the new consumer group.
+ RecordHelpers.newMemberSubscriptionRecord(groupId, expectedMember3),
+
+ // The subscription metadata hasn't been updated during the conversion, so a new one is computed.
+ RecordHelpers.newGroupSubscriptionMetadataRecord(groupId, new HashMap<String, TopicMetadata>() {
+ {
+ put(fooTopicName, new TopicMetadata(fooTopicId, fooTopicName, 2, new HashMap<Integer, Set<String>>() {
+ {
+ put(0, new HashSet<>(Arrays.asList("rack0", "rack1")));
+ put(1, new HashSet<>(Arrays.asList("rack1", "rack2")));
+ }
+ }));
+ put(barTopicName, new TopicMetadata(barTopicId, barTopicName, 1, new HashMap<Integer, Set<String>>() {
+ {
+ put(0, new HashSet<>(Arrays.asList("rack0", "rack1")));
+ }
+ }));
+ }
+ }),
+
+ // Newly joining member 3 bumps the group epoch. A new target assignment is computed.
+ RecordHelpers.newGroupEpochRecord(groupId, 2),
+ RecordHelpers.newTargetAssignmentRecord(groupId, memberId1, new HashMap<Uuid, Set<Integer>>() {
+ {
+ put(fooTopicId, new HashSet<>(Collections.singletonList(0)));
+ }
+ }),
+ RecordHelpers.newTargetAssignmentRecord(groupId, memberId3, new HashMap<Uuid, Set<Integer>>() {
+ {
+ put(fooTopicId, new HashSet<>(Collections.singletonList(1)));
+ }
+ }),
+ RecordHelpers.newTargetAssignmentEpochRecord(groupId, 2),
+
+ // Member 3 has no pending revoking partition. Bump its member epoch and transition to UNRELEASED_PARTITIONS.
+ RecordHelpers.newCurrentAssignmentRecord(groupId, expectedMember3)
+ );
+
+ assertRecordsEquals(expectedRecords, consumerGroupHeartbeatResult.records());
+ assertTrue(syncResult.syncFuture.isDone());
+ assertEquals(Errors.REBALANCE_IN_PROGRESS.code(), syncResult.syncFuture.get().errorCode());
+
+ context.assertSessionTimeout(groupId, memberId1, 45000);
+ context.assertSessionTimeout(groupId, memberId2, 45000);
+ context.assertSessionTimeout(groupId, memberId3, 45000);
+
+ // Simulate a failed replay. The context is rolled back and the group is converted back to the classic group.
+ context.rollback();
+ assertEquals(group, context.groupMetadataManager.getOrMaybeCreateClassicGroup("group-id", false));
+ }
+
@Test
public void testClassicGroupOnUnloadedEmptyAndPreparingRebalance() throws Exception {
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTestContext.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTestContext.java
index 86b0b12d998..d60ec65ae86 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTestContext.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTestContext.java
@@ -353,6 +353,7 @@ public class GroupMetadataManagerTestContext {
private int classicGroupMinSessionTimeoutMs = 10;
private int classicGroupMaxSessionTimeoutMs = 10 * 60 * 1000;
final private GroupCoordinatorMetricsShard metrics = mock(GroupCoordinatorMetricsShard.class);
+ private ConsumerGroupMigrationPolicy consumerGroupMigrationPolicy = ConsumerGroupMigrationPolicy.DISABLED;
public Builder withMetadataImage(MetadataImage metadataImage) {
this.metadataImage = metadataImage;
@@ -399,6 +400,11 @@ public class GroupMetadataManagerTestContext {
return this;
}
+ public Builder withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy consumerGroupMigrationPolicy) {
+ this.consumerGroupMigrationPolicy = consumerGroupMigrationPolicy;
+ return this;
+ }
+
public GroupMetadataManagerTestContext build() {
if (metadataImage == null) metadataImage = MetadataImage.EMPTY;
if (consumerGroupAssignors == null) consumerGroupAssignors = Collections.emptyList();
@@ -425,6 +431,7 @@ public class GroupMetadataManagerTestContext {
.withClassicGroupInitialRebalanceDelayMs(classicGroupInitialRebalanceDelayMs)
.withClassicGroupNewMemberJoinTimeoutMs(classicGroupNewMemberJoinTimeoutMs)
.withGroupCoordinatorMetricsShard(metrics)
+ .withConsumerGroupMigrationPolicy(consumerGroupMigrationPolicy)
.build(),
classicGroupInitialRebalanceDelayMs,
classicGroupNewMemberJoinTimeoutMs
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/RecordHelpersTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/RecordHelpersTest.java
index 17d24959a1d..e8fb736aa8b 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/RecordHelpersTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/RecordHelpersTest.java
@@ -92,6 +92,11 @@ public class RecordHelpersTest {
@Test
public void testNewMemberSubscriptionRecord() {
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = new ArrayList<>();
+ protocols.add(new ConsumerGroupMemberMetadataValue.ClassicProtocol()
+ .setName("range")
+ .setMetadata(new byte[0]));
+
ConsumerGroupMember member = new ConsumerGroupMember.Builder("member-id")
.setInstanceId("instance-id")
.setRackId("rack-id")
@@ -101,6 +106,8 @@ public class RecordHelpersTest {
.setSubscribedTopicNames(Arrays.asList("foo", "zar", "bar"))
.setSubscribedTopicRegex("regex")
.setServerAssignorName("range")
+ .setClassicMemberMetadata(new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(protocols))
.build();
Record expectedRecord = new Record(
@@ -118,7 +125,9 @@ public class RecordHelpersTest {
.setClientHost("client-host")
.setSubscribedTopicNames(Arrays.asList("bar", "foo", "zar"))
.setSubscribedTopicRegex("regex")
- .setServerAssignor("range"),
+ .setServerAssignor("range")
+ .setClassicMemberMetadata(new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(protocols)),
(short) 0));
assertEquals(expectedRecord, newMemberSubscriptionRecord(
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/classic/ClassicGroupTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/classic/ClassicGroupTest.java
index 0d4f4d956c7..14327ae9672 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/classic/ClassicGroupTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/classic/ClassicGroupTest.java
@@ -42,14 +42,17 @@ import org.apache.kafka.timeline.SnapshotRegistry;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
+import java.util.List;
import java.util.Optional;
import java.util.OptionalInt;
import java.util.OptionalLong;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
import static org.apache.kafka.common.utils.Utils.mkSet;
import static org.apache.kafka.coordinator.group.classic.ClassicGroupState.COMPLETING_REBALANCE;
@@ -1296,6 +1299,84 @@ public class ClassicGroupTest {
assertTrue(group.isInStates(new HashSet<>(Arrays.asList("dead", " ")), 0));
}
+ @Test
+ public void testCompleteAllJoinFutures() throws ExecutionException, InterruptedException {
+ JoinGroupRequestProtocolCollection protocols = new JoinGroupRequestProtocolCollection();
+ protocols.add(new JoinGroupRequestProtocol()
+ .setName("roundrobin")
+ .setMetadata(new byte[0]));
+
+ List<ClassicGroupMember> memberList = new ArrayList<>();
+ for (int i = 0; i < 3; i++) {
+ memberList.add(new ClassicGroupMember(
+ memberId + i,
+ Optional.empty(),
+ clientId,
+ clientHost,
+ rebalanceTimeoutMs,
+ sessionTimeoutMs,
+ protocolType,
+ protocols
+ ));
+ }
+
+ List<CompletableFuture<JoinGroupResponseData>> joinGroupFutureList = new ArrayList<>();
+ for (int i = 0; i < 3; i++) {
+ CompletableFuture<JoinGroupResponseData> future = new CompletableFuture<>();
+ group.add(memberList.get(i), future);
+ joinGroupFutureList.add(future);
+ }
+
+ assertEquals(3, group.numAwaitingJoinResponse());
+
+ group.completeAllJoinFutures(Errors.REBALANCE_IN_PROGRESS);
+
+ for (int i = 0; i < 3; i++) {
+ assertEquals(Errors.REBALANCE_IN_PROGRESS.code(), joinGroupFutureList.get(i).get().errorCode());
+ assertEquals(memberId + i, joinGroupFutureList.get(i).get().memberId());
+ assertFalse(memberList.get(i).isAwaitingJoin());
+ }
+ assertEquals(0, group.numAwaitingJoinResponse());
+ }
+
+ @Test
+ public void testCompleteAllSyncFutures() throws ExecutionException, InterruptedException {
+ JoinGroupRequestProtocolCollection protocols = new JoinGroupRequestProtocolCollection();
+ protocols.add(new JoinGroupRequestProtocol()
+ .setName("roundrobin")
+ .setMetadata(new byte[0]));
+
+ List<ClassicGroupMember> memberList = new ArrayList<>();
+ for (int i = 0; i < 3; i++) {
+ ClassicGroupMember member = new ClassicGroupMember(
+ memberId + i,
+ Optional.empty(),
+ clientId,
+ clientHost,
+ rebalanceTimeoutMs,
+ sessionTimeoutMs,
+ protocolType,
+ protocols
+ );
+ memberList.add(member);
+ group.add(member);
+ }
+
+ List<CompletableFuture<SyncGroupResponseData>> syncGroupFutureList = new ArrayList<>();
+ for (int i = 0; i < 3; i++) {
+ CompletableFuture<SyncGroupResponseData> syncGroupFuture = new CompletableFuture<>();
+ syncGroupFutureList.add(syncGroupFuture);
+ memberList.get(i).setAwaitingSyncFuture(syncGroupFuture);
+ }
+
+ group.completeAllSyncFutures(Errors.REBALANCE_IN_PROGRESS);
+
+ for (int i = 0; i < 3; i++) {
+ assertFalse(memberList.get(i).isAwaitingSync());
+ assertEquals(Errors.REBALANCE_IN_PROGRESS.code(), syncGroupFutureList.get(i).get().errorCode());
+ }
+ }
+
private void assertState(ClassicGroup group, ClassicGroupState targetState) {
Set<ClassicGroupState> otherStates = new HashSet<>();
otherStates.add(STABLE);
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupMemberTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupMemberTest.java
index cfb9a79a2d6..ffc350d9c28 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupMemberTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupMemberTest.java
@@ -18,6 +18,7 @@ package org.apache.kafka.coordinator.group.consumer;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.message.ConsumerGroupDescribeResponseData;
+import org.apache.kafka.common.message.JoinGroupRequestData;
import org.apache.kafka.coordinator.group.MetadataImageBuilder;
import org.apache.kafka.coordinator.group.generated.ConsumerGroupCurrentMemberAssignmentValue;
import org.apache.kafka.coordinator.group.generated.ConsumerGroupMemberMetadataValue;
@@ -39,6 +40,7 @@ import java.util.stream.Collectors;
import static org.apache.kafka.coordinator.group.AssignmentTestUtil.mkAssignment;
import static org.apache.kafka.coordinator.group.AssignmentTestUtil.mkTopicAssignment;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
public class ConsumerGroupMemberTest {
@@ -46,7 +48,6 @@ public class ConsumerGroupMemberTest {
public void testNewMember() {
Uuid topicId1 = Uuid.randomUuid();
Uuid topicId2 = Uuid.randomUuid();
- Uuid topicId3 = Uuid.randomUuid();
ConsumerGroupMember member = new ConsumerGroupMember.Builder("member-id")
.setMemberEpoch(10)
@@ -63,6 +64,8 @@ public class ConsumerGroupMemberTest {
mkTopicAssignment(topicId1, 1, 2, 3)))
.setPartitionsPendingRevocation(mkAssignment(
mkTopicAssignment(topicId2, 4, 5, 6)))
+ .setClassicMemberMetadata(new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(toClassicProtocolCollection("range")))
.build();
assertEquals("member-id", member.memberId());
@@ -78,13 +81,18 @@ public class ConsumerGroupMemberTest {
assertEquals("range", member.serverAssignorName().get());
assertEquals(mkAssignment(mkTopicAssignment(topicId1, 1, 2, 3)), member.assignedPartitions());
assertEquals(mkAssignment(mkTopicAssignment(topicId2, 4, 5, 6)), member.partitionsPendingRevocation());
+ assertEquals(
+ new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(toClassicProtocolCollection("range")),
+ member.classicMemberMetadata().get()
+ );
+ assertEquals(toClassicProtocolCollection("range"), member.supportedClassicProtocols().get());
}
@Test
public void testEquals() {
Uuid topicId1 = Uuid.randomUuid();
Uuid topicId2 = Uuid.randomUuid();
- Uuid topicId3 = Uuid.randomUuid();
ConsumerGroupMember member1 = new ConsumerGroupMember.Builder("member-id")
.setMemberEpoch(10)
@@ -101,6 +109,8 @@ public class ConsumerGroupMemberTest {
mkTopicAssignment(topicId1, 1, 2, 3)))
.setPartitionsPendingRevocation(mkAssignment(
mkTopicAssignment(topicId2, 4, 5, 6)))
+ .setClassicMemberMetadata(new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(toClassicProtocolCollection("range")))
.build();
ConsumerGroupMember member2 = new ConsumerGroupMember.Builder("member-id")
@@ -118,6 +128,8 @@ public class ConsumerGroupMemberTest {
mkTopicAssignment(topicId1, 1, 2, 3)))
.setPartitionsPendingRevocation(mkAssignment(
mkTopicAssignment(topicId2, 4, 5, 6)))
+ .setClassicMemberMetadata(new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(toClassicProtocolCollection("range")))
.build();
assertEquals(member1, member2);
@@ -127,7 +139,6 @@ public class ConsumerGroupMemberTest {
public void testUpdateMember() {
Uuid topicId1 = Uuid.randomUuid();
Uuid topicId2 = Uuid.randomUuid();
- Uuid topicId3 = Uuid.randomUuid();
ConsumerGroupMember member = new ConsumerGroupMember.Builder("member-id")
.setMemberEpoch(10)
@@ -144,6 +155,8 @@ public class ConsumerGroupMemberTest {
mkTopicAssignment(topicId1, 1, 2, 3)))
.setPartitionsPendingRevocation(mkAssignment(
mkTopicAssignment(topicId2, 4, 5, 6)))
+ .setClassicMemberMetadata(new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(toClassicProtocolCollection("range")))
.build();
// This is a no-op.
@@ -185,7 +198,9 @@ public class ConsumerGroupMemberTest {
.setRackId("rack-id")
.setRebalanceTimeoutMs(1000)
.setSubscribedTopicNames(Arrays.asList("foo", "bar"))
- .setSubscribedTopicRegex("regex");
+ .setSubscribedTopicRegex("regex")
+ .setClassicMemberMetadata(new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(toClassicProtocolCollection("range")));
ConsumerGroupMember member = new ConsumerGroupMember.Builder("member-id")
.updateWith(record)
@@ -199,13 +214,17 @@ public class ConsumerGroupMemberTest {
assertEquals(Arrays.asList("bar", "foo"), member.subscribedTopicNames());
assertEquals("regex", member.subscribedTopicRegex());
assertEquals("range", member.serverAssignorName().get());
+ assertEquals(
+ new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(toClassicProtocolCollection("range")),
+ member.classicMemberMetadata().get()
+ );
}
@Test
public void testUpdateWithConsumerGroupCurrentMemberAssignmentValue() {
Uuid topicId1 = Uuid.randomUuid();
Uuid topicId2 = Uuid.randomUuid();
- Uuid topicId3 = Uuid.randomUuid();
ConsumerGroupCurrentMemberAssignmentValue record = new ConsumerGroupCurrentMemberAssignmentValue()
.setMemberEpoch(10)
@@ -333,4 +352,48 @@ public class ConsumerGroupMemberTest {
);
assertEquals(expected, actual);
}
+
+ @Test
+ public void testSetSupportedClassicProtocolsWithJoinGroupRequestProtocolCollections() {
+ Uuid topicId1 = Uuid.randomUuid();
+ Uuid topicId2 = Uuid.randomUuid();
+
+ JoinGroupRequestData.JoinGroupRequestProtocolCollection protocols = new JoinGroupRequestData.JoinGroupRequestProtocolCollection();
+ protocols.addAll(Arrays.asList(
+ new JoinGroupRequestData.JoinGroupRequestProtocol()
+ .setName("range")
+ .setMetadata(new byte[]{1, 2, 3})
+ ));
+
+ ConsumerGroupMember member = new ConsumerGroupMember.Builder("member-id")
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setInstanceId("instance-id")
+ .setRackId("rack-id")
+ .setRebalanceTimeoutMs(5000)
+ .setClientId("client-id")
+ .setClientHost("hostname")
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setSubscribedTopicRegex("regex")
+ .setServerAssignorName("range")
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(topicId1, 1, 2, 3)))
+ .setPartitionsPendingRevocation(mkAssignment(
+ mkTopicAssignment(topicId2, 4, 5, 6)))
+ .setSupportedClassicProtocols(protocols)
+ .build();
+
+ assertEquals(toClassicProtocolCollection("range"), member.supportedClassicProtocols().get());
+ assertTrue(member.useClassicProtocol());
+ }
+
+ private List<ConsumerGroupMemberMetadataValue.ClassicProtocol> toClassicProtocolCollection(String name) {
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = new ArrayList<>();
+ protocols.add(
+ new ConsumerGroupMemberMetadataValue.ClassicProtocol()
+ .setName(name)
+ .setMetadata(new byte[]{1, 2, 3})
+ );
+ return protocols;
+ }
}
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupTest.java
index 15261527067..8a25a752ccf 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.kafka.coordinator.group.consumer;
+import org.apache.kafka.clients.consumer.internals.ConsumerProtocol;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.errors.GroupNotEmptyException;
@@ -29,6 +30,7 @@ import org.apache.kafka.coordinator.group.MetadataImageBuilder;
import org.apache.kafka.coordinator.group.OffsetAndMetadata;
import org.apache.kafka.coordinator.group.OffsetExpirationCondition;
import org.apache.kafka.coordinator.group.OffsetExpirationConditionImpl;
+import org.apache.kafka.coordinator.group.generated.ConsumerGroupMemberMetadataValue;
import org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetricsShard;
import org.apache.kafka.image.MetadataImage;
import org.apache.kafka.timeline.SnapshotRegistry;
@@ -36,8 +38,11 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
import java.util.Optional;
import java.util.OptionalInt;
import java.util.OptionalLong;
@@ -1046,4 +1051,101 @@ public class ConsumerGroupTest {
assertTrue(group.isInStates(Collections.singleton("stable"), 1));
assertFalse(group.isInStates(Collections.singleton("empty"), 1));
}
+
+ @Test
+ public void testClassicMembersSupportedProtocols() {
+ ConsumerGroup consumerGroup = createConsumerGroup("foo");
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> rangeProtocol = new ArrayList<>();
+ rangeProtocol.add(new ConsumerGroupMemberMetadataValue.ClassicProtocol()
+ .setName("range")
+ .setMetadata(new byte[0]));
+
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> roundRobinAndRangeProtocols = new ArrayList<>();
+ roundRobinAndRangeProtocols.add(new ConsumerGroupMemberMetadataValue.ClassicProtocol()
+ .setName("roundrobin")
+ .setMetadata(new byte[0]));
+ roundRobinAndRangeProtocols.add(new ConsumerGroupMemberMetadataValue.ClassicProtocol()
+ .setName("range")
+ .setMetadata(new byte[0]));
+
+ ConsumerGroupMember member1 = new ConsumerGroupMember.Builder("member-1")
+ .setClassicMemberMetadata(new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(rangeProtocol))
+ .build();
+ consumerGroup.updateMember(member1);
+
+ ConsumerGroupMember member2 = new ConsumerGroupMember.Builder("member-2")
+ .setClassicMemberMetadata(new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(roundRobinAndRangeProtocols))
+ .build();
+ consumerGroup.updateMember(member2);
+
+ assertEquals(2, consumerGroup.classicMembersSupportedProtocols().get("range"));
+ assertEquals(1, consumerGroup.classicMembersSupportedProtocols().get("roundrobin"));
+ assertTrue(consumerGroup.supportsClassicProtocols(ConsumerProtocol.PROTOCOL_TYPE, new HashSet<>(Arrays.asList("range", "sticky"))));
+ assertFalse(consumerGroup.supportsClassicProtocols(ConsumerProtocol.PROTOCOL_TYPE, new HashSet<>(Arrays.asList("sticky", "roundrobin"))));
+
+ member2 = new ConsumerGroupMember.Builder(member2)
+ .setClassicMemberMetadata(new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(rangeProtocol))
+ .build();
+ consumerGroup.updateMember(member2);
+
+ assertEquals(2, consumerGroup.classicMembersSupportedProtocols().get("range"));
+ assertFalse(consumerGroup.classicMembersSupportedProtocols().containsKey("roundrobin"));
+
+ member1 = new ConsumerGroupMember.Builder(member1)
+ .setClassicMemberMetadata(new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(roundRobinAndRangeProtocols))
+ .build();
+ consumerGroup.updateMember(member1);
+ member2 = new ConsumerGroupMember.Builder(member2)
+ .setClassicMemberMetadata(new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(roundRobinAndRangeProtocols))
+ .build();
+ consumerGroup.updateMember(member2);
+
+ assertEquals(2, consumerGroup.classicMembersSupportedProtocols().get("range"));
+ assertEquals(2, consumerGroup.classicMembersSupportedProtocols().get("roundrobin"));
+ assertTrue(consumerGroup.supportsClassicProtocols(ConsumerProtocol.PROTOCOL_TYPE, new HashSet<>(Arrays.asList("sticky", "roundrobin"))));
+ }
+
+ @Test
+ public void testAllMembersUseClassicProtocol() {
+ ConsumerGroup consumerGroup = createConsumerGroup("foo");
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = new ArrayList<>();
+ protocols.add(new ConsumerGroupMemberMetadataValue.ClassicProtocol()
+ .setName("range")
+ .setMetadata(new byte[0]));
+
+ // The group has member 1 (using the classic protocol).
+ ConsumerGroupMember member1 = new ConsumerGroupMember.Builder("member-1")
+ .setClassicMemberMetadata(new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(protocols))
+ .build();
+ consumerGroup.updateMember(member1);
+ assertEquals(1, consumerGroup.numClassicProtocolMembers());
+ assertTrue(consumerGroup.allMembersUseClassicProtocol());
+
+ // The group has member 1 (using the classic protocol) and member 2 (using the consumer protocol).
+ ConsumerGroupMember member2 = new ConsumerGroupMember.Builder("member-2")
+ .build();
+ consumerGroup.updateMember(member2);
+ assertEquals(1, consumerGroup.numClassicProtocolMembers());
+ assertFalse(consumerGroup.allMembersUseClassicProtocol());
+
+ // The group has member 2 (using the consumer protocol).
+ consumerGroup.removeMember(member1.memberId());
+ assertEquals(0, consumerGroup.numClassicProtocolMembers());
+ assertFalse(consumerGroup.allMembersUseClassicProtocol());
+
+ // The group has member 2 (using the classic protocol).
+ member2 = new ConsumerGroupMember.Builder("member-2")
+ .setClassicMemberMetadata(new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(protocols))
+ .build();
+ consumerGroup.updateMember(member2);
+ assertEquals(1, consumerGroup.numClassicProtocolMembers());
+ assertTrue(consumerGroup.allMembersUseClassicProtocol());
+ }
}