You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "dongnuo123 (via GitHub)" <gi...@apache.org> on 2024/04/15 18:17:45 UTC
[PR] KAFKA-16554: Online downgrade triggering and group type conversion [kafka]
dongnuo123 opened a new pull request, #15721:
URL: https://github.com/apache/kafka/pull/15721
Online downgrade from a consumer group to a classic group is triggered when the last consumer that uses the consumer protocol leaves the group. A rebalance is manually triggered after the group conversion.
This patch adds consumer group downgrade validation and conversion.
https://issues.apache.org/jira/browse/KAFKA-16554
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16554: Online downgrade triggering and group type conversion [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #15721:
URL: https://github.com/apache/kafka/pull/15721#discussion_r1569077848
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -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);
Review Comment:
I think that we should explicitly remove the previous group before adding the new one because we update metrics when the previous group is removed. We could likely call `removeGroup` for this purpose.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16554: Online downgrade triggering and group type conversion [kafka]
Posted by "dongnuo123 (via GitHub)" <gi...@apache.org>.
dongnuo123 commented on code in PR #15721:
URL: https://github.com/apache/kafka/pull/15721#discussion_r1568994033
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java:
##########
@@ -998,4 +1094,232 @@ public ConsumerGroupDescribeResponseData.DescribedGroup asDescribedGroup(
);
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;
+ }
+
+ /**
+ * 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(),
Review Comment:
I'm not sure if I get it. The stored protocols should keep updating whenever the group receives the JoinGroupRequest during the mixed mode. So the protocol here was stored in the member's latest JoinGroupRequest.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16554: Online downgrade triggering and group type conversion [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #15721:
URL: https://github.com/apache/kafka/pull/15721#discussion_r1569055496
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java:
##########
@@ -998,4 +1094,232 @@ public ConsumerGroupDescribeResponseData.DescribedGroup asDescribedGroup(
);
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;
+ }
+
+ /**
+ * 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(),
Review Comment:
Oh, I see. Then my comment does not make any sense.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16554: Online downgrade triggering and group type conversion [kafka]
Posted by "dongnuo123 (via GitHub)" <gi...@apache.org>.
dongnuo123 commented on code in PR #15721:
URL: https://github.com/apache/kafka/pull/15721#discussion_r1568994033
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java:
##########
@@ -998,4 +1094,232 @@ public ConsumerGroupDescribeResponseData.DescribedGroup asDescribedGroup(
);
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;
+ }
+
+ /**
+ * 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(),
Review Comment:
I'm not sure if I get it. The stored protocols should keep updating whenever the group receives the JoinGroupRequest during the mixed mode. So the protocol here is stored in the member's latest JoinGroupRequest.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16554: Online downgrade triggering and group type conversion [kafka]
Posted by "dongnuo123 (via GitHub)" <gi...@apache.org>.
dongnuo123 commented on code in PR #15721:
URL: https://github.com/apache/kafka/pull/15721#discussion_r1571469219
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -777,11 +778,78 @@ public ClassicGroup classicGroup(
}
}
+ /**
+ * Validates the online downgrade if a consumer member is fenced from the consumer group.
+ *
+ * @param consumerGroup The ConsumerGroup.
+ * @param memberId The fenced member id.
+ * @return A boolean indicating whether it's valid to online downgrade the consumer group.
+ */
+ private 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)) {
Review Comment:
For explicitly leaving group we know the remaining one is using the consumer group protocol. I'm not sure about the timeout cases. It could be an expired classic protocol member.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16554: Online downgrade triggering and group type conversion [kafka]
Posted by "dongnuo123 (via GitHub)" <gi...@apache.org>.
dongnuo123 commented on code in PR #15721:
URL: https://github.com/apache/kafka/pull/15721#discussion_r1569405350
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -775,6 +777,126 @@ 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()));
Review Comment:
Yeah that makes sense.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16554: Online downgrade triggering and group type conversion [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #15721:
URL: https://github.com/apache/kafka/pull/15721#discussion_r1571079283
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -777,11 +778,78 @@ public ClassicGroup classicGroup(
}
}
+ /**
+ * Validates the online downgrade if a consumer member is fenced from the consumer group.
+ *
+ * @param consumerGroup The ConsumerGroup.
+ * @param memberId The fenced member id.
+ * @return A boolean indicating whether it's valid to online downgrade the consumer group.
+ */
+ private 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 = ClassicGroup.fromConsumerGroup(
+ consumerGroup,
+ leavingMemberId,
+ logContext,
+ time,
+ metrics,
+ consumerGroupSessionTimeoutMs,
+ metadataImage
+ );
+ } 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()));
+ }
+ classicGroup.createConsumerGroupRecords(metadataImage.features().metadataVersion(), records);
+
+ removeGroup(consumerGroup.groupId());
+
+ groups.put(consumerGroup.groupId(), classicGroup);
+ metrics.onClassicGroupStateTransition(null, classicGroup.currentState());
+
+ classicGroup.allMembers().forEach(member -> rescheduleClassicGroupMemberHeartbeat(classicGroup, member));
+ prepareRebalance(classicGroup, String.format("Downgrade group %s.", classicGroup.groupId()));
+
+ CompletableFuture<Void> appendFuture = new CompletableFuture<>();
+ appendFuture.whenComplete((__, t) -> {
Review Comment:
I wonder if we could use `exceptionally` here.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -777,11 +778,78 @@ public ClassicGroup classicGroup(
}
}
+ /**
+ * Validates the online downgrade if a consumer member is fenced from the consumer group.
+ *
+ * @param consumerGroup The ConsumerGroup.
+ * @param memberId The fenced member id.
+ * @return A boolean indicating whether it's valid to online downgrade the consumer group.
+ */
+ private 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 = ClassicGroup.fromConsumerGroup(
+ consumerGroup,
+ leavingMemberId,
+ logContext,
+ time,
+ metrics,
+ consumerGroupSessionTimeoutMs,
+ metadataImage
+ );
+ } 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()));
+ }
+ classicGroup.createConsumerGroupRecords(metadataImage.features().metadataVersion(), records);
+
+ removeGroup(consumerGroup.groupId());
Review Comment:
Let's put a comment about this one. We should explain that we don't replay so we have to remove it ourselves.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -777,11 +778,78 @@ public ClassicGroup classicGroup(
}
}
+ /**
+ * Validates the online downgrade if a consumer member is fenced from the consumer group.
+ *
+ * @param consumerGroup The ConsumerGroup.
+ * @param memberId The fenced member id.
+ * @return A boolean indicating whether it's valid to online downgrade the consumer group.
+ */
+ private 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) {
Review Comment:
Could we make it private or package private?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -10331,6 +10333,312 @@ public void testClassicGroupOnUnloadedCompletingRebalance() throws Exception {
.setErrorCode(NOT_COORDINATOR.code()), pendingMemberSyncResult.syncFuture.get());
}
+ @Test
+ public void testLastClassicProtocolMemberLeavingConsumerGroup() {
+ String groupId = "group-id";
+ String memberId1 = Uuid.randomUuid().toString();
+ String memberId2 = Uuid.randomUuid().toString();
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+ Uuid barTopicId = Uuid.randomUuid();
+ String barTopicName = "bar";
+ Uuid zarTopicId = Uuid.randomUuid();
+ String zarTopicName = "zar";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = Collections.singletonList(
+ new ConsumerGroupMemberMetadataValue.ClassicProtocol()
+ .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),
+ new TopicPartition(fooTopicName, 2),
+ new TopicPartition(barTopicName, 0),
+ new TopicPartition(barTopicName, 1)
+ )
+ ))))
+ );
+
+ ConsumerGroupMember member1 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setClassicMemberMetadata(new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(protocols))
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .build();
+ ConsumerGroupMember member2 = new ConsumerGroupMember.Builder(memberId2)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ // Use zar only here to ensure that metadata needs to be recomputed.
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar", "zar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .build();
+
+ // Consumer group with two members.
+ // Member 1 uses the classic protocol and member 2 uses the consumer protocol.
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.DOWNGRADE)
+ .withAssignors(Collections.singletonList(assignor))
+ .withMetadataImage(new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .addTopic(barTopicId, barTopicName, 3)
+ .addTopic(zarTopicId, zarTopicName, 1)
+ .addRacks()
+ .build())
+ .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
+ .withMember(member1)
+ .withMember(member2)
+ .withAssignment(memberId1, mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .withAssignment(memberId2, mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .withAssignmentEpoch(10))
+ .build();
+
+ context.commit();
+ ConsumerGroup consumerGroup = context.groupMetadataManager.consumerGroup(groupId);
+
+ // Member 2 leaves the consumer group, triggering the downgrade.
+ CoordinatorResult<ConsumerGroupHeartbeatResponseData, Record> result = context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId2)
+ .setMemberEpoch(LEAVE_GROUP_MEMBER_EPOCH)
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setTopicPartitions(Collections.emptyList()));
+
+
+ byte[] assignment = Utils.toArray(ConsumerProtocol.serializeAssignment(new ConsumerPartitionAssignor.Assignment(Arrays.asList(
+ new TopicPartition(fooTopicName, 0), new TopicPartition(fooTopicName, 1), new TopicPartition(fooTopicName, 2),
+ new TopicPartition(barTopicName, 0), new TopicPartition(barTopicName, 1)
+ ))));
+ Map<String, byte[]> assignments = new HashMap<String, byte[]>() {
+ {
+ put(memberId1, assignment);
+ }
+ };
+
+ ClassicGroup expectedClassicGroup = new ClassicGroup(
+ new LogContext(),
+ groupId,
+ STABLE,
+ context.time,
+ context.metrics,
+ 10,
+ Optional.ofNullable(ConsumerProtocol.PROTOCOL_TYPE),
+ Optional.ofNullable("range"),
+ Optional.ofNullable(memberId1),
+ Optional.of(context.time.milliseconds())
+ );
+ expectedClassicGroup.add(
+ new ClassicGroupMember(
+ memberId1,
+ Optional.ofNullable(member1.instanceId()),
+ member1.clientId(),
+ member1.clientHost(),
+ member1.rebalanceTimeoutMs(),
+ 45000,
+ ConsumerProtocol.PROTOCOL_TYPE,
+ member1.supportedJoinGroupRequestProtocols(),
+ assignment
+ )
+ );
+
+ List<Record> expectedRecords = Arrays.asList(
+ RecordHelpers.newCurrentAssignmentTombstoneRecord(groupId, memberId2),
+ RecordHelpers.newTargetAssignmentTombstoneRecord(groupId, memberId2),
+ RecordHelpers.newMemberSubscriptionTombstoneRecord(groupId, memberId2),
+ // Subscription metadata is recomputed because zar is no longer there.
+ RecordHelpers.newGroupSubscriptionMetadataRecord(groupId, new HashMap<String, TopicMetadata>() {
+ {
+ put(fooTopicName, new TopicMetadata(fooTopicId, fooTopicName, 6, mkMapOfPartitionRacks(6)));
+ put(barTopicName, new TopicMetadata(barTopicId, barTopicName, 3, mkMapOfPartitionRacks(3)));
+ }
+ }),
+ RecordHelpers.newGroupEpochRecord(groupId, 11),
+
+ RecordHelpers.newTargetAssignmentEpochTombstoneRecord(groupId),
+ RecordHelpers.newGroupSubscriptionMetadataTombstoneRecord(groupId),
+ RecordHelpers.newGroupEpochTombstoneRecord(groupId),
+ RecordHelpers.newGroupMetadataRecord(expectedClassicGroup, assignments, MetadataVersion.latestTesting())
+ );
+
+ assertRecordsEquals(expectedRecords, result.records());
+ verify(context.metrics, times(1)).onConsumerGroupStateTransition(ConsumerGroup.ConsumerGroupState.STABLE, null);
+ verify(context.metrics, times(1)).onClassicGroupStateTransition(null, STABLE);
+
+ // The new classic member 1 has a heartbeat timeout.
+ ScheduledTimeout<Void, Record> heartbeatTimeout = context.timer.timeout(
+ classicGroupHeartbeatKey(groupId, memberId1));
+ assertNotNull(heartbeatTimeout);
+ // The new rebalance has a groupJoin timeout.
+ ScheduledTimeout<Void, Record> groupJoinTimeout = context.timer.timeout(
+ classicGroupJoinKey(groupId));
+ assertNotNull(groupJoinTimeout);
+
+ // A new rebalance is triggered.
+ ClassicGroup classicGroup = context.groupMetadataManager.getOrMaybeCreateClassicGroup(groupId, false);
+ assertTrue(classicGroup.isInState(PREPARING_REBALANCE));
+
+ // Simulate a failed write to the log.
+ result.appendFuture().completeExceptionally(new NotLeaderOrFollowerException());
+ context.rollback();
+
+ // The group is reverted back to the consumer group.
+ assertEquals(consumerGroup, context.groupMetadataManager.consumerGroup(groupId));
+ verify(context.metrics, times(1)).onClassicGroupStateTransition(PREPARING_REBALANCE, null);
+ }
+
+ @Test
+ public void testLastClassicProtocolMemberSessionTimeoutInConsumerGroup() {
Review Comment:
Should we also add a test case for the rebalance timeout path?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/Assertions.java:
##########
@@ -198,6 +204,62 @@ private static void assertApiMessageAndVersionEquals(
}
}
}
+ } else if (actual.message() instanceof GroupMetadataValue) {
Review Comment:
I understand that you replicated the pattern already in place. However, I don't like it even if I wrote it. Recently, I have been using a different approach which is better, I think.
I do the following:
1) I duplicate both messages (with `duplicate()` method)
2) I normalize them in place (e.g. sort lists, etc.)
3) I use `assertEquals` to compare them.
The benefit of this approach is that it automatically includes new fields. Would it work here too?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -777,11 +778,78 @@ public ClassicGroup classicGroup(
}
}
+ /**
+ * Validates the online downgrade if a consumer member is fenced from the consumer group.
+ *
+ * @param consumerGroup The ConsumerGroup.
+ * @param memberId The fenced member id.
+ * @return A boolean indicating whether it's valid to online downgrade the consumer group.
+ */
+ private 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)) {
Review Comment:
Could we use `numMembers() - numClassicProtocolMembers() <= 1`? I think that we know that the remaining member is the one using the consumer group protocol.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -1440,9 +1508,20 @@ private CoordinatorResult<ConsumerGroupHeartbeatResponseData, Record> consumerGr
records = consumerGroupFenceMember(group, member);
}
}
- return new CoordinatorResult<>(records, new ConsumerGroupHeartbeatResponseData()
- .setMemberId(memberId)
- .setMemberEpoch(memberEpoch));
+
+ CompletableFuture<Void> appendFuture = null;
+ if ((instanceId == null || memberEpoch != LEAVE_GROUP_STATIC_MEMBER_EPOCH) &&
Review Comment:
Instead of repeating the conditions, could we use a boolean that we set in the relevant places?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -777,11 +778,78 @@ public ClassicGroup classicGroup(
}
}
+ /**
+ * Validates the online downgrade if a consumer member is fenced from the consumer group.
+ *
+ * @param consumerGroup The ConsumerGroup.
+ * @param memberId The fenced member id.
+ * @return A boolean indicating whether it's valid to online downgrade the consumer group.
+ */
+ private 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 = ClassicGroup.fromConsumerGroup(
+ consumerGroup,
+ leavingMemberId,
+ logContext,
+ time,
+ metrics,
+ consumerGroupSessionTimeoutMs,
+ metadataImage
+ );
+ } 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()));
+ }
+ classicGroup.createConsumerGroupRecords(metadataImage.features().metadataVersion(), records);
Review Comment:
nit: `createClassicGroupRecords`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16554: Online downgrade triggering and group type conversion [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #15721:
URL: https://github.com/apache/kafka/pull/15721#discussion_r1576113492
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -1422,10 +1499,16 @@ 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)) {
+ records = new ArrayList<>();
+ appendFuture = convertToClassicGroup(group, memberId, records);
+ } else {
+ records = consumerGroupFenceMember(group, member);
+ }
Review Comment:
Having this duplicated code in multiple places is a tad annoying, don't you think? I wonder if we could push it down into `consumerGroupFenceMember` as it seems that the downgrade must be checked whenever `consumerGroupFenceMember` is called. `consumerGroupFenceMember` could return a future or null and take the records as an argument.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16554: Online downgrade triggering and group type conversion [kafka]
Posted by "dongnuo123 (via GitHub)" <gi...@apache.org>.
dongnuo123 commented on code in PR #15721:
URL: https://github.com/apache/kafka/pull/15721#discussion_r1569413973
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -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);
Review Comment:
Ah yes. Thanks for the catch!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16554: Online downgrade triggering and group type conversion [kafka]
Posted by "dongnuo123 (via GitHub)" <gi...@apache.org>.
dongnuo123 commented on code in PR #15721:
URL: https://github.com/apache/kafka/pull/15721#discussion_r1571520271
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/Assertions.java:
##########
@@ -198,6 +204,62 @@ private static void assertApiMessageAndVersionEquals(
}
}
}
+ } else if (actual.message() instanceof GroupMetadataValue) {
Review Comment:
Yeah it's more convenient. Let me change this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16554: Online downgrade triggering and group type conversion [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #15721:
URL: https://github.com/apache/kafka/pull/15721#discussion_r1568737120
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -775,6 +777,126 @@ 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;
Review Comment:
nit: Let's add an empty line before this one.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -775,6 +777,126 @@ 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) {
Review Comment:
Does it need to be public? Let's add some javadoc please.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -9623,6 +10332,186 @@ public void testClassicGroupOnUnloadedCompletingRebalance() throws Exception {
.setErrorCode(NOT_COORDINATOR.code()), pendingMemberSyncResult.syncFuture.get());
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testLastClassicProtocolMemberLeavingConsumerGroup(boolean appendLogSuccessfully) {
Review Comment:
Should we also test the session expiration path and the rebalance expiration path?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java:
##########
@@ -998,4 +1094,232 @@ public ConsumerGroupDescribeResponseData.DescribedGroup asDescribedGroup(
);
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;
+ }
+
+ /**
+ * 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(),
Review Comment:
I think that we could use `Optional.empty()` here because a leader will be picked when the members are added.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -775,6 +777,126 @@ 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()));
Review Comment:
I am not sure to understand why we do this here. In my opinion, it is better to create all the state immediately. However, if t != null, I think that we need to revert changes made by `onClassicGroupStateTransition` at L821.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -775,6 +777,126 @@ 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(
Review Comment:
I wonder if we should rather follower the same structure than we used in this other PR. I mean that we could have a static method in ClassicGroup and we could have a createClassicGroupRecords. What do you think?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -1369,11 +1493,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<>(
Review Comment:
Could we extract the code from both branches and have it here?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java:
##########
@@ -998,4 +1094,232 @@ public ConsumerGroupDescribeResponseData.DescribedGroup asDescribedGroup(
);
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;
+ }
+
+ /**
+ * 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(),
Review Comment:
This does not seem correct. The issue is that the stored protocols were stored when the group was converted so they are outdated here. I wonder whether we should generate them with the minimal informations (e.g. the subscribed topics).
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java:
##########
@@ -998,4 +1094,232 @@ public ConsumerGroupDescribeResponseData.DescribedGroup asDescribedGroup(
);
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;
+ }
+
+ /**
+ * 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("")))
+ )
Review Comment:
This one maybe worth a comment explaining why we need to use this version.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -9623,6 +10332,186 @@ public void testClassicGroupOnUnloadedCompletingRebalance() throws Exception {
.setErrorCode(NOT_COORDINATOR.code()), pendingMemberSyncResult.syncFuture.get());
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testLastClassicProtocolMemberLeavingConsumerGroup(boolean appendLogSuccessfully) {
+ String groupId = "group-id";
+ String memberId1 = Uuid.randomUuid().toString();
+ String memberId2 = Uuid.randomUuid().toString();
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+ Uuid barTopicId = Uuid.randomUuid();
+ String barTopicName = "bar";
+ Uuid zarTopicId = Uuid.randomUuid();
+ String zarTopicName = "zar";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = Collections.singletonList(
+ new ConsumerGroupMemberMetadataValue.ClassicProtocol()
+ .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),
+ new TopicPartition(fooTopicName, 2),
+ new TopicPartition(barTopicName, 0),
+ new TopicPartition(barTopicName, 1)
+ )
+ ))))
+ );
+
+ ConsumerGroupMember member1 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setServerAssignorName("range")
+ .setClassicMemberMetadata(new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(protocols))
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .build();
+ ConsumerGroupMember member2 = new ConsumerGroupMember.Builder(memberId2)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ // Use zar only here to ensure that metadata needs to be recomputed.
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar", "zar"))
+ .setServerAssignorName("range")
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .build();
+
+ // Consumer group with two members.
+ // Member 1 uses the classic protocol and member 2 uses the consumer protocol.
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.DOWNGRADE)
+ .withAssignors(Collections.singletonList(assignor))
+ .withMetadataImage(new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .addTopic(barTopicId, barTopicName, 3)
+ .addTopic(zarTopicId, zarTopicName, 1)
+ .addRacks()
+ .build())
+ .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
+ .withMember(member1)
+ .withMember(member2)
+ .withAssignment(memberId1, mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .withAssignment(memberId2, mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .withAssignmentEpoch(10))
+ .build();
+
+ context.commit();
+ ConsumerGroup consumerGroup = context.groupMetadataManager.consumerGroup(groupId);
+
+ // Member 2 leaves the consumer group, triggering the downgrade.
+ CoordinatorResult<ConsumerGroupHeartbeatResponseData, Record> result = context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId2)
+ .setMemberEpoch(LEAVE_GROUP_MEMBER_EPOCH)
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setTopicPartitions(Collections.emptyList()));
+
+
+ byte[] assignment = Utils.toArray(ConsumerProtocol.serializeAssignment(new ConsumerPartitionAssignor.Assignment(Arrays.asList(
+ new TopicPartition(fooTopicName, 0), new TopicPartition(fooTopicName, 1), new TopicPartition(fooTopicName, 2),
+ new TopicPartition(barTopicName, 0), new TopicPartition(barTopicName, 1)
+ ))));
+ Map<String, byte[]> assignments = new HashMap<String, byte[]>() {
+ {
+ put(memberId1, assignment);
+ }
+ };
+
+ ClassicGroup expectedClassicGroup = new ClassicGroup(
+ new LogContext(),
+ groupId,
+ STABLE,
+ context.time,
+ context.metrics,
+ 10,
+ Optional.ofNullable(ConsumerProtocol.PROTOCOL_TYPE),
+ Optional.ofNullable("range"),
+ Optional.ofNullable(memberId1),
+ Optional.of(context.time.milliseconds())
+ );
+ expectedClassicGroup.add(
+ new ClassicGroupMember(
+ memberId1,
+ Optional.ofNullable(member1.instanceId()),
+ member1.clientId(),
+ member1.clientHost(),
+ member1.rebalanceTimeoutMs(),
+ 45000,
+ ConsumerProtocol.PROTOCOL_TYPE,
+ member1.supportedJoinGroupRequestProtocols(),
+ assignment
+ )
+ );
+
+ List<Record> expectedRecords = Arrays.asList(
+ RecordHelpers.newCurrentAssignmentTombstoneRecord(groupId, memberId2),
+ RecordHelpers.newTargetAssignmentTombstoneRecord(groupId, memberId2),
+ RecordHelpers.newMemberSubscriptionTombstoneRecord(groupId, memberId2),
+ // Subscription metadata is recomputed because zar is no longer there.
+ RecordHelpers.newGroupSubscriptionMetadataRecord(groupId, new HashMap<String, TopicMetadata>() {
+ {
+ put(fooTopicName, new TopicMetadata(fooTopicId, fooTopicName, 6, mkMapOfPartitionRacks(6)));
+ put(barTopicName, new TopicMetadata(barTopicId, barTopicName, 3, mkMapOfPartitionRacks(3)));
+ }
+ }),
+ RecordHelpers.newGroupEpochRecord(groupId, 11),
+
+ RecordHelpers.newTargetAssignmentEpochTombstoneRecord(groupId),
+ RecordHelpers.newGroupSubscriptionMetadataTombstoneRecord(groupId),
+ RecordHelpers.newGroupEpochTombstoneRecord(groupId),
+ RecordHelpers.newGroupMetadataRecord(expectedClassicGroup, assignments, MetadataVersion.latestTesting())
+ );
+
+ assertRecordsEquals(expectedRecords, result.records());
+
+ if (appendLogSuccessfully) {
+ ClassicGroup classicGroup = context.groupMetadataManager.getOrMaybeCreateClassicGroup(groupId, false);
+
+ // Simulate a successful write to the log.
+ result.appendFuture().complete(null);
+
+ ScheduledTimeout<Void, Record> timeout = context.timer.timeout(
+ classicGroupHeartbeatKey(groupId, memberId1));
+ assertNotNull(timeout);
+
+ // A new rebalance is triggered.
+ assertTrue(classicGroup.isInState(PREPARING_REBALANCE));
+ } else {
+ // Simulate a failed write to the log.
+ result.appendFuture().completeExceptionally(new NotLeaderOrFollowerException());
+ context.rollback();
+
+ ScheduledTimeout<Void, Record> timeout = context.timer.timeout(
+ classicGroupHeartbeatKey(groupId, memberId1));
+ assertNull(timeout);
+
+ // The group is reverted back to the consumer group.
+ assertEquals(consumerGroup, context.groupMetadataManager.consumerGroup(groupId));
+ }
Review Comment:
We should also verify the metrics as we had code related the group conversion.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java:
##########
@@ -998,4 +1094,232 @@ public ConsumerGroupDescribeResponseData.DescribedGroup asDescribedGroup(
);
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;
+ }
+
+ /**
+ * 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()));
Review Comment:
nit: Style.
```
records.add(RecordHelpers.newGroupMetadataRecord(
classicGroup,
assignments,
metadataImage.features().metadataVersion()
));
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16554: Online downgrade triggering and group type conversion [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #15721:
URL: https://github.com/apache/kafka/pull/15721#discussion_r1577407602
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/Assertions.java:
##########
@@ -198,6 +207,43 @@ private static void assertApiMessageAndVersionEquals(
}
}
}
+ } else if (actual.message() instanceof GroupMetadataValue) {
+ GroupMetadataValue expectedValue = (GroupMetadataValue) expected.message().duplicate();
+ GroupMetadataValue actualValue = (GroupMetadataValue) actual.message().duplicate();
+
+ expectedValue.members().sort(Comparator.comparing(GroupMetadataValue.MemberMetadata::memberId));
+ actualValue.members().sort(Comparator.comparing(GroupMetadataValue.MemberMetadata::memberId));
+ try {
+ Arrays.asList(expectedValue, actualValue).forEach(value ->
+ value.members().forEach(memberMetadata -> {
+ // Sort topics and ownedPartitions in Subscription.
+ ConsumerPartitionAssignor.Subscription subscription =
+ ConsumerProtocol.deserializeSubscription(ByteBuffer.wrap(memberMetadata.subscription()));
+ subscription.topics().sort(String::compareTo);
+ subscription.ownedPartitions().sort(
+ Comparator.comparing(TopicPartition::topic).thenComparing(TopicPartition::partition)
+ );
+ memberMetadata.setSubscription(Utils.toArray(ConsumerProtocol.serializeSubscription(
+ subscription,
+ ConsumerProtocol.deserializeVersion(ByteBuffer.wrap(memberMetadata.subscription()))
+ )));
+
+ // Sort partitions in Assignment.
+ ConsumerPartitionAssignor.Assignment assignment =
+ ConsumerProtocol.deserializeAssignment(ByteBuffer.wrap(memberMetadata.assignment()));
+ assignment.partitions().sort(
+ Comparator.comparing(TopicPartition::topic).thenComparing(TopicPartition::partition)
+ );
+ memberMetadata.setAssignment(Utils.toArray(ConsumerProtocol.serializeAssignment(
+ assignment,
+ ConsumerProtocol.deserializeVersion(ByteBuffer.wrap(memberMetadata.assignment()))
+ )));
+ })
+ );
+ } catch (SchemaException ex) {
+ System.out.println("Failed deserialization: " + ex.getMessage());
Review Comment:
nit: Sorry, I missed this one. We usually don't keep `println` in tests. It may be better to call `fail` with an appropriate message.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16554: Online downgrade triggering and group type conversion [kafka]
Posted by "dongnuo123 (via GitHub)" <gi...@apache.org>.
dongnuo123 commented on code in PR #15721:
URL: https://github.com/apache/kafka/pull/15721#discussion_r1568962281
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -775,6 +777,126 @@ 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()));
Review Comment:
If creating all the state immediately, we'll get error when replaying the old ConsumerGroup tombstone because `group.get(groupId)` has become a ClassicGroup. We can't rely on replaying the records to update the states either, because we need the new classicGroup reference to trigger the rebalance. so the only way is not to replay the records by setting the appendFuture.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16554: Online downgrade triggering and group type conversion [kafka]
Posted by "dongnuo123 (via GitHub)" <gi...@apache.org>.
dongnuo123 commented on code in PR #15721:
URL: https://github.com/apache/kafka/pull/15721#discussion_r1568962281
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -775,6 +777,126 @@ 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()));
Review Comment:
If creating all the state immediately, we'll get error when replaying the old ConsumerGroup tombstone because `group.get(groupId)` has become a ClassicGroup. We can't rely on replaying the records to update the states either, because we need the new classicGroup reference to trigger the rebalance. so the only way is not to replay the records.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16554: Online downgrade triggering and group type conversion [kafka]
Posted by "dongnuo123 (via GitHub)" <gi...@apache.org>.
dongnuo123 commented on code in PR #15721:
URL: https://github.com/apache/kafka/pull/15721#discussion_r1572412240
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/Assertions.java:
##########
@@ -198,6 +206,40 @@ private static void assertApiMessageAndVersionEquals(
}
}
}
+ } else if (actual.message() instanceof GroupMetadataValue) {
+ GroupMetadataValue expectedValue = (GroupMetadataValue) expected.message().duplicate();
+ GroupMetadataValue actualValue = (GroupMetadataValue) actual.message().duplicate();
+
+ expectedValue.members().sort(Comparator.comparing(GroupMetadataValue.MemberMetadata::memberId));
+ actualValue.members().sort(Comparator.comparing(GroupMetadataValue.MemberMetadata::memberId));
+
+ Arrays.asList(expectedValue, actualValue).forEach(value ->
+ value.members().forEach(memberMetadata -> {
+ // Sort topics and ownedPartitions in Subscription.
+ ConsumerPartitionAssignor.Subscription subscription =
+ ConsumerProtocol.deserializeSubscription(ByteBuffer.wrap(memberMetadata.subscription()));
Review Comment:
Handle the schema exception
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16554: Online downgrade triggering and group type conversion [kafka]
Posted by "dongnuo123 (via GitHub)" <gi...@apache.org>.
dongnuo123 commented on code in PR #15721:
URL: https://github.com/apache/kafka/pull/15721#discussion_r1574897141
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -10331,6 +10333,461 @@ public void testClassicGroupOnUnloadedCompletingRebalance() throws Exception {
.setErrorCode(NOT_COORDINATOR.code()), pendingMemberSyncResult.syncFuture.get());
}
+ @Test
+ public void testLastClassicProtocolMemberLeavingConsumerGroup() {
+ String groupId = "group-id";
+ String memberId1 = Uuid.randomUuid().toString();
+ String memberId2 = Uuid.randomUuid().toString();
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+ Uuid barTopicId = Uuid.randomUuid();
+ String barTopicName = "bar";
+ Uuid zarTopicId = Uuid.randomUuid();
+ String zarTopicName = "zar";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = Collections.singletonList(
+ new ConsumerGroupMemberMetadataValue.ClassicProtocol()
+ .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),
+ new TopicPartition(fooTopicName, 2),
+ new TopicPartition(barTopicName, 0),
+ new TopicPartition(barTopicName, 1)
+ )
+ ))))
+ );
+
+ ConsumerGroupMember member1 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setClassicMemberMetadata(new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(protocols))
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .build();
+ ConsumerGroupMember member2 = new ConsumerGroupMember.Builder(memberId2)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ // Use zar only here to ensure that metadata needs to be recomputed.
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar", "zar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .build();
+
+ // Consumer group with two members.
+ // Member 1 uses the classic protocol and member 2 uses the consumer protocol.
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.DOWNGRADE)
+ .withAssignors(Collections.singletonList(assignor))
+ .withMetadataImage(new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .addTopic(barTopicId, barTopicName, 3)
+ .addTopic(zarTopicId, zarTopicName, 1)
+ .addRacks()
+ .build())
+ .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
+ .withMember(member1)
+ .withMember(member2)
+ .withAssignment(memberId1, mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .withAssignment(memberId2, mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .withAssignmentEpoch(10))
+ .build();
+
+ context.commit();
+ ConsumerGroup consumerGroup = context.groupMetadataManager.consumerGroup(groupId);
+
+ // Member 2 leaves the consumer group, triggering the downgrade.
+ CoordinatorResult<ConsumerGroupHeartbeatResponseData, Record> result = context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId2)
+ .setMemberEpoch(LEAVE_GROUP_MEMBER_EPOCH)
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setTopicPartitions(Collections.emptyList()));
+
+
+ byte[] assignment = Utils.toArray(ConsumerProtocol.serializeAssignment(new ConsumerPartitionAssignor.Assignment(Arrays.asList(
+ new TopicPartition(fooTopicName, 0), new TopicPartition(fooTopicName, 1), new TopicPartition(fooTopicName, 2),
+ new TopicPartition(barTopicName, 0), new TopicPartition(barTopicName, 1)
+ ))));
+ Map<String, byte[]> assignments = new HashMap<String, byte[]>() {
+ {
+ put(memberId1, assignment);
+ }
+ };
+
+ ClassicGroup expectedClassicGroup = new ClassicGroup(
+ new LogContext(),
+ groupId,
+ STABLE,
+ context.time,
+ context.metrics,
+ 10,
+ Optional.ofNullable(ConsumerProtocol.PROTOCOL_TYPE),
+ Optional.ofNullable("range"),
+ Optional.ofNullable(memberId1),
+ Optional.of(context.time.milliseconds())
+ );
+ expectedClassicGroup.add(
+ new ClassicGroupMember(
+ memberId1,
+ Optional.ofNullable(member1.instanceId()),
+ member1.clientId(),
+ member1.clientHost(),
+ member1.rebalanceTimeoutMs(),
+ 45000,
+ ConsumerProtocol.PROTOCOL_TYPE,
+ member1.supportedJoinGroupRequestProtocols(),
+ assignment
+ )
+ );
+
+ List<Record> expectedRecords = Arrays.asList(
+ RecordHelpers.newCurrentAssignmentTombstoneRecord(groupId, memberId2),
+ RecordHelpers.newTargetAssignmentTombstoneRecord(groupId, memberId2),
+ RecordHelpers.newMemberSubscriptionTombstoneRecord(groupId, memberId2),
+ // Subscription metadata is recomputed because zar is no longer there.
+ RecordHelpers.newGroupSubscriptionMetadataRecord(groupId, new HashMap<String, TopicMetadata>() {
+ {
+ put(fooTopicName, new TopicMetadata(fooTopicId, fooTopicName, 6, mkMapOfPartitionRacks(6)));
+ put(barTopicName, new TopicMetadata(barTopicId, barTopicName, 3, mkMapOfPartitionRacks(3)));
+ }
+ }),
+ RecordHelpers.newGroupEpochRecord(groupId, 11),
+
+ RecordHelpers.newTargetAssignmentEpochTombstoneRecord(groupId),
Review Comment:
Should we also add tombstones for all the members when deleting a group?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16554: Online downgrade triggering and group type conversion [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac merged PR #15721:
URL: https://github.com/apache/kafka/pull/15721
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16554: Online downgrade triggering and group type conversion [kafka]
Posted by "dongnuo123 (via GitHub)" <gi...@apache.org>.
dongnuo123 commented on code in PR #15721:
URL: https://github.com/apache/kafka/pull/15721#discussion_r1568963034
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -775,6 +777,126 @@ 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()));
Review Comment:
Yes we should revert `onClassicGroupStateTransition`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16554: Online downgrade triggering and group type conversion [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #15721:
URL: https://github.com/apache/kafka/pull/15721#discussion_r1569051446
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -775,6 +777,126 @@ 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()));
Review Comment:
I do agree with the `appendFuture` part of your explanation. However, I still believe that we should schedule the session timeouts and the start the rebalance before it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16554: Online downgrade triggering and group type conversion [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #15721:
URL: https://github.com/apache/kafka/pull/15721#discussion_r1572283723
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -777,11 +778,79 @@ public ClassicGroup classicGroup(
}
}
+ /**
+ * Validates the online downgrade if a consumer member is fenced from the consumer group.
+ *
+ * @param consumerGroup The ConsumerGroup.
+ * @param memberId The fenced member id.
+ * @return A boolean indicating whether it's valid to online downgrade the consumer group.
+ */
+ private 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.",
Review Comment:
nit: We could use `debug` for this one.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -777,11 +778,79 @@ public ClassicGroup classicGroup(
}
}
+ /**
+ * Validates the online downgrade if a consumer member is fenced from the consumer group.
+ *
+ * @param consumerGroup The ConsumerGroup.
+ * @param memberId The fenced member id.
+ * @return A boolean indicating whether it's valid to online downgrade the consumer group.
+ */
+ private 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;
+ }
+
+ private CompletableFuture<Void> convertToClassicGroup(ConsumerGroup consumerGroup, String leavingMemberId, List<Record> records) {
+ consumerGroup.createGroupTombstoneRecords(records);
+
+ ClassicGroup classicGroup;
+ try {
+ classicGroup = ClassicGroup.fromConsumerGroup(
+ consumerGroup,
+ leavingMemberId,
+ logContext,
+ time,
+ metrics,
+ consumerGroupSessionTimeoutMs,
+ metadataImage
+ );
+ } 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()));
+ }
+ classicGroup.createClassicGroupRecords(metadataImage.features().metadataVersion(), records);
+
+ // Directly update the states instead of replaying the records because
+ // the classicGroup reference is needed for triggering the rebalance.
+ // Set the appendFuture to prevent the records from being replayed.
+ removeGroup(consumerGroup.groupId());
+ groups.put(consumerGroup.groupId(), classicGroup);
+ metrics.onClassicGroupStateTransition(null, classicGroup.currentState());
+
+ classicGroup.allMembers().forEach(member -> rescheduleClassicGroupMemberHeartbeat(classicGroup, member));
+ prepareRebalance(classicGroup, String.format("Downgrade group %s.", classicGroup.groupId()));
Review Comment:
nit: `Downgrade group %s from consumer to classic.`.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -10331,6 +10333,461 @@ public void testClassicGroupOnUnloadedCompletingRebalance() throws Exception {
.setErrorCode(NOT_COORDINATOR.code()), pendingMemberSyncResult.syncFuture.get());
}
+ @Test
+ public void testLastClassicProtocolMemberLeavingConsumerGroup() {
Review Comment:
nit: Is the name correct? I thought that it is actually the last member using the consumer protocol leaving, no?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java:
##########
@@ -1212,9 +1212,12 @@ public boolean supportsClassicProtocols(String memberProtocolType, Set<String> m
}
/**
+ * @param memberId The member to remove.
Review Comment:
nit: We would usually have a first line to describe the method in this case.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -1421,16 +1490,19 @@ private CoordinatorResult<ConsumerGroupHeartbeatResponseData, Record> consumerGr
int memberEpoch
) throws ApiException {
ConsumerGroup group = consumerGroup(groupId);
+ boolean isDynamicMember = instanceId == null;
+ boolean isStaticMemberTemporaryLeave = memberEpoch == LEAVE_GROUP_STATIC_MEMBER_EPOCH;
Review Comment:
Hum... I thought that we could use a boolean like `maybeDowngrade` and set it to true in the relevant places. I think that it would be clearer than duplicating the conditions.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -10331,6 +10333,461 @@ public void testClassicGroupOnUnloadedCompletingRebalance() throws Exception {
.setErrorCode(NOT_COORDINATOR.code()), pendingMemberSyncResult.syncFuture.get());
}
+ @Test
+ public void testLastClassicProtocolMemberLeavingConsumerGroup() {
+ String groupId = "group-id";
+ String memberId1 = Uuid.randomUuid().toString();
+ String memberId2 = Uuid.randomUuid().toString();
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+ Uuid barTopicId = Uuid.randomUuid();
+ String barTopicName = "bar";
+ Uuid zarTopicId = Uuid.randomUuid();
+ String zarTopicName = "zar";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = Collections.singletonList(
+ new ConsumerGroupMemberMetadataValue.ClassicProtocol()
+ .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),
+ new TopicPartition(fooTopicName, 2),
+ new TopicPartition(barTopicName, 0),
+ new TopicPartition(barTopicName, 1)
+ )
+ ))))
+ );
+
+ ConsumerGroupMember member1 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setClassicMemberMetadata(new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(protocols))
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .build();
+ ConsumerGroupMember member2 = new ConsumerGroupMember.Builder(memberId2)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ // Use zar only here to ensure that metadata needs to be recomputed.
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar", "zar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .build();
+
+ // Consumer group with two members.
+ // Member 1 uses the classic protocol and member 2 uses the consumer protocol.
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.DOWNGRADE)
+ .withAssignors(Collections.singletonList(assignor))
+ .withMetadataImage(new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .addTopic(barTopicId, barTopicName, 3)
+ .addTopic(zarTopicId, zarTopicName, 1)
+ .addRacks()
+ .build())
+ .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
+ .withMember(member1)
+ .withMember(member2)
+ .withAssignment(memberId1, mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .withAssignment(memberId2, mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .withAssignmentEpoch(10))
+ .build();
+
+ context.commit();
+ ConsumerGroup consumerGroup = context.groupMetadataManager.consumerGroup(groupId);
+
+ // Member 2 leaves the consumer group, triggering the downgrade.
+ CoordinatorResult<ConsumerGroupHeartbeatResponseData, Record> result = context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId2)
+ .setMemberEpoch(LEAVE_GROUP_MEMBER_EPOCH)
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setTopicPartitions(Collections.emptyList()));
+
+
+ byte[] assignment = Utils.toArray(ConsumerProtocol.serializeAssignment(new ConsumerPartitionAssignor.Assignment(Arrays.asList(
+ new TopicPartition(fooTopicName, 0), new TopicPartition(fooTopicName, 1), new TopicPartition(fooTopicName, 2),
+ new TopicPartition(barTopicName, 0), new TopicPartition(barTopicName, 1)
+ ))));
+ Map<String, byte[]> assignments = new HashMap<String, byte[]>() {
+ {
+ put(memberId1, assignment);
+ }
+ };
+
+ ClassicGroup expectedClassicGroup = new ClassicGroup(
+ new LogContext(),
+ groupId,
+ STABLE,
+ context.time,
+ context.metrics,
+ 10,
+ Optional.ofNullable(ConsumerProtocol.PROTOCOL_TYPE),
+ Optional.ofNullable("range"),
+ Optional.ofNullable(memberId1),
+ Optional.of(context.time.milliseconds())
+ );
+ expectedClassicGroup.add(
+ new ClassicGroupMember(
+ memberId1,
+ Optional.ofNullable(member1.instanceId()),
+ member1.clientId(),
+ member1.clientHost(),
+ member1.rebalanceTimeoutMs(),
+ 45000,
+ ConsumerProtocol.PROTOCOL_TYPE,
+ member1.supportedJoinGroupRequestProtocols(),
+ assignment
+ )
+ );
+
+ List<Record> expectedRecords = Arrays.asList(
+ RecordHelpers.newCurrentAssignmentTombstoneRecord(groupId, memberId2),
+ RecordHelpers.newTargetAssignmentTombstoneRecord(groupId, memberId2),
+ RecordHelpers.newMemberSubscriptionTombstoneRecord(groupId, memberId2),
+ // Subscription metadata is recomputed because zar is no longer there.
+ RecordHelpers.newGroupSubscriptionMetadataRecord(groupId, new HashMap<String, TopicMetadata>() {
+ {
+ put(fooTopicName, new TopicMetadata(fooTopicId, fooTopicName, 6, mkMapOfPartitionRacks(6)));
+ put(barTopicName, new TopicMetadata(barTopicId, barTopicName, 3, mkMapOfPartitionRacks(3)));
+ }
+ }),
+ RecordHelpers.newGroupEpochRecord(groupId, 11),
+
+ RecordHelpers.newTargetAssignmentEpochTombstoneRecord(groupId),
+ RecordHelpers.newGroupSubscriptionMetadataTombstoneRecord(groupId),
+ RecordHelpers.newGroupEpochTombstoneRecord(groupId),
+ RecordHelpers.newGroupMetadataRecord(expectedClassicGroup, assignments, MetadataVersion.latestTesting())
+ );
+
+ assertRecordsEquals(expectedRecords, result.records());
+ verify(context.metrics, times(1)).onConsumerGroupStateTransition(ConsumerGroup.ConsumerGroupState.STABLE, null);
+ verify(context.metrics, times(1)).onClassicGroupStateTransition(null, STABLE);
+
+ // The new classic member 1 has a heartbeat timeout.
+ ScheduledTimeout<Void, Record> heartbeatTimeout = context.timer.timeout(
+ classicGroupHeartbeatKey(groupId, memberId1));
+ assertNotNull(heartbeatTimeout);
+ // The new rebalance has a groupJoin timeout.
+ ScheduledTimeout<Void, Record> groupJoinTimeout = context.timer.timeout(
+ classicGroupJoinKey(groupId));
+ assertNotNull(groupJoinTimeout);
+
+ // A new rebalance is triggered.
+ ClassicGroup classicGroup = context.groupMetadataManager.getOrMaybeCreateClassicGroup(groupId, false);
+ assertTrue(classicGroup.isInState(PREPARING_REBALANCE));
+
+ // Simulate a failed write to the log.
+ result.appendFuture().completeExceptionally(new NotLeaderOrFollowerException());
+ context.rollback();
+
+ // The group is reverted back to the consumer group.
+ assertEquals(consumerGroup, context.groupMetadataManager.consumerGroup(groupId));
+ verify(context.metrics, times(1)).onClassicGroupStateTransition(PREPARING_REBALANCE, null);
+ }
+
+ @Test
+ public void testLastClassicProtocolMemberSessionTimeoutInConsumerGroup() {
+ String groupId = "group-id";
+ String memberId1 = Uuid.randomUuid().toString();
+ String memberId2 = Uuid.randomUuid().toString();
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+ Uuid barTopicId = Uuid.randomUuid();
+ String barTopicName = "bar";
+ Uuid zarTopicId = Uuid.randomUuid();
+ String zarTopicName = "zar";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = Collections.singletonList(
+ new ConsumerGroupMemberMetadataValue.ClassicProtocol()
+ .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),
+ new TopicPartition(fooTopicName, 2),
+ new TopicPartition(barTopicName, 0),
+ new TopicPartition(barTopicName, 1)
+ )
+ ))))
+ );
+
+ ConsumerGroupMember member1 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setClassicMemberMetadata(new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(protocols))
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .build();
+ ConsumerGroupMember member2 = new ConsumerGroupMember.Builder(memberId2)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar", "zar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .build();
+
+ // Consumer group with two members.
+ // Member 1 uses the classic protocol and member 2 uses the consumer protocol.
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.DOWNGRADE)
+ .withAssignors(Collections.singletonList(assignor))
+ .withMetadataImage(new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .addTopic(barTopicId, barTopicName, 3)
+ .addTopic(zarTopicId, zarTopicName, 1)
+ .addRacks()
+ .build())
+ .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
+ .withMember(member1)
+ .withMember(member2)
+ .withAssignment(memberId1, mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .withAssignment(memberId2, mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .withAssignmentEpoch(10))
+ .build();
+
+ context.replay(RecordHelpers.newGroupSubscriptionMetadataRecord(groupId, new HashMap<String, TopicMetadata>() {
+ {
+ put(fooTopicName, new TopicMetadata(fooTopicId, fooTopicName, 6, mkMapOfPartitionRacks(6)));
+ put(barTopicName, new TopicMetadata(barTopicId, barTopicName, 3, mkMapOfPartitionRacks(3)));
+ put(zarTopicName, new TopicMetadata(zarTopicId, zarTopicName, 1, mkMapOfPartitionRacks(1)));
+ }
+ }));
+
+ context.commit();
+
+ // Session timer is scheduled on the heartbeat.
+ context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId2)
+ .setMemberEpoch(10)
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar", "zar"))
+ .setTopicPartitions(Collections.emptyList()));
+
+ // Verify that there is a session timeout.
+ context.assertSessionTimeout(groupId, memberId2, 45000);
+
+ // Advance time past the session timeout.
+ // Member 2 should be fenced from the group, thus triggering the downgrade.
+ context.sleep(45000 + 1);
+
+ verify(context.metrics, times(1)).onConsumerGroupStateTransition(ConsumerGroup.ConsumerGroupState.STABLE, null);
+ verify(context.metrics, times(1)).onClassicGroupStateTransition(null, STABLE);
+
+ // The new classic member 1 has a heartbeat timeout.
+ ScheduledTimeout<Void, Record> heartbeatTimeout = context.timer.timeout(
+ classicGroupHeartbeatKey(groupId, memberId1));
+ assertNotNull(heartbeatTimeout);
+ // The new rebalance has a groupJoin timeout.
+ ScheduledTimeout<Void, Record> groupJoinTimeout = context.timer.timeout(
+ classicGroupJoinKey(groupId));
+ assertNotNull(groupJoinTimeout);
+
+ // A new rebalance is triggered.
+ ClassicGroup classicGroup = context.groupMetadataManager.getOrMaybeCreateClassicGroup(groupId, false);
+ assertTrue(classicGroup.isInState(PREPARING_REBALANCE));
+ }
+
+ @Test
+ public void testLastClassicProtocolMemberRebalanceTimeoutInConsumerGroup() {
Review Comment:
ditto about the name.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -10331,6 +10333,461 @@ public void testClassicGroupOnUnloadedCompletingRebalance() throws Exception {
.setErrorCode(NOT_COORDINATOR.code()), pendingMemberSyncResult.syncFuture.get());
}
+ @Test
+ public void testLastClassicProtocolMemberLeavingConsumerGroup() {
+ String groupId = "group-id";
+ String memberId1 = Uuid.randomUuid().toString();
+ String memberId2 = Uuid.randomUuid().toString();
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+ Uuid barTopicId = Uuid.randomUuid();
+ String barTopicName = "bar";
+ Uuid zarTopicId = Uuid.randomUuid();
+ String zarTopicName = "zar";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = Collections.singletonList(
+ new ConsumerGroupMemberMetadataValue.ClassicProtocol()
+ .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),
+ new TopicPartition(fooTopicName, 2),
+ new TopicPartition(barTopicName, 0),
+ new TopicPartition(barTopicName, 1)
+ )
+ ))))
+ );
+
+ ConsumerGroupMember member1 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setClassicMemberMetadata(new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(protocols))
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .build();
+ ConsumerGroupMember member2 = new ConsumerGroupMember.Builder(memberId2)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ // Use zar only here to ensure that metadata needs to be recomputed.
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar", "zar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .build();
+
+ // Consumer group with two members.
+ // Member 1 uses the classic protocol and member 2 uses the consumer protocol.
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.DOWNGRADE)
+ .withAssignors(Collections.singletonList(assignor))
+ .withMetadataImage(new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .addTopic(barTopicId, barTopicName, 3)
+ .addTopic(zarTopicId, zarTopicName, 1)
+ .addRacks()
+ .build())
+ .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
+ .withMember(member1)
+ .withMember(member2)
+ .withAssignment(memberId1, mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .withAssignment(memberId2, mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .withAssignmentEpoch(10))
+ .build();
+
+ context.commit();
+ ConsumerGroup consumerGroup = context.groupMetadataManager.consumerGroup(groupId);
+
+ // Member 2 leaves the consumer group, triggering the downgrade.
+ CoordinatorResult<ConsumerGroupHeartbeatResponseData, Record> result = context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId2)
+ .setMemberEpoch(LEAVE_GROUP_MEMBER_EPOCH)
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setTopicPartitions(Collections.emptyList()));
+
+
+ byte[] assignment = Utils.toArray(ConsumerProtocol.serializeAssignment(new ConsumerPartitionAssignor.Assignment(Arrays.asList(
+ new TopicPartition(fooTopicName, 0), new TopicPartition(fooTopicName, 1), new TopicPartition(fooTopicName, 2),
+ new TopicPartition(barTopicName, 0), new TopicPartition(barTopicName, 1)
+ ))));
+ Map<String, byte[]> assignments = new HashMap<String, byte[]>() {
+ {
+ put(memberId1, assignment);
+ }
+ };
+
+ ClassicGroup expectedClassicGroup = new ClassicGroup(
+ new LogContext(),
+ groupId,
+ STABLE,
+ context.time,
+ context.metrics,
+ 10,
+ Optional.ofNullable(ConsumerProtocol.PROTOCOL_TYPE),
+ Optional.ofNullable("range"),
+ Optional.ofNullable(memberId1),
+ Optional.of(context.time.milliseconds())
+ );
+ expectedClassicGroup.add(
+ new ClassicGroupMember(
+ memberId1,
+ Optional.ofNullable(member1.instanceId()),
+ member1.clientId(),
+ member1.clientHost(),
+ member1.rebalanceTimeoutMs(),
+ 45000,
+ ConsumerProtocol.PROTOCOL_TYPE,
+ member1.supportedJoinGroupRequestProtocols(),
+ assignment
+ )
+ );
+
+ List<Record> expectedRecords = Arrays.asList(
+ RecordHelpers.newCurrentAssignmentTombstoneRecord(groupId, memberId2),
+ RecordHelpers.newTargetAssignmentTombstoneRecord(groupId, memberId2),
+ RecordHelpers.newMemberSubscriptionTombstoneRecord(groupId, memberId2),
+ // Subscription metadata is recomputed because zar is no longer there.
+ RecordHelpers.newGroupSubscriptionMetadataRecord(groupId, new HashMap<String, TopicMetadata>() {
+ {
+ put(fooTopicName, new TopicMetadata(fooTopicId, fooTopicName, 6, mkMapOfPartitionRacks(6)));
+ put(barTopicName, new TopicMetadata(barTopicId, barTopicName, 3, mkMapOfPartitionRacks(3)));
+ }
+ }),
+ RecordHelpers.newGroupEpochRecord(groupId, 11),
+
+ RecordHelpers.newTargetAssignmentEpochTombstoneRecord(groupId),
+ RecordHelpers.newGroupSubscriptionMetadataTombstoneRecord(groupId),
+ RecordHelpers.newGroupEpochTombstoneRecord(groupId),
+ RecordHelpers.newGroupMetadataRecord(expectedClassicGroup, assignments, MetadataVersion.latestTesting())
+ );
+
+ assertRecordsEquals(expectedRecords, result.records());
+ verify(context.metrics, times(1)).onConsumerGroupStateTransition(ConsumerGroup.ConsumerGroupState.STABLE, null);
+ verify(context.metrics, times(1)).onClassicGroupStateTransition(null, STABLE);
+
+ // The new classic member 1 has a heartbeat timeout.
+ ScheduledTimeout<Void, Record> heartbeatTimeout = context.timer.timeout(
+ classicGroupHeartbeatKey(groupId, memberId1));
+ assertNotNull(heartbeatTimeout);
+ // The new rebalance has a groupJoin timeout.
+ ScheduledTimeout<Void, Record> groupJoinTimeout = context.timer.timeout(
+ classicGroupJoinKey(groupId));
+ assertNotNull(groupJoinTimeout);
+
+ // A new rebalance is triggered.
+ ClassicGroup classicGroup = context.groupMetadataManager.getOrMaybeCreateClassicGroup(groupId, false);
+ assertTrue(classicGroup.isInState(PREPARING_REBALANCE));
+
+ // Simulate a failed write to the log.
+ result.appendFuture().completeExceptionally(new NotLeaderOrFollowerException());
+ context.rollback();
+
+ // The group is reverted back to the consumer group.
+ assertEquals(consumerGroup, context.groupMetadataManager.consumerGroup(groupId));
+ verify(context.metrics, times(1)).onClassicGroupStateTransition(PREPARING_REBALANCE, null);
+ }
+
+ @Test
+ public void testLastClassicProtocolMemberSessionTimeoutInConsumerGroup() {
+ String groupId = "group-id";
+ String memberId1 = Uuid.randomUuid().toString();
+ String memberId2 = Uuid.randomUuid().toString();
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+ Uuid barTopicId = Uuid.randomUuid();
+ String barTopicName = "bar";
+ Uuid zarTopicId = Uuid.randomUuid();
+ String zarTopicName = "zar";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = Collections.singletonList(
+ new ConsumerGroupMemberMetadataValue.ClassicProtocol()
+ .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),
+ new TopicPartition(fooTopicName, 2),
+ new TopicPartition(barTopicName, 0),
+ new TopicPartition(barTopicName, 1)
+ )
+ ))))
+ );
+
+ ConsumerGroupMember member1 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setClassicMemberMetadata(new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(protocols))
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .build();
+ ConsumerGroupMember member2 = new ConsumerGroupMember.Builder(memberId2)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar", "zar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .build();
+
+ // Consumer group with two members.
+ // Member 1 uses the classic protocol and member 2 uses the consumer protocol.
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.DOWNGRADE)
+ .withAssignors(Collections.singletonList(assignor))
+ .withMetadataImage(new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .addTopic(barTopicId, barTopicName, 3)
+ .addTopic(zarTopicId, zarTopicName, 1)
+ .addRacks()
+ .build())
+ .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
+ .withMember(member1)
+ .withMember(member2)
+ .withAssignment(memberId1, mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .withAssignment(memberId2, mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .withAssignmentEpoch(10))
+ .build();
+
+ context.replay(RecordHelpers.newGroupSubscriptionMetadataRecord(groupId, new HashMap<String, TopicMetadata>() {
+ {
+ put(fooTopicName, new TopicMetadata(fooTopicId, fooTopicName, 6, mkMapOfPartitionRacks(6)));
+ put(barTopicName, new TopicMetadata(barTopicId, barTopicName, 3, mkMapOfPartitionRacks(3)));
+ put(zarTopicName, new TopicMetadata(zarTopicId, zarTopicName, 1, mkMapOfPartitionRacks(1)));
+ }
+ }));
+
+ context.commit();
+
+ // Session timer is scheduled on the heartbeat.
+ context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId2)
+ .setMemberEpoch(10)
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar", "zar"))
+ .setTopicPartitions(Collections.emptyList()));
+
+ // Verify that there is a session timeout.
+ context.assertSessionTimeout(groupId, memberId2, 45000);
+
+ // Advance time past the session timeout.
+ // Member 2 should be fenced from the group, thus triggering the downgrade.
+ context.sleep(45000 + 1);
+
+ verify(context.metrics, times(1)).onConsumerGroupStateTransition(ConsumerGroup.ConsumerGroupState.STABLE, null);
+ verify(context.metrics, times(1)).onClassicGroupStateTransition(null, STABLE);
+
+ // The new classic member 1 has a heartbeat timeout.
+ ScheduledTimeout<Void, Record> heartbeatTimeout = context.timer.timeout(
+ classicGroupHeartbeatKey(groupId, memberId1));
+ assertNotNull(heartbeatTimeout);
+ // The new rebalance has a groupJoin timeout.
+ ScheduledTimeout<Void, Record> groupJoinTimeout = context.timer.timeout(
+ classicGroupJoinKey(groupId));
Review Comment:
ditto.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -10331,6 +10333,461 @@ public void testClassicGroupOnUnloadedCompletingRebalance() throws Exception {
.setErrorCode(NOT_COORDINATOR.code()), pendingMemberSyncResult.syncFuture.get());
}
+ @Test
+ public void testLastClassicProtocolMemberLeavingConsumerGroup() {
+ String groupId = "group-id";
+ String memberId1 = Uuid.randomUuid().toString();
+ String memberId2 = Uuid.randomUuid().toString();
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+ Uuid barTopicId = Uuid.randomUuid();
+ String barTopicName = "bar";
+ Uuid zarTopicId = Uuid.randomUuid();
+ String zarTopicName = "zar";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = Collections.singletonList(
+ new ConsumerGroupMemberMetadataValue.ClassicProtocol()
+ .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),
+ new TopicPartition(fooTopicName, 2),
+ new TopicPartition(barTopicName, 0),
+ new TopicPartition(barTopicName, 1)
+ )
+ ))))
+ );
+
+ ConsumerGroupMember member1 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setClassicMemberMetadata(new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(protocols))
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .build();
+ ConsumerGroupMember member2 = new ConsumerGroupMember.Builder(memberId2)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ // Use zar only here to ensure that metadata needs to be recomputed.
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar", "zar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .build();
+
+ // Consumer group with two members.
+ // Member 1 uses the classic protocol and member 2 uses the consumer protocol.
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.DOWNGRADE)
+ .withAssignors(Collections.singletonList(assignor))
+ .withMetadataImage(new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .addTopic(barTopicId, barTopicName, 3)
+ .addTopic(zarTopicId, zarTopicName, 1)
+ .addRacks()
+ .build())
+ .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
+ .withMember(member1)
+ .withMember(member2)
+ .withAssignment(memberId1, mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .withAssignment(memberId2, mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .withAssignmentEpoch(10))
+ .build();
+
+ context.commit();
+ ConsumerGroup consumerGroup = context.groupMetadataManager.consumerGroup(groupId);
+
+ // Member 2 leaves the consumer group, triggering the downgrade.
+ CoordinatorResult<ConsumerGroupHeartbeatResponseData, Record> result = context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId2)
+ .setMemberEpoch(LEAVE_GROUP_MEMBER_EPOCH)
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setTopicPartitions(Collections.emptyList()));
+
+
+ byte[] assignment = Utils.toArray(ConsumerProtocol.serializeAssignment(new ConsumerPartitionAssignor.Assignment(Arrays.asList(
+ new TopicPartition(fooTopicName, 0), new TopicPartition(fooTopicName, 1), new TopicPartition(fooTopicName, 2),
+ new TopicPartition(barTopicName, 0), new TopicPartition(barTopicName, 1)
+ ))));
+ Map<String, byte[]> assignments = new HashMap<String, byte[]>() {
+ {
+ put(memberId1, assignment);
+ }
+ };
+
+ ClassicGroup expectedClassicGroup = new ClassicGroup(
+ new LogContext(),
+ groupId,
+ STABLE,
+ context.time,
+ context.metrics,
+ 10,
+ Optional.ofNullable(ConsumerProtocol.PROTOCOL_TYPE),
+ Optional.ofNullable("range"),
+ Optional.ofNullable(memberId1),
+ Optional.of(context.time.milliseconds())
+ );
+ expectedClassicGroup.add(
+ new ClassicGroupMember(
+ memberId1,
+ Optional.ofNullable(member1.instanceId()),
+ member1.clientId(),
+ member1.clientHost(),
+ member1.rebalanceTimeoutMs(),
+ 45000,
+ ConsumerProtocol.PROTOCOL_TYPE,
+ member1.supportedJoinGroupRequestProtocols(),
+ assignment
+ )
+ );
+
+ List<Record> expectedRecords = Arrays.asList(
+ RecordHelpers.newCurrentAssignmentTombstoneRecord(groupId, memberId2),
+ RecordHelpers.newTargetAssignmentTombstoneRecord(groupId, memberId2),
+ RecordHelpers.newMemberSubscriptionTombstoneRecord(groupId, memberId2),
+ // Subscription metadata is recomputed because zar is no longer there.
+ RecordHelpers.newGroupSubscriptionMetadataRecord(groupId, new HashMap<String, TopicMetadata>() {
+ {
+ put(fooTopicName, new TopicMetadata(fooTopicId, fooTopicName, 6, mkMapOfPartitionRacks(6)));
+ put(barTopicName, new TopicMetadata(barTopicId, barTopicName, 3, mkMapOfPartitionRacks(3)));
+ }
+ }),
+ RecordHelpers.newGroupEpochRecord(groupId, 11),
+
+ RecordHelpers.newTargetAssignmentEpochTombstoneRecord(groupId),
+ RecordHelpers.newGroupSubscriptionMetadataTombstoneRecord(groupId),
+ RecordHelpers.newGroupEpochTombstoneRecord(groupId),
+ RecordHelpers.newGroupMetadataRecord(expectedClassicGroup, assignments, MetadataVersion.latestTesting())
+ );
+
+ assertRecordsEquals(expectedRecords, result.records());
+ verify(context.metrics, times(1)).onConsumerGroupStateTransition(ConsumerGroup.ConsumerGroupState.STABLE, null);
+ verify(context.metrics, times(1)).onClassicGroupStateTransition(null, STABLE);
+
+ // The new classic member 1 has a heartbeat timeout.
+ ScheduledTimeout<Void, Record> heartbeatTimeout = context.timer.timeout(
+ classicGroupHeartbeatKey(groupId, memberId1));
+ assertNotNull(heartbeatTimeout);
+ // The new rebalance has a groupJoin timeout.
+ ScheduledTimeout<Void, Record> groupJoinTimeout = context.timer.timeout(
+ classicGroupJoinKey(groupId));
+ assertNotNull(groupJoinTimeout);
+
+ // A new rebalance is triggered.
+ ClassicGroup classicGroup = context.groupMetadataManager.getOrMaybeCreateClassicGroup(groupId, false);
+ assertTrue(classicGroup.isInState(PREPARING_REBALANCE));
+
+ // Simulate a failed write to the log.
+ result.appendFuture().completeExceptionally(new NotLeaderOrFollowerException());
+ context.rollback();
+
+ // The group is reverted back to the consumer group.
+ assertEquals(consumerGroup, context.groupMetadataManager.consumerGroup(groupId));
+ verify(context.metrics, times(1)).onClassicGroupStateTransition(PREPARING_REBALANCE, null);
+ }
+
+ @Test
+ public void testLastClassicProtocolMemberSessionTimeoutInConsumerGroup() {
+ String groupId = "group-id";
+ String memberId1 = Uuid.randomUuid().toString();
+ String memberId2 = Uuid.randomUuid().toString();
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+ Uuid barTopicId = Uuid.randomUuid();
+ String barTopicName = "bar";
+ Uuid zarTopicId = Uuid.randomUuid();
+ String zarTopicName = "zar";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = Collections.singletonList(
+ new ConsumerGroupMemberMetadataValue.ClassicProtocol()
+ .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),
+ new TopicPartition(fooTopicName, 2),
+ new TopicPartition(barTopicName, 0),
+ new TopicPartition(barTopicName, 1)
+ )
+ ))))
+ );
+
+ ConsumerGroupMember member1 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setClassicMemberMetadata(new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(protocols))
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .build();
+ ConsumerGroupMember member2 = new ConsumerGroupMember.Builder(memberId2)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar", "zar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .build();
+
+ // Consumer group with two members.
+ // Member 1 uses the classic protocol and member 2 uses the consumer protocol.
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.DOWNGRADE)
+ .withAssignors(Collections.singletonList(assignor))
+ .withMetadataImage(new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .addTopic(barTopicId, barTopicName, 3)
+ .addTopic(zarTopicId, zarTopicName, 1)
+ .addRacks()
+ .build())
+ .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
+ .withMember(member1)
+ .withMember(member2)
+ .withAssignment(memberId1, mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .withAssignment(memberId2, mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .withAssignmentEpoch(10))
+ .build();
+
+ context.replay(RecordHelpers.newGroupSubscriptionMetadataRecord(groupId, new HashMap<String, TopicMetadata>() {
+ {
+ put(fooTopicName, new TopicMetadata(fooTopicId, fooTopicName, 6, mkMapOfPartitionRacks(6)));
+ put(barTopicName, new TopicMetadata(barTopicId, barTopicName, 3, mkMapOfPartitionRacks(3)));
+ put(zarTopicName, new TopicMetadata(zarTopicId, zarTopicName, 1, mkMapOfPartitionRacks(1)));
+ }
+ }));
+
+ context.commit();
+
+ // Session timer is scheduled on the heartbeat.
+ context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId2)
+ .setMemberEpoch(10)
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar", "zar"))
+ .setTopicPartitions(Collections.emptyList()));
+
+ // Verify that there is a session timeout.
+ context.assertSessionTimeout(groupId, memberId2, 45000);
+
+ // Advance time past the session timeout.
+ // Member 2 should be fenced from the group, thus triggering the downgrade.
+ context.sleep(45000 + 1);
+
+ verify(context.metrics, times(1)).onConsumerGroupStateTransition(ConsumerGroup.ConsumerGroupState.STABLE, null);
+ verify(context.metrics, times(1)).onClassicGroupStateTransition(null, STABLE);
+
+ // The new classic member 1 has a heartbeat timeout.
+ ScheduledTimeout<Void, Record> heartbeatTimeout = context.timer.timeout(
+ classicGroupHeartbeatKey(groupId, memberId1));
+ assertNotNull(heartbeatTimeout);
+ // The new rebalance has a groupJoin timeout.
+ ScheduledTimeout<Void, Record> groupJoinTimeout = context.timer.timeout(
+ classicGroupJoinKey(groupId));
+ assertNotNull(groupJoinTimeout);
+
+ // A new rebalance is triggered.
+ ClassicGroup classicGroup = context.groupMetadataManager.getOrMaybeCreateClassicGroup(groupId, false);
+ assertTrue(classicGroup.isInState(PREPARING_REBALANCE));
+ }
+
+ @Test
+ public void testLastClassicProtocolMemberRebalanceTimeoutInConsumerGroup() {
+ String groupId = "group-id";
+ String memberId1 = Uuid.randomUuid().toString();
+ String memberId2 = Uuid.randomUuid().toString();
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+ Uuid barTopicId = Uuid.randomUuid();
+ String barTopicName = "bar";
+ Uuid zarTopicId = Uuid.randomUuid();
+ String zarTopicName = "zar";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = Collections.singletonList(
+ new ConsumerGroupMemberMetadataValue.ClassicProtocol()
+ .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),
+ new TopicPartition(fooTopicName, 2),
+ new TopicPartition(barTopicName, 0),
+ new TopicPartition(barTopicName, 1)
+ )
+ ))))
+ );
+
+ ConsumerGroupMember member1 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(30000)
+ .setClassicMemberMetadata(new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(protocols))
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .build();
+ ConsumerGroupMember member2 = new ConsumerGroupMember.Builder(memberId2)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar", "zar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(30000)
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .build();
+
+ // Consumer group with two members.
+ // Member 1 uses the classic protocol and member 2 uses the consumer protocol.
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.DOWNGRADE)
+ .withAssignors(Collections.singletonList(assignor))
+ .withMetadataImage(new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .addTopic(barTopicId, barTopicName, 3)
+ .addTopic(zarTopicId, zarTopicName, 1)
+ .addRacks()
+ .build())
+ .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
+ .withMember(member1)
+ .withMember(member2)
+ .withAssignment(memberId1, mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .withAssignment(memberId2, mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .withAssignmentEpoch(10))
+ .build();
+
+ context.replay(RecordHelpers.newGroupSubscriptionMetadataRecord(groupId, new HashMap<String, TopicMetadata>() {
+ {
+ put(fooTopicName, new TopicMetadata(fooTopicId, fooTopicName, 6, mkMapOfPartitionRacks(6)));
+ put(barTopicName, new TopicMetadata(barTopicId, barTopicName, 3, mkMapOfPartitionRacks(3)));
+ put(zarTopicName, new TopicMetadata(zarTopicId, zarTopicName, 1, mkMapOfPartitionRacks(1)));
+ }
+ }));
+
+ context.commit();
+
+ // Prepare the new assignment.
+ assignor.prepareGroupAssignment(new GroupAssignment(
+ new HashMap<String, MemberAssignment>() {
+ {
+ put(memberId1, new MemberAssignment(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)
+ )));
+ put(memberId2, new MemberAssignment(mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5)
+ )));
+ }
+ }
+ ));
+
+ // Member 2 heartbeats with a different subscribedTopicNames. The assignor computes a new assignment
+ // where member 2 will need to revoke topic partition bar-2 thus transitions to the REVOKING state.
+ context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId2)
+ .setMemberEpoch(10)
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setTopicPartitions(Arrays.asList(
+ new ConsumerGroupHeartbeatRequestData.TopicPartitions()
+ .setTopicId(fooTopicId)
+ .setPartitions(Arrays.asList(0, 1, 2)),
+ new ConsumerGroupHeartbeatRequestData.TopicPartitions()
+ .setTopicId(barTopicId)
+ .setPartitions(Arrays.asList(0, 1))
+ ))
+ );
+
+ // Verify that there is a rebalance timeout.
+ context.assertRebalanceTimeout(groupId, memberId2, 30000);
+
+ // Advance time past the session timeout.
+ // Member 2 should be fenced from the group, thus triggering the downgrade.
Review Comment:
Should we verify the records here too?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -777,11 +778,78 @@ public ClassicGroup classicGroup(
}
}
+ /**
+ * Validates the online downgrade if a consumer member is fenced from the consumer group.
+ *
+ * @param consumerGroup The ConsumerGroup.
+ * @param memberId The fenced member id.
+ * @return A boolean indicating whether it's valid to online downgrade the consumer group.
+ */
+ private 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)) {
Review Comment:
Ah, it makes sense.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -10331,6 +10333,461 @@ public void testClassicGroupOnUnloadedCompletingRebalance() throws Exception {
.setErrorCode(NOT_COORDINATOR.code()), pendingMemberSyncResult.syncFuture.get());
}
+ @Test
+ public void testLastClassicProtocolMemberLeavingConsumerGroup() {
+ String groupId = "group-id";
+ String memberId1 = Uuid.randomUuid().toString();
+ String memberId2 = Uuid.randomUuid().toString();
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+ Uuid barTopicId = Uuid.randomUuid();
+ String barTopicName = "bar";
+ Uuid zarTopicId = Uuid.randomUuid();
+ String zarTopicName = "zar";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = Collections.singletonList(
+ new ConsumerGroupMemberMetadataValue.ClassicProtocol()
+ .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),
+ new TopicPartition(fooTopicName, 2),
+ new TopicPartition(barTopicName, 0),
+ new TopicPartition(barTopicName, 1)
+ )
+ ))))
+ );
+
+ ConsumerGroupMember member1 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setClassicMemberMetadata(new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(protocols))
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .build();
+ ConsumerGroupMember member2 = new ConsumerGroupMember.Builder(memberId2)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ // Use zar only here to ensure that metadata needs to be recomputed.
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar", "zar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .build();
+
+ // Consumer group with two members.
+ // Member 1 uses the classic protocol and member 2 uses the consumer protocol.
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.DOWNGRADE)
+ .withAssignors(Collections.singletonList(assignor))
+ .withMetadataImage(new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .addTopic(barTopicId, barTopicName, 3)
+ .addTopic(zarTopicId, zarTopicName, 1)
+ .addRacks()
+ .build())
+ .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
+ .withMember(member1)
+ .withMember(member2)
+ .withAssignment(memberId1, mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .withAssignment(memberId2, mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .withAssignmentEpoch(10))
+ .build();
+
+ context.commit();
+ ConsumerGroup consumerGroup = context.groupMetadataManager.consumerGroup(groupId);
+
+ // Member 2 leaves the consumer group, triggering the downgrade.
+ CoordinatorResult<ConsumerGroupHeartbeatResponseData, Record> result = context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId2)
+ .setMemberEpoch(LEAVE_GROUP_MEMBER_EPOCH)
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setTopicPartitions(Collections.emptyList()));
+
+
+ byte[] assignment = Utils.toArray(ConsumerProtocol.serializeAssignment(new ConsumerPartitionAssignor.Assignment(Arrays.asList(
+ new TopicPartition(fooTopicName, 0), new TopicPartition(fooTopicName, 1), new TopicPartition(fooTopicName, 2),
+ new TopicPartition(barTopicName, 0), new TopicPartition(barTopicName, 1)
+ ))));
+ Map<String, byte[]> assignments = new HashMap<String, byte[]>() {
+ {
+ put(memberId1, assignment);
+ }
+ };
+
+ ClassicGroup expectedClassicGroup = new ClassicGroup(
+ new LogContext(),
+ groupId,
+ STABLE,
+ context.time,
+ context.metrics,
+ 10,
+ Optional.ofNullable(ConsumerProtocol.PROTOCOL_TYPE),
+ Optional.ofNullable("range"),
+ Optional.ofNullable(memberId1),
+ Optional.of(context.time.milliseconds())
+ );
+ expectedClassicGroup.add(
+ new ClassicGroupMember(
+ memberId1,
+ Optional.ofNullable(member1.instanceId()),
+ member1.clientId(),
+ member1.clientHost(),
+ member1.rebalanceTimeoutMs(),
+ 45000,
+ ConsumerProtocol.PROTOCOL_TYPE,
+ member1.supportedJoinGroupRequestProtocols(),
+ assignment
+ )
+ );
+
+ List<Record> expectedRecords = Arrays.asList(
+ RecordHelpers.newCurrentAssignmentTombstoneRecord(groupId, memberId2),
+ RecordHelpers.newTargetAssignmentTombstoneRecord(groupId, memberId2),
+ RecordHelpers.newMemberSubscriptionTombstoneRecord(groupId, memberId2),
+ // Subscription metadata is recomputed because zar is no longer there.
+ RecordHelpers.newGroupSubscriptionMetadataRecord(groupId, new HashMap<String, TopicMetadata>() {
+ {
+ put(fooTopicName, new TopicMetadata(fooTopicId, fooTopicName, 6, mkMapOfPartitionRacks(6)));
+ put(barTopicName, new TopicMetadata(barTopicId, barTopicName, 3, mkMapOfPartitionRacks(3)));
+ }
+ }),
+ RecordHelpers.newGroupEpochRecord(groupId, 11),
+
+ RecordHelpers.newTargetAssignmentEpochTombstoneRecord(groupId),
+ RecordHelpers.newGroupSubscriptionMetadataTombstoneRecord(groupId),
+ RecordHelpers.newGroupEpochTombstoneRecord(groupId),
+ RecordHelpers.newGroupMetadataRecord(expectedClassicGroup, assignments, MetadataVersion.latestTesting())
+ );
+
+ assertRecordsEquals(expectedRecords, result.records());
+ verify(context.metrics, times(1)).onConsumerGroupStateTransition(ConsumerGroup.ConsumerGroupState.STABLE, null);
+ verify(context.metrics, times(1)).onClassicGroupStateTransition(null, STABLE);
+
+ // The new classic member 1 has a heartbeat timeout.
+ ScheduledTimeout<Void, Record> heartbeatTimeout = context.timer.timeout(
+ classicGroupHeartbeatKey(groupId, memberId1));
+ assertNotNull(heartbeatTimeout);
+ // The new rebalance has a groupJoin timeout.
+ ScheduledTimeout<Void, Record> groupJoinTimeout = context.timer.timeout(
+ classicGroupJoinKey(groupId));
+ assertNotNull(groupJoinTimeout);
+
+ // A new rebalance is triggered.
+ ClassicGroup classicGroup = context.groupMetadataManager.getOrMaybeCreateClassicGroup(groupId, false);
+ assertTrue(classicGroup.isInState(PREPARING_REBALANCE));
+
+ // Simulate a failed write to the log.
+ result.appendFuture().completeExceptionally(new NotLeaderOrFollowerException());
+ context.rollback();
+
+ // The group is reverted back to the consumer group.
+ assertEquals(consumerGroup, context.groupMetadataManager.consumerGroup(groupId));
+ verify(context.metrics, times(1)).onClassicGroupStateTransition(PREPARING_REBALANCE, null);
+ }
+
+ @Test
+ public void testLastClassicProtocolMemberSessionTimeoutInConsumerGroup() {
+ String groupId = "group-id";
+ String memberId1 = Uuid.randomUuid().toString();
+ String memberId2 = Uuid.randomUuid().toString();
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+ Uuid barTopicId = Uuid.randomUuid();
+ String barTopicName = "bar";
+ Uuid zarTopicId = Uuid.randomUuid();
+ String zarTopicName = "zar";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = Collections.singletonList(
+ new ConsumerGroupMemberMetadataValue.ClassicProtocol()
+ .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),
+ new TopicPartition(fooTopicName, 2),
+ new TopicPartition(barTopicName, 0),
+ new TopicPartition(barTopicName, 1)
+ )
+ ))))
+ );
+
+ ConsumerGroupMember member1 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setClassicMemberMetadata(new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(protocols))
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .build();
+ ConsumerGroupMember member2 = new ConsumerGroupMember.Builder(memberId2)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar", "zar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .build();
+
+ // Consumer group with two members.
+ // Member 1 uses the classic protocol and member 2 uses the consumer protocol.
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.DOWNGRADE)
+ .withAssignors(Collections.singletonList(assignor))
+ .withMetadataImage(new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .addTopic(barTopicId, barTopicName, 3)
+ .addTopic(zarTopicId, zarTopicName, 1)
+ .addRacks()
+ .build())
+ .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
+ .withMember(member1)
+ .withMember(member2)
+ .withAssignment(memberId1, mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .withAssignment(memberId2, mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .withAssignmentEpoch(10))
+ .build();
+
+ context.replay(RecordHelpers.newGroupSubscriptionMetadataRecord(groupId, new HashMap<String, TopicMetadata>() {
+ {
+ put(fooTopicName, new TopicMetadata(fooTopicId, fooTopicName, 6, mkMapOfPartitionRacks(6)));
+ put(barTopicName, new TopicMetadata(barTopicId, barTopicName, 3, mkMapOfPartitionRacks(3)));
+ put(zarTopicName, new TopicMetadata(zarTopicId, zarTopicName, 1, mkMapOfPartitionRacks(1)));
+ }
+ }));
+
+ context.commit();
+
+ // Session timer is scheduled on the heartbeat.
+ context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId2)
+ .setMemberEpoch(10)
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar", "zar"))
+ .setTopicPartitions(Collections.emptyList()));
+
+ // Verify that there is a session timeout.
+ context.assertSessionTimeout(groupId, memberId2, 45000);
+
+ // Advance time past the session timeout.
+ // Member 2 should be fenced from the group, thus triggering the downgrade.
+ context.sleep(45000 + 1);
+
+ verify(context.metrics, times(1)).onConsumerGroupStateTransition(ConsumerGroup.ConsumerGroupState.STABLE, null);
+ verify(context.metrics, times(1)).onClassicGroupStateTransition(null, STABLE);
+
+ // The new classic member 1 has a heartbeat timeout.
+ ScheduledTimeout<Void, Record> heartbeatTimeout = context.timer.timeout(
+ classicGroupHeartbeatKey(groupId, memberId1));
+ assertNotNull(heartbeatTimeout);
+ // The new rebalance has a groupJoin timeout.
+ ScheduledTimeout<Void, Record> groupJoinTimeout = context.timer.timeout(
+ classicGroupJoinKey(groupId));
+ assertNotNull(groupJoinTimeout);
+
+ // A new rebalance is triggered.
+ ClassicGroup classicGroup = context.groupMetadataManager.getOrMaybeCreateClassicGroup(groupId, false);
+ assertTrue(classicGroup.isInState(PREPARING_REBALANCE));
+ }
+
+ @Test
+ public void testLastClassicProtocolMemberRebalanceTimeoutInConsumerGroup() {
+ String groupId = "group-id";
+ String memberId1 = Uuid.randomUuid().toString();
+ String memberId2 = Uuid.randomUuid().toString();
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+ Uuid barTopicId = Uuid.randomUuid();
+ String barTopicName = "bar";
+ Uuid zarTopicId = Uuid.randomUuid();
+ String zarTopicName = "zar";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = Collections.singletonList(
+ new ConsumerGroupMemberMetadataValue.ClassicProtocol()
+ .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),
+ new TopicPartition(fooTopicName, 2),
+ new TopicPartition(barTopicName, 0),
+ new TopicPartition(barTopicName, 1)
+ )
+ ))))
+ );
+
+ ConsumerGroupMember member1 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(30000)
+ .setClassicMemberMetadata(new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(protocols))
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .build();
+ ConsumerGroupMember member2 = new ConsumerGroupMember.Builder(memberId2)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar", "zar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(30000)
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .build();
+
+ // Consumer group with two members.
+ // Member 1 uses the classic protocol and member 2 uses the consumer protocol.
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.DOWNGRADE)
+ .withAssignors(Collections.singletonList(assignor))
+ .withMetadataImage(new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .addTopic(barTopicId, barTopicName, 3)
+ .addTopic(zarTopicId, zarTopicName, 1)
+ .addRacks()
+ .build())
+ .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
+ .withMember(member1)
+ .withMember(member2)
+ .withAssignment(memberId1, mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .withAssignment(memberId2, mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .withAssignmentEpoch(10))
+ .build();
+
+ context.replay(RecordHelpers.newGroupSubscriptionMetadataRecord(groupId, new HashMap<String, TopicMetadata>() {
+ {
+ put(fooTopicName, new TopicMetadata(fooTopicId, fooTopicName, 6, mkMapOfPartitionRacks(6)));
+ put(barTopicName, new TopicMetadata(barTopicId, barTopicName, 3, mkMapOfPartitionRacks(3)));
+ put(zarTopicName, new TopicMetadata(zarTopicId, zarTopicName, 1, mkMapOfPartitionRacks(1)));
+ }
+ }));
+
+ context.commit();
+
+ // Prepare the new assignment.
+ assignor.prepareGroupAssignment(new GroupAssignment(
+ new HashMap<String, MemberAssignment>() {
+ {
+ put(memberId1, new MemberAssignment(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)
+ )));
+ put(memberId2, new MemberAssignment(mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5)
+ )));
+ }
+ }
+ ));
+
+ // Member 2 heartbeats with a different subscribedTopicNames. The assignor computes a new assignment
+ // where member 2 will need to revoke topic partition bar-2 thus transitions to the REVOKING state.
+ context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId2)
+ .setMemberEpoch(10)
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setTopicPartitions(Arrays.asList(
+ new ConsumerGroupHeartbeatRequestData.TopicPartitions()
+ .setTopicId(fooTopicId)
+ .setPartitions(Arrays.asList(0, 1, 2)),
+ new ConsumerGroupHeartbeatRequestData.TopicPartitions()
+ .setTopicId(barTopicId)
+ .setPartitions(Arrays.asList(0, 1))
+ ))
+ );
+
+ // Verify that there is a rebalance timeout.
+ context.assertRebalanceTimeout(groupId, memberId2, 30000);
+
+ // Advance time past the session timeout.
+ // Member 2 should be fenced from the group, thus triggering the downgrade.
+ context.sleep(30000 + 1);
+
+ verify(context.metrics, times(1)).onConsumerGroupStateTransition(ConsumerGroup.ConsumerGroupState.RECONCILING, null);
+ verify(context.metrics, times(1)).onClassicGroupStateTransition(null, STABLE);
+
+ // The new classic member 1 has a heartbeat timeout.
+ ScheduledTimeout<Void, Record> heartbeatTimeout = context.timer.timeout(
+ classicGroupHeartbeatKey(groupId, memberId1));
Review Comment:
ditto about the style.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -10331,6 +10333,461 @@ public void testClassicGroupOnUnloadedCompletingRebalance() throws Exception {
.setErrorCode(NOT_COORDINATOR.code()), pendingMemberSyncResult.syncFuture.get());
}
+ @Test
+ public void testLastClassicProtocolMemberLeavingConsumerGroup() {
+ String groupId = "group-id";
+ String memberId1 = Uuid.randomUuid().toString();
+ String memberId2 = Uuid.randomUuid().toString();
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+ Uuid barTopicId = Uuid.randomUuid();
+ String barTopicName = "bar";
+ Uuid zarTopicId = Uuid.randomUuid();
+ String zarTopicName = "zar";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = Collections.singletonList(
+ new ConsumerGroupMemberMetadataValue.ClassicProtocol()
+ .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),
+ new TopicPartition(fooTopicName, 2),
+ new TopicPartition(barTopicName, 0),
+ new TopicPartition(barTopicName, 1)
+ )
+ ))))
+ );
+
+ ConsumerGroupMember member1 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setClassicMemberMetadata(new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(protocols))
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .build();
+ ConsumerGroupMember member2 = new ConsumerGroupMember.Builder(memberId2)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ // Use zar only here to ensure that metadata needs to be recomputed.
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar", "zar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .build();
+
+ // Consumer group with two members.
+ // Member 1 uses the classic protocol and member 2 uses the consumer protocol.
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.DOWNGRADE)
+ .withAssignors(Collections.singletonList(assignor))
+ .withMetadataImage(new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .addTopic(barTopicId, barTopicName, 3)
+ .addTopic(zarTopicId, zarTopicName, 1)
+ .addRacks()
+ .build())
+ .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
+ .withMember(member1)
+ .withMember(member2)
+ .withAssignment(memberId1, mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .withAssignment(memberId2, mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .withAssignmentEpoch(10))
+ .build();
+
+ context.commit();
+ ConsumerGroup consumerGroup = context.groupMetadataManager.consumerGroup(groupId);
+
+ // Member 2 leaves the consumer group, triggering the downgrade.
+ CoordinatorResult<ConsumerGroupHeartbeatResponseData, Record> result = context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId2)
+ .setMemberEpoch(LEAVE_GROUP_MEMBER_EPOCH)
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setTopicPartitions(Collections.emptyList()));
+
+
+ byte[] assignment = Utils.toArray(ConsumerProtocol.serializeAssignment(new ConsumerPartitionAssignor.Assignment(Arrays.asList(
+ new TopicPartition(fooTopicName, 0), new TopicPartition(fooTopicName, 1), new TopicPartition(fooTopicName, 2),
+ new TopicPartition(barTopicName, 0), new TopicPartition(barTopicName, 1)
Review Comment:
nit: Let's put one per line.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -777,11 +778,79 @@ public ClassicGroup classicGroup(
}
}
+ /**
+ * Validates the online downgrade if a consumer member is fenced from the consumer group.
+ *
+ * @param consumerGroup The ConsumerGroup.
+ * @param memberId The fenced member id.
+ * @return A boolean indicating whether it's valid to online downgrade the consumer group.
+ */
+ private 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;
+ }
+
+ private CompletableFuture<Void> convertToClassicGroup(ConsumerGroup consumerGroup, String leavingMemberId, List<Record> records) {
Review Comment:
nit: Let's add some javadoc.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -10331,6 +10333,461 @@ public void testClassicGroupOnUnloadedCompletingRebalance() throws Exception {
.setErrorCode(NOT_COORDINATOR.code()), pendingMemberSyncResult.syncFuture.get());
}
+ @Test
+ public void testLastClassicProtocolMemberLeavingConsumerGroup() {
+ String groupId = "group-id";
+ String memberId1 = Uuid.randomUuid().toString();
+ String memberId2 = Uuid.randomUuid().toString();
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+ Uuid barTopicId = Uuid.randomUuid();
+ String barTopicName = "bar";
+ Uuid zarTopicId = Uuid.randomUuid();
+ String zarTopicName = "zar";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = Collections.singletonList(
+ new ConsumerGroupMemberMetadataValue.ClassicProtocol()
+ .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),
+ new TopicPartition(fooTopicName, 2),
+ new TopicPartition(barTopicName, 0),
+ new TopicPartition(barTopicName, 1)
+ )
+ ))))
+ );
+
+ ConsumerGroupMember member1 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setClassicMemberMetadata(new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(protocols))
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .build();
+ ConsumerGroupMember member2 = new ConsumerGroupMember.Builder(memberId2)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ // Use zar only here to ensure that metadata needs to be recomputed.
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar", "zar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .build();
+
+ // Consumer group with two members.
+ // Member 1 uses the classic protocol and member 2 uses the consumer protocol.
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.DOWNGRADE)
+ .withAssignors(Collections.singletonList(assignor))
+ .withMetadataImage(new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .addTopic(barTopicId, barTopicName, 3)
+ .addTopic(zarTopicId, zarTopicName, 1)
+ .addRacks()
+ .build())
+ .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
+ .withMember(member1)
+ .withMember(member2)
+ .withAssignment(memberId1, mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .withAssignment(memberId2, mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .withAssignmentEpoch(10))
+ .build();
+
+ context.commit();
+ ConsumerGroup consumerGroup = context.groupMetadataManager.consumerGroup(groupId);
+
+ // Member 2 leaves the consumer group, triggering the downgrade.
+ CoordinatorResult<ConsumerGroupHeartbeatResponseData, Record> result = context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId2)
+ .setMemberEpoch(LEAVE_GROUP_MEMBER_EPOCH)
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setTopicPartitions(Collections.emptyList()));
+
+
+ byte[] assignment = Utils.toArray(ConsumerProtocol.serializeAssignment(new ConsumerPartitionAssignor.Assignment(Arrays.asList(
+ new TopicPartition(fooTopicName, 0), new TopicPartition(fooTopicName, 1), new TopicPartition(fooTopicName, 2),
+ new TopicPartition(barTopicName, 0), new TopicPartition(barTopicName, 1)
+ ))));
+ Map<String, byte[]> assignments = new HashMap<String, byte[]>() {
+ {
+ put(memberId1, assignment);
+ }
+ };
+
+ ClassicGroup expectedClassicGroup = new ClassicGroup(
+ new LogContext(),
+ groupId,
+ STABLE,
+ context.time,
+ context.metrics,
+ 10,
+ Optional.ofNullable(ConsumerProtocol.PROTOCOL_TYPE),
+ Optional.ofNullable("range"),
+ Optional.ofNullable(memberId1),
+ Optional.of(context.time.milliseconds())
+ );
+ expectedClassicGroup.add(
+ new ClassicGroupMember(
+ memberId1,
+ Optional.ofNullable(member1.instanceId()),
+ member1.clientId(),
+ member1.clientHost(),
+ member1.rebalanceTimeoutMs(),
+ 45000,
+ ConsumerProtocol.PROTOCOL_TYPE,
+ member1.supportedJoinGroupRequestProtocols(),
+ assignment
+ )
+ );
+
+ List<Record> expectedRecords = Arrays.asList(
+ RecordHelpers.newCurrentAssignmentTombstoneRecord(groupId, memberId2),
+ RecordHelpers.newTargetAssignmentTombstoneRecord(groupId, memberId2),
+ RecordHelpers.newMemberSubscriptionTombstoneRecord(groupId, memberId2),
+ // Subscription metadata is recomputed because zar is no longer there.
+ RecordHelpers.newGroupSubscriptionMetadataRecord(groupId, new HashMap<String, TopicMetadata>() {
+ {
+ put(fooTopicName, new TopicMetadata(fooTopicId, fooTopicName, 6, mkMapOfPartitionRacks(6)));
+ put(barTopicName, new TopicMetadata(barTopicId, barTopicName, 3, mkMapOfPartitionRacks(3)));
+ }
+ }),
+ RecordHelpers.newGroupEpochRecord(groupId, 11),
Review Comment:
It is actually a bit weird to have those here because we are about to delete the entire group. I wonder if we could optimise it. Thoughts?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -10331,6 +10333,461 @@ public void testClassicGroupOnUnloadedCompletingRebalance() throws Exception {
.setErrorCode(NOT_COORDINATOR.code()), pendingMemberSyncResult.syncFuture.get());
}
+ @Test
+ public void testLastClassicProtocolMemberLeavingConsumerGroup() {
+ String groupId = "group-id";
+ String memberId1 = Uuid.randomUuid().toString();
+ String memberId2 = Uuid.randomUuid().toString();
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+ Uuid barTopicId = Uuid.randomUuid();
+ String barTopicName = "bar";
+ Uuid zarTopicId = Uuid.randomUuid();
+ String zarTopicName = "zar";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = Collections.singletonList(
+ new ConsumerGroupMemberMetadataValue.ClassicProtocol()
+ .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),
+ new TopicPartition(fooTopicName, 2),
+ new TopicPartition(barTopicName, 0),
+ new TopicPartition(barTopicName, 1)
+ )
+ ))))
+ );
+
+ ConsumerGroupMember member1 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setClassicMemberMetadata(new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(protocols))
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .build();
+ ConsumerGroupMember member2 = new ConsumerGroupMember.Builder(memberId2)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ // Use zar only here to ensure that metadata needs to be recomputed.
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar", "zar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .build();
+
+ // Consumer group with two members.
+ // Member 1 uses the classic protocol and member 2 uses the consumer protocol.
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.DOWNGRADE)
+ .withAssignors(Collections.singletonList(assignor))
+ .withMetadataImage(new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .addTopic(barTopicId, barTopicName, 3)
+ .addTopic(zarTopicId, zarTopicName, 1)
+ .addRacks()
+ .build())
+ .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
+ .withMember(member1)
+ .withMember(member2)
+ .withAssignment(memberId1, mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .withAssignment(memberId2, mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .withAssignmentEpoch(10))
+ .build();
+
+ context.commit();
+ ConsumerGroup consumerGroup = context.groupMetadataManager.consumerGroup(groupId);
+
+ // Member 2 leaves the consumer group, triggering the downgrade.
+ CoordinatorResult<ConsumerGroupHeartbeatResponseData, Record> result = context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId2)
+ .setMemberEpoch(LEAVE_GROUP_MEMBER_EPOCH)
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setTopicPartitions(Collections.emptyList()));
+
+
+ byte[] assignment = Utils.toArray(ConsumerProtocol.serializeAssignment(new ConsumerPartitionAssignor.Assignment(Arrays.asList(
+ new TopicPartition(fooTopicName, 0), new TopicPartition(fooTopicName, 1), new TopicPartition(fooTopicName, 2),
+ new TopicPartition(barTopicName, 0), new TopicPartition(barTopicName, 1)
+ ))));
+ Map<String, byte[]> assignments = new HashMap<String, byte[]>() {
+ {
+ put(memberId1, assignment);
+ }
+ };
+
+ ClassicGroup expectedClassicGroup = new ClassicGroup(
+ new LogContext(),
+ groupId,
+ STABLE,
+ context.time,
+ context.metrics,
+ 10,
+ Optional.ofNullable(ConsumerProtocol.PROTOCOL_TYPE),
+ Optional.ofNullable("range"),
+ Optional.ofNullable(memberId1),
+ Optional.of(context.time.milliseconds())
+ );
+ expectedClassicGroup.add(
+ new ClassicGroupMember(
+ memberId1,
+ Optional.ofNullable(member1.instanceId()),
+ member1.clientId(),
+ member1.clientHost(),
+ member1.rebalanceTimeoutMs(),
+ 45000,
+ ConsumerProtocol.PROTOCOL_TYPE,
+ member1.supportedJoinGroupRequestProtocols(),
+ assignment
+ )
+ );
+
+ List<Record> expectedRecords = Arrays.asList(
+ RecordHelpers.newCurrentAssignmentTombstoneRecord(groupId, memberId2),
+ RecordHelpers.newTargetAssignmentTombstoneRecord(groupId, memberId2),
+ RecordHelpers.newMemberSubscriptionTombstoneRecord(groupId, memberId2),
+ // Subscription metadata is recomputed because zar is no longer there.
+ RecordHelpers.newGroupSubscriptionMetadataRecord(groupId, new HashMap<String, TopicMetadata>() {
+ {
+ put(fooTopicName, new TopicMetadata(fooTopicId, fooTopicName, 6, mkMapOfPartitionRacks(6)));
+ put(barTopicName, new TopicMetadata(barTopicId, barTopicName, 3, mkMapOfPartitionRacks(3)));
+ }
+ }),
+ RecordHelpers.newGroupEpochRecord(groupId, 11),
+
+ RecordHelpers.newTargetAssignmentEpochTombstoneRecord(groupId),
+ RecordHelpers.newGroupSubscriptionMetadataTombstoneRecord(groupId),
+ RecordHelpers.newGroupEpochTombstoneRecord(groupId),
+ RecordHelpers.newGroupMetadataRecord(expectedClassicGroup, assignments, MetadataVersion.latestTesting())
+ );
+
+ assertRecordsEquals(expectedRecords, result.records());
+ verify(context.metrics, times(1)).onConsumerGroupStateTransition(ConsumerGroup.ConsumerGroupState.STABLE, null);
+ verify(context.metrics, times(1)).onClassicGroupStateTransition(null, STABLE);
+
+ // The new classic member 1 has a heartbeat timeout.
+ ScheduledTimeout<Void, Record> heartbeatTimeout = context.timer.timeout(
+ classicGroupHeartbeatKey(groupId, memberId1));
+ assertNotNull(heartbeatTimeout);
+ // The new rebalance has a groupJoin timeout.
+ ScheduledTimeout<Void, Record> groupJoinTimeout = context.timer.timeout(
+ classicGroupJoinKey(groupId));
+ assertNotNull(groupJoinTimeout);
+
+ // A new rebalance is triggered.
+ ClassicGroup classicGroup = context.groupMetadataManager.getOrMaybeCreateClassicGroup(groupId, false);
+ assertTrue(classicGroup.isInState(PREPARING_REBALANCE));
+
+ // Simulate a failed write to the log.
+ result.appendFuture().completeExceptionally(new NotLeaderOrFollowerException());
+ context.rollback();
+
+ // The group is reverted back to the consumer group.
+ assertEquals(consumerGroup, context.groupMetadataManager.consumerGroup(groupId));
+ verify(context.metrics, times(1)).onClassicGroupStateTransition(PREPARING_REBALANCE, null);
+ }
+
+ @Test
+ public void testLastClassicProtocolMemberSessionTimeoutInConsumerGroup() {
Review Comment:
ditto about the name.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -10331,6 +10333,461 @@ public void testClassicGroupOnUnloadedCompletingRebalance() throws Exception {
.setErrorCode(NOT_COORDINATOR.code()), pendingMemberSyncResult.syncFuture.get());
}
+ @Test
+ public void testLastClassicProtocolMemberLeavingConsumerGroup() {
+ String groupId = "group-id";
+ String memberId1 = Uuid.randomUuid().toString();
+ String memberId2 = Uuid.randomUuid().toString();
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+ Uuid barTopicId = Uuid.randomUuid();
+ String barTopicName = "bar";
+ Uuid zarTopicId = Uuid.randomUuid();
+ String zarTopicName = "zar";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = Collections.singletonList(
+ new ConsumerGroupMemberMetadataValue.ClassicProtocol()
+ .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),
+ new TopicPartition(fooTopicName, 2),
+ new TopicPartition(barTopicName, 0),
+ new TopicPartition(barTopicName, 1)
+ )
+ ))))
+ );
+
+ ConsumerGroupMember member1 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setClassicMemberMetadata(new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(protocols))
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .build();
+ ConsumerGroupMember member2 = new ConsumerGroupMember.Builder(memberId2)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ // Use zar only here to ensure that metadata needs to be recomputed.
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar", "zar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .build();
+
+ // Consumer group with two members.
+ // Member 1 uses the classic protocol and member 2 uses the consumer protocol.
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.DOWNGRADE)
+ .withAssignors(Collections.singletonList(assignor))
+ .withMetadataImage(new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .addTopic(barTopicId, barTopicName, 3)
+ .addTopic(zarTopicId, zarTopicName, 1)
+ .addRacks()
+ .build())
+ .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
+ .withMember(member1)
+ .withMember(member2)
+ .withAssignment(memberId1, mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .withAssignment(memberId2, mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .withAssignmentEpoch(10))
+ .build();
+
+ context.commit();
+ ConsumerGroup consumerGroup = context.groupMetadataManager.consumerGroup(groupId);
+
+ // Member 2 leaves the consumer group, triggering the downgrade.
+ CoordinatorResult<ConsumerGroupHeartbeatResponseData, Record> result = context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId2)
+ .setMemberEpoch(LEAVE_GROUP_MEMBER_EPOCH)
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setTopicPartitions(Collections.emptyList()));
+
+
+ byte[] assignment = Utils.toArray(ConsumerProtocol.serializeAssignment(new ConsumerPartitionAssignor.Assignment(Arrays.asList(
+ new TopicPartition(fooTopicName, 0), new TopicPartition(fooTopicName, 1), new TopicPartition(fooTopicName, 2),
+ new TopicPartition(barTopicName, 0), new TopicPartition(barTopicName, 1)
+ ))));
+ Map<String, byte[]> assignments = new HashMap<String, byte[]>() {
+ {
+ put(memberId1, assignment);
+ }
+ };
+
+ ClassicGroup expectedClassicGroup = new ClassicGroup(
+ new LogContext(),
+ groupId,
+ STABLE,
+ context.time,
+ context.metrics,
+ 10,
+ Optional.ofNullable(ConsumerProtocol.PROTOCOL_TYPE),
+ Optional.ofNullable("range"),
+ Optional.ofNullable(memberId1),
+ Optional.of(context.time.milliseconds())
+ );
+ expectedClassicGroup.add(
+ new ClassicGroupMember(
+ memberId1,
+ Optional.ofNullable(member1.instanceId()),
+ member1.clientId(),
+ member1.clientHost(),
+ member1.rebalanceTimeoutMs(),
+ 45000,
+ ConsumerProtocol.PROTOCOL_TYPE,
+ member1.supportedJoinGroupRequestProtocols(),
+ assignment
+ )
+ );
+
+ List<Record> expectedRecords = Arrays.asList(
+ RecordHelpers.newCurrentAssignmentTombstoneRecord(groupId, memberId2),
+ RecordHelpers.newTargetAssignmentTombstoneRecord(groupId, memberId2),
+ RecordHelpers.newMemberSubscriptionTombstoneRecord(groupId, memberId2),
+ // Subscription metadata is recomputed because zar is no longer there.
+ RecordHelpers.newGroupSubscriptionMetadataRecord(groupId, new HashMap<String, TopicMetadata>() {
+ {
+ put(fooTopicName, new TopicMetadata(fooTopicId, fooTopicName, 6, mkMapOfPartitionRacks(6)));
+ put(barTopicName, new TopicMetadata(barTopicId, barTopicName, 3, mkMapOfPartitionRacks(3)));
+ }
+ }),
+ RecordHelpers.newGroupEpochRecord(groupId, 11),
+
+ RecordHelpers.newTargetAssignmentEpochTombstoneRecord(groupId),
Review Comment:
We miss records here. We also need tombstones for all the remaining members, isn't it?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -10331,6 +10333,461 @@ public void testClassicGroupOnUnloadedCompletingRebalance() throws Exception {
.setErrorCode(NOT_COORDINATOR.code()), pendingMemberSyncResult.syncFuture.get());
}
+ @Test
+ public void testLastClassicProtocolMemberLeavingConsumerGroup() {
+ String groupId = "group-id";
+ String memberId1 = Uuid.randomUuid().toString();
+ String memberId2 = Uuid.randomUuid().toString();
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+ Uuid barTopicId = Uuid.randomUuid();
+ String barTopicName = "bar";
+ Uuid zarTopicId = Uuid.randomUuid();
+ String zarTopicName = "zar";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = Collections.singletonList(
+ new ConsumerGroupMemberMetadataValue.ClassicProtocol()
+ .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),
+ new TopicPartition(fooTopicName, 2),
+ new TopicPartition(barTopicName, 0),
+ new TopicPartition(barTopicName, 1)
+ )
+ ))))
+ );
+
+ ConsumerGroupMember member1 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setClassicMemberMetadata(new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(protocols))
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .build();
+ ConsumerGroupMember member2 = new ConsumerGroupMember.Builder(memberId2)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ // Use zar only here to ensure that metadata needs to be recomputed.
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar", "zar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .build();
+
+ // Consumer group with two members.
+ // Member 1 uses the classic protocol and member 2 uses the consumer protocol.
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.DOWNGRADE)
+ .withAssignors(Collections.singletonList(assignor))
+ .withMetadataImage(new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .addTopic(barTopicId, barTopicName, 3)
+ .addTopic(zarTopicId, zarTopicName, 1)
+ .addRacks()
+ .build())
+ .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
+ .withMember(member1)
+ .withMember(member2)
+ .withAssignment(memberId1, mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .withAssignment(memberId2, mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .withAssignmentEpoch(10))
+ .build();
+
+ context.commit();
+ ConsumerGroup consumerGroup = context.groupMetadataManager.consumerGroup(groupId);
+
+ // Member 2 leaves the consumer group, triggering the downgrade.
+ CoordinatorResult<ConsumerGroupHeartbeatResponseData, Record> result = context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId2)
+ .setMemberEpoch(LEAVE_GROUP_MEMBER_EPOCH)
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setTopicPartitions(Collections.emptyList()));
+
+
+ byte[] assignment = Utils.toArray(ConsumerProtocol.serializeAssignment(new ConsumerPartitionAssignor.Assignment(Arrays.asList(
+ new TopicPartition(fooTopicName, 0), new TopicPartition(fooTopicName, 1), new TopicPartition(fooTopicName, 2),
+ new TopicPartition(barTopicName, 0), new TopicPartition(barTopicName, 1)
+ ))));
+ Map<String, byte[]> assignments = new HashMap<String, byte[]>() {
+ {
+ put(memberId1, assignment);
+ }
+ };
+
+ ClassicGroup expectedClassicGroup = new ClassicGroup(
+ new LogContext(),
+ groupId,
+ STABLE,
+ context.time,
+ context.metrics,
+ 10,
+ Optional.ofNullable(ConsumerProtocol.PROTOCOL_TYPE),
+ Optional.ofNullable("range"),
+ Optional.ofNullable(memberId1),
+ Optional.of(context.time.milliseconds())
+ );
+ expectedClassicGroup.add(
+ new ClassicGroupMember(
+ memberId1,
+ Optional.ofNullable(member1.instanceId()),
+ member1.clientId(),
+ member1.clientHost(),
+ member1.rebalanceTimeoutMs(),
+ 45000,
+ ConsumerProtocol.PROTOCOL_TYPE,
+ member1.supportedJoinGroupRequestProtocols(),
+ assignment
+ )
+ );
+
+ List<Record> expectedRecords = Arrays.asList(
+ RecordHelpers.newCurrentAssignmentTombstoneRecord(groupId, memberId2),
+ RecordHelpers.newTargetAssignmentTombstoneRecord(groupId, memberId2),
+ RecordHelpers.newMemberSubscriptionTombstoneRecord(groupId, memberId2),
+ // Subscription metadata is recomputed because zar is no longer there.
+ RecordHelpers.newGroupSubscriptionMetadataRecord(groupId, new HashMap<String, TopicMetadata>() {
+ {
+ put(fooTopicName, new TopicMetadata(fooTopicId, fooTopicName, 6, mkMapOfPartitionRacks(6)));
+ put(barTopicName, new TopicMetadata(barTopicId, barTopicName, 3, mkMapOfPartitionRacks(3)));
+ }
+ }),
+ RecordHelpers.newGroupEpochRecord(groupId, 11),
+
+ RecordHelpers.newTargetAssignmentEpochTombstoneRecord(groupId),
+ RecordHelpers.newGroupSubscriptionMetadataTombstoneRecord(groupId),
+ RecordHelpers.newGroupEpochTombstoneRecord(groupId),
+ RecordHelpers.newGroupMetadataRecord(expectedClassicGroup, assignments, MetadataVersion.latestTesting())
+ );
+
+ assertRecordsEquals(expectedRecords, result.records());
+ verify(context.metrics, times(1)).onConsumerGroupStateTransition(ConsumerGroup.ConsumerGroupState.STABLE, null);
+ verify(context.metrics, times(1)).onClassicGroupStateTransition(null, STABLE);
+
+ // The new classic member 1 has a heartbeat timeout.
+ ScheduledTimeout<Void, Record> heartbeatTimeout = context.timer.timeout(
+ classicGroupHeartbeatKey(groupId, memberId1));
+ assertNotNull(heartbeatTimeout);
+ // The new rebalance has a groupJoin timeout.
+ ScheduledTimeout<Void, Record> groupJoinTimeout = context.timer.timeout(
+ classicGroupJoinKey(groupId));
+ assertNotNull(groupJoinTimeout);
+
+ // A new rebalance is triggered.
+ ClassicGroup classicGroup = context.groupMetadataManager.getOrMaybeCreateClassicGroup(groupId, false);
+ assertTrue(classicGroup.isInState(PREPARING_REBALANCE));
+
+ // Simulate a failed write to the log.
+ result.appendFuture().completeExceptionally(new NotLeaderOrFollowerException());
+ context.rollback();
+
+ // The group is reverted back to the consumer group.
+ assertEquals(consumerGroup, context.groupMetadataManager.consumerGroup(groupId));
+ verify(context.metrics, times(1)).onClassicGroupStateTransition(PREPARING_REBALANCE, null);
+ }
+
+ @Test
+ public void testLastClassicProtocolMemberSessionTimeoutInConsumerGroup() {
+ String groupId = "group-id";
+ String memberId1 = Uuid.randomUuid().toString();
+ String memberId2 = Uuid.randomUuid().toString();
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+ Uuid barTopicId = Uuid.randomUuid();
+ String barTopicName = "bar";
+ Uuid zarTopicId = Uuid.randomUuid();
+ String zarTopicName = "zar";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = Collections.singletonList(
+ new ConsumerGroupMemberMetadataValue.ClassicProtocol()
+ .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),
+ new TopicPartition(fooTopicName, 2),
+ new TopicPartition(barTopicName, 0),
+ new TopicPartition(barTopicName, 1)
+ )
+ ))))
+ );
+
+ ConsumerGroupMember member1 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setClassicMemberMetadata(new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(protocols))
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .build();
+ ConsumerGroupMember member2 = new ConsumerGroupMember.Builder(memberId2)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar", "zar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .build();
+
+ // Consumer group with two members.
+ // Member 1 uses the classic protocol and member 2 uses the consumer protocol.
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.DOWNGRADE)
+ .withAssignors(Collections.singletonList(assignor))
+ .withMetadataImage(new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .addTopic(barTopicId, barTopicName, 3)
+ .addTopic(zarTopicId, zarTopicName, 1)
+ .addRacks()
+ .build())
+ .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
+ .withMember(member1)
+ .withMember(member2)
+ .withAssignment(memberId1, mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .withAssignment(memberId2, mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .withAssignmentEpoch(10))
+ .build();
+
+ context.replay(RecordHelpers.newGroupSubscriptionMetadataRecord(groupId, new HashMap<String, TopicMetadata>() {
+ {
+ put(fooTopicName, new TopicMetadata(fooTopicId, fooTopicName, 6, mkMapOfPartitionRacks(6)));
+ put(barTopicName, new TopicMetadata(barTopicId, barTopicName, 3, mkMapOfPartitionRacks(3)));
+ put(zarTopicName, new TopicMetadata(zarTopicId, zarTopicName, 1, mkMapOfPartitionRacks(1)));
+ }
+ }));
+
+ context.commit();
+
+ // Session timer is scheduled on the heartbeat.
+ context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId2)
+ .setMemberEpoch(10)
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar", "zar"))
+ .setTopicPartitions(Collections.emptyList()));
+
+ // Verify that there is a session timeout.
+ context.assertSessionTimeout(groupId, memberId2, 45000);
+
+ // Advance time past the session timeout.
+ // Member 2 should be fenced from the group, thus triggering the downgrade.
+ context.sleep(45000 + 1);
Review Comment:
Should we also verify the records?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -10331,6 +10333,461 @@ public void testClassicGroupOnUnloadedCompletingRebalance() throws Exception {
.setErrorCode(NOT_COORDINATOR.code()), pendingMemberSyncResult.syncFuture.get());
}
+ @Test
+ public void testLastClassicProtocolMemberLeavingConsumerGroup() {
+ String groupId = "group-id";
+ String memberId1 = Uuid.randomUuid().toString();
+ String memberId2 = Uuid.randomUuid().toString();
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+ Uuid barTopicId = Uuid.randomUuid();
+ String barTopicName = "bar";
+ Uuid zarTopicId = Uuid.randomUuid();
+ String zarTopicName = "zar";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = Collections.singletonList(
+ new ConsumerGroupMemberMetadataValue.ClassicProtocol()
+ .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),
+ new TopicPartition(fooTopicName, 2),
+ new TopicPartition(barTopicName, 0),
+ new TopicPartition(barTopicName, 1)
+ )
+ ))))
+ );
+
+ ConsumerGroupMember member1 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setClassicMemberMetadata(new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(protocols))
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .build();
+ ConsumerGroupMember member2 = new ConsumerGroupMember.Builder(memberId2)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ // Use zar only here to ensure that metadata needs to be recomputed.
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar", "zar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .build();
+
+ // Consumer group with two members.
+ // Member 1 uses the classic protocol and member 2 uses the consumer protocol.
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.DOWNGRADE)
+ .withAssignors(Collections.singletonList(assignor))
+ .withMetadataImage(new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .addTopic(barTopicId, barTopicName, 3)
+ .addTopic(zarTopicId, zarTopicName, 1)
+ .addRacks()
+ .build())
+ .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
+ .withMember(member1)
+ .withMember(member2)
+ .withAssignment(memberId1, mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .withAssignment(memberId2, mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .withAssignmentEpoch(10))
+ .build();
+
+ context.commit();
+ ConsumerGroup consumerGroup = context.groupMetadataManager.consumerGroup(groupId);
+
+ // Member 2 leaves the consumer group, triggering the downgrade.
+ CoordinatorResult<ConsumerGroupHeartbeatResponseData, Record> result = context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId2)
+ .setMemberEpoch(LEAVE_GROUP_MEMBER_EPOCH)
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setTopicPartitions(Collections.emptyList()));
+
+
+ byte[] assignment = Utils.toArray(ConsumerProtocol.serializeAssignment(new ConsumerPartitionAssignor.Assignment(Arrays.asList(
+ new TopicPartition(fooTopicName, 0), new TopicPartition(fooTopicName, 1), new TopicPartition(fooTopicName, 2),
+ new TopicPartition(barTopicName, 0), new TopicPartition(barTopicName, 1)
+ ))));
+ Map<String, byte[]> assignments = new HashMap<String, byte[]>() {
+ {
+ put(memberId1, assignment);
+ }
+ };
+
+ ClassicGroup expectedClassicGroup = new ClassicGroup(
+ new LogContext(),
+ groupId,
+ STABLE,
+ context.time,
+ context.metrics,
+ 10,
+ Optional.ofNullable(ConsumerProtocol.PROTOCOL_TYPE),
+ Optional.ofNullable("range"),
+ Optional.ofNullable(memberId1),
+ Optional.of(context.time.milliseconds())
+ );
+ expectedClassicGroup.add(
+ new ClassicGroupMember(
+ memberId1,
+ Optional.ofNullable(member1.instanceId()),
+ member1.clientId(),
+ member1.clientHost(),
+ member1.rebalanceTimeoutMs(),
+ 45000,
+ ConsumerProtocol.PROTOCOL_TYPE,
+ member1.supportedJoinGroupRequestProtocols(),
+ assignment
+ )
+ );
+
+ List<Record> expectedRecords = Arrays.asList(
+ RecordHelpers.newCurrentAssignmentTombstoneRecord(groupId, memberId2),
+ RecordHelpers.newTargetAssignmentTombstoneRecord(groupId, memberId2),
+ RecordHelpers.newMemberSubscriptionTombstoneRecord(groupId, memberId2),
+ // Subscription metadata is recomputed because zar is no longer there.
+ RecordHelpers.newGroupSubscriptionMetadataRecord(groupId, new HashMap<String, TopicMetadata>() {
+ {
+ put(fooTopicName, new TopicMetadata(fooTopicId, fooTopicName, 6, mkMapOfPartitionRacks(6)));
+ put(barTopicName, new TopicMetadata(barTopicId, barTopicName, 3, mkMapOfPartitionRacks(3)));
+ }
+ }),
+ RecordHelpers.newGroupEpochRecord(groupId, 11),
+
+ RecordHelpers.newTargetAssignmentEpochTombstoneRecord(groupId),
+ RecordHelpers.newGroupSubscriptionMetadataTombstoneRecord(groupId),
+ RecordHelpers.newGroupEpochTombstoneRecord(groupId),
+ RecordHelpers.newGroupMetadataRecord(expectedClassicGroup, assignments, MetadataVersion.latestTesting())
+ );
+
+ assertRecordsEquals(expectedRecords, result.records());
+ verify(context.metrics, times(1)).onConsumerGroupStateTransition(ConsumerGroup.ConsumerGroupState.STABLE, null);
+ verify(context.metrics, times(1)).onClassicGroupStateTransition(null, STABLE);
+
+ // The new classic member 1 has a heartbeat timeout.
+ ScheduledTimeout<Void, Record> heartbeatTimeout = context.timer.timeout(
+ classicGroupHeartbeatKey(groupId, memberId1));
+ assertNotNull(heartbeatTimeout);
+ // The new rebalance has a groupJoin timeout.
+ ScheduledTimeout<Void, Record> groupJoinTimeout = context.timer.timeout(
+ classicGroupJoinKey(groupId));
+ assertNotNull(groupJoinTimeout);
+
+ // A new rebalance is triggered.
+ ClassicGroup classicGroup = context.groupMetadataManager.getOrMaybeCreateClassicGroup(groupId, false);
+ assertTrue(classicGroup.isInState(PREPARING_REBALANCE));
+
+ // Simulate a failed write to the log.
+ result.appendFuture().completeExceptionally(new NotLeaderOrFollowerException());
+ context.rollback();
+
+ // The group is reverted back to the consumer group.
+ assertEquals(consumerGroup, context.groupMetadataManager.consumerGroup(groupId));
+ verify(context.metrics, times(1)).onClassicGroupStateTransition(PREPARING_REBALANCE, null);
+ }
+
+ @Test
+ public void testLastClassicProtocolMemberSessionTimeoutInConsumerGroup() {
+ String groupId = "group-id";
+ String memberId1 = Uuid.randomUuid().toString();
+ String memberId2 = Uuid.randomUuid().toString();
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+ Uuid barTopicId = Uuid.randomUuid();
+ String barTopicName = "bar";
+ Uuid zarTopicId = Uuid.randomUuid();
+ String zarTopicName = "zar";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = Collections.singletonList(
+ new ConsumerGroupMemberMetadataValue.ClassicProtocol()
+ .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),
+ new TopicPartition(fooTopicName, 2),
+ new TopicPartition(barTopicName, 0),
+ new TopicPartition(barTopicName, 1)
+ )
+ ))))
+ );
+
+ ConsumerGroupMember member1 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setClassicMemberMetadata(new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(protocols))
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .build();
+ ConsumerGroupMember member2 = new ConsumerGroupMember.Builder(memberId2)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar", "zar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .build();
+
+ // Consumer group with two members.
+ // Member 1 uses the classic protocol and member 2 uses the consumer protocol.
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.DOWNGRADE)
+ .withAssignors(Collections.singletonList(assignor))
+ .withMetadataImage(new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .addTopic(barTopicId, barTopicName, 3)
+ .addTopic(zarTopicId, zarTopicName, 1)
+ .addRacks()
+ .build())
+ .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
+ .withMember(member1)
+ .withMember(member2)
+ .withAssignment(memberId1, mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .withAssignment(memberId2, mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .withAssignmentEpoch(10))
+ .build();
+
+ context.replay(RecordHelpers.newGroupSubscriptionMetadataRecord(groupId, new HashMap<String, TopicMetadata>() {
+ {
+ put(fooTopicName, new TopicMetadata(fooTopicId, fooTopicName, 6, mkMapOfPartitionRacks(6)));
+ put(barTopicName, new TopicMetadata(barTopicId, barTopicName, 3, mkMapOfPartitionRacks(3)));
+ put(zarTopicName, new TopicMetadata(zarTopicId, zarTopicName, 1, mkMapOfPartitionRacks(1)));
+ }
+ }));
+
+ context.commit();
+
+ // Session timer is scheduled on the heartbeat.
+ context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId2)
+ .setMemberEpoch(10)
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar", "zar"))
+ .setTopicPartitions(Collections.emptyList()));
+
+ // Verify that there is a session timeout.
+ context.assertSessionTimeout(groupId, memberId2, 45000);
+
+ // Advance time past the session timeout.
+ // Member 2 should be fenced from the group, thus triggering the downgrade.
+ context.sleep(45000 + 1);
+
+ verify(context.metrics, times(1)).onConsumerGroupStateTransition(ConsumerGroup.ConsumerGroupState.STABLE, null);
+ verify(context.metrics, times(1)).onClassicGroupStateTransition(null, STABLE);
+
+ // The new classic member 1 has a heartbeat timeout.
+ ScheduledTimeout<Void, Record> heartbeatTimeout = context.timer.timeout(
+ classicGroupHeartbeatKey(groupId, memberId1));
Review Comment:
nit: Last `)`. on new line.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -777,11 +778,79 @@ public ClassicGroup classicGroup(
}
}
+ /**
+ * Validates the online downgrade if a consumer member is fenced from the consumer group.
+ *
+ * @param consumerGroup The ConsumerGroup.
+ * @param memberId The fenced member id.
+ * @return A boolean indicating whether it's valid to online downgrade the consumer group.
+ */
+ private 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;
+ }
+
+ private CompletableFuture<Void> convertToClassicGroup(ConsumerGroup consumerGroup, String leavingMemberId, List<Record> records) {
+ consumerGroup.createGroupTombstoneRecords(records);
Review Comment:
Coming back to one of my previous comments, this is not enough as we need to fully delete the group (including all members, etc).
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -10331,6 +10333,461 @@ public void testClassicGroupOnUnloadedCompletingRebalance() throws Exception {
.setErrorCode(NOT_COORDINATOR.code()), pendingMemberSyncResult.syncFuture.get());
}
+ @Test
+ public void testLastClassicProtocolMemberLeavingConsumerGroup() {
+ String groupId = "group-id";
+ String memberId1 = Uuid.randomUuid().toString();
+ String memberId2 = Uuid.randomUuid().toString();
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+ Uuid barTopicId = Uuid.randomUuid();
+ String barTopicName = "bar";
+ Uuid zarTopicId = Uuid.randomUuid();
+ String zarTopicName = "zar";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = Collections.singletonList(
+ new ConsumerGroupMemberMetadataValue.ClassicProtocol()
+ .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),
+ new TopicPartition(fooTopicName, 2),
+ new TopicPartition(barTopicName, 0),
+ new TopicPartition(barTopicName, 1)
+ )
+ ))))
+ );
+
+ ConsumerGroupMember member1 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setClassicMemberMetadata(new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(protocols))
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .build();
+ ConsumerGroupMember member2 = new ConsumerGroupMember.Builder(memberId2)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ // Use zar only here to ensure that metadata needs to be recomputed.
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar", "zar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .build();
+
+ // Consumer group with two members.
+ // Member 1 uses the classic protocol and member 2 uses the consumer protocol.
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.DOWNGRADE)
+ .withAssignors(Collections.singletonList(assignor))
+ .withMetadataImage(new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .addTopic(barTopicId, barTopicName, 3)
+ .addTopic(zarTopicId, zarTopicName, 1)
+ .addRacks()
+ .build())
+ .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
+ .withMember(member1)
+ .withMember(member2)
+ .withAssignment(memberId1, mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .withAssignment(memberId2, mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .withAssignmentEpoch(10))
+ .build();
+
+ context.commit();
+ ConsumerGroup consumerGroup = context.groupMetadataManager.consumerGroup(groupId);
+
+ // Member 2 leaves the consumer group, triggering the downgrade.
+ CoordinatorResult<ConsumerGroupHeartbeatResponseData, Record> result = context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId2)
+ .setMemberEpoch(LEAVE_GROUP_MEMBER_EPOCH)
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setTopicPartitions(Collections.emptyList()));
+
+
+ byte[] assignment = Utils.toArray(ConsumerProtocol.serializeAssignment(new ConsumerPartitionAssignor.Assignment(Arrays.asList(
+ new TopicPartition(fooTopicName, 0), new TopicPartition(fooTopicName, 1), new TopicPartition(fooTopicName, 2),
+ new TopicPartition(barTopicName, 0), new TopicPartition(barTopicName, 1)
+ ))));
+ Map<String, byte[]> assignments = new HashMap<String, byte[]>() {
+ {
+ put(memberId1, assignment);
+ }
+ };
+
+ ClassicGroup expectedClassicGroup = new ClassicGroup(
+ new LogContext(),
+ groupId,
+ STABLE,
+ context.time,
+ context.metrics,
+ 10,
+ Optional.ofNullable(ConsumerProtocol.PROTOCOL_TYPE),
+ Optional.ofNullable("range"),
+ Optional.ofNullable(memberId1),
+ Optional.of(context.time.milliseconds())
+ );
+ expectedClassicGroup.add(
+ new ClassicGroupMember(
+ memberId1,
+ Optional.ofNullable(member1.instanceId()),
+ member1.clientId(),
+ member1.clientHost(),
+ member1.rebalanceTimeoutMs(),
+ 45000,
+ ConsumerProtocol.PROTOCOL_TYPE,
+ member1.supportedJoinGroupRequestProtocols(),
+ assignment
+ )
+ );
+
+ List<Record> expectedRecords = Arrays.asList(
+ RecordHelpers.newCurrentAssignmentTombstoneRecord(groupId, memberId2),
+ RecordHelpers.newTargetAssignmentTombstoneRecord(groupId, memberId2),
+ RecordHelpers.newMemberSubscriptionTombstoneRecord(groupId, memberId2),
+ // Subscription metadata is recomputed because zar is no longer there.
+ RecordHelpers.newGroupSubscriptionMetadataRecord(groupId, new HashMap<String, TopicMetadata>() {
+ {
+ put(fooTopicName, new TopicMetadata(fooTopicId, fooTopicName, 6, mkMapOfPartitionRacks(6)));
+ put(barTopicName, new TopicMetadata(barTopicId, barTopicName, 3, mkMapOfPartitionRacks(3)));
+ }
+ }),
+ RecordHelpers.newGroupEpochRecord(groupId, 11),
+
+ RecordHelpers.newTargetAssignmentEpochTombstoneRecord(groupId),
+ RecordHelpers.newGroupSubscriptionMetadataTombstoneRecord(groupId),
+ RecordHelpers.newGroupEpochTombstoneRecord(groupId),
+ RecordHelpers.newGroupMetadataRecord(expectedClassicGroup, assignments, MetadataVersion.latestTesting())
+ );
+
+ assertRecordsEquals(expectedRecords, result.records());
+ verify(context.metrics, times(1)).onConsumerGroupStateTransition(ConsumerGroup.ConsumerGroupState.STABLE, null);
+ verify(context.metrics, times(1)).onClassicGroupStateTransition(null, STABLE);
+
+ // The new classic member 1 has a heartbeat timeout.
+ ScheduledTimeout<Void, Record> heartbeatTimeout = context.timer.timeout(
+ classicGroupHeartbeatKey(groupId, memberId1));
+ assertNotNull(heartbeatTimeout);
+ // The new rebalance has a groupJoin timeout.
+ ScheduledTimeout<Void, Record> groupJoinTimeout = context.timer.timeout(
+ classicGroupJoinKey(groupId));
+ assertNotNull(groupJoinTimeout);
+
+ // A new rebalance is triggered.
+ ClassicGroup classicGroup = context.groupMetadataManager.getOrMaybeCreateClassicGroup(groupId, false);
+ assertTrue(classicGroup.isInState(PREPARING_REBALANCE));
+
+ // Simulate a failed write to the log.
+ result.appendFuture().completeExceptionally(new NotLeaderOrFollowerException());
+ context.rollback();
+
+ // The group is reverted back to the consumer group.
+ assertEquals(consumerGroup, context.groupMetadataManager.consumerGroup(groupId));
+ verify(context.metrics, times(1)).onClassicGroupStateTransition(PREPARING_REBALANCE, null);
+ }
+
+ @Test
+ public void testLastClassicProtocolMemberSessionTimeoutInConsumerGroup() {
+ String groupId = "group-id";
+ String memberId1 = Uuid.randomUuid().toString();
+ String memberId2 = Uuid.randomUuid().toString();
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+ Uuid barTopicId = Uuid.randomUuid();
+ String barTopicName = "bar";
+ Uuid zarTopicId = Uuid.randomUuid();
+ String zarTopicName = "zar";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = Collections.singletonList(
+ new ConsumerGroupMemberMetadataValue.ClassicProtocol()
+ .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),
+ new TopicPartition(fooTopicName, 2),
+ new TopicPartition(barTopicName, 0),
+ new TopicPartition(barTopicName, 1)
+ )
+ ))))
+ );
+
+ ConsumerGroupMember member1 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setClassicMemberMetadata(new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(protocols))
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .build();
+ ConsumerGroupMember member2 = new ConsumerGroupMember.Builder(memberId2)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar", "zar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(45000)
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .build();
+
+ // Consumer group with two members.
+ // Member 1 uses the classic protocol and member 2 uses the consumer protocol.
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.DOWNGRADE)
+ .withAssignors(Collections.singletonList(assignor))
+ .withMetadataImage(new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .addTopic(barTopicId, barTopicName, 3)
+ .addTopic(zarTopicId, zarTopicName, 1)
+ .addRacks()
+ .build())
+ .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
+ .withMember(member1)
+ .withMember(member2)
+ .withAssignment(memberId1, mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .withAssignment(memberId2, mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .withAssignmentEpoch(10))
+ .build();
+
+ context.replay(RecordHelpers.newGroupSubscriptionMetadataRecord(groupId, new HashMap<String, TopicMetadata>() {
+ {
+ put(fooTopicName, new TopicMetadata(fooTopicId, fooTopicName, 6, mkMapOfPartitionRacks(6)));
+ put(barTopicName, new TopicMetadata(barTopicId, barTopicName, 3, mkMapOfPartitionRacks(3)));
+ put(zarTopicName, new TopicMetadata(zarTopicId, zarTopicName, 1, mkMapOfPartitionRacks(1)));
+ }
+ }));
+
+ context.commit();
+
+ // Session timer is scheduled on the heartbeat.
+ context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId2)
+ .setMemberEpoch(10)
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar", "zar"))
+ .setTopicPartitions(Collections.emptyList()));
+
+ // Verify that there is a session timeout.
+ context.assertSessionTimeout(groupId, memberId2, 45000);
+
+ // Advance time past the session timeout.
+ // Member 2 should be fenced from the group, thus triggering the downgrade.
+ context.sleep(45000 + 1);
+
+ verify(context.metrics, times(1)).onConsumerGroupStateTransition(ConsumerGroup.ConsumerGroupState.STABLE, null);
+ verify(context.metrics, times(1)).onClassicGroupStateTransition(null, STABLE);
+
+ // The new classic member 1 has a heartbeat timeout.
+ ScheduledTimeout<Void, Record> heartbeatTimeout = context.timer.timeout(
+ classicGroupHeartbeatKey(groupId, memberId1));
+ assertNotNull(heartbeatTimeout);
+ // The new rebalance has a groupJoin timeout.
+ ScheduledTimeout<Void, Record> groupJoinTimeout = context.timer.timeout(
+ classicGroupJoinKey(groupId));
+ assertNotNull(groupJoinTimeout);
+
+ // A new rebalance is triggered.
+ ClassicGroup classicGroup = context.groupMetadataManager.getOrMaybeCreateClassicGroup(groupId, false);
+ assertTrue(classicGroup.isInState(PREPARING_REBALANCE));
+ }
+
+ @Test
+ public void testLastClassicProtocolMemberRebalanceTimeoutInConsumerGroup() {
+ String groupId = "group-id";
+ String memberId1 = Uuid.randomUuid().toString();
+ String memberId2 = Uuid.randomUuid().toString();
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+ Uuid barTopicId = Uuid.randomUuid();
+ String barTopicName = "bar";
+ Uuid zarTopicId = Uuid.randomUuid();
+ String zarTopicName = "zar";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = Collections.singletonList(
+ new ConsumerGroupMemberMetadataValue.ClassicProtocol()
+ .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),
+ new TopicPartition(fooTopicName, 2),
+ new TopicPartition(barTopicName, 0),
+ new TopicPartition(barTopicName, 1)
+ )
+ ))))
+ );
+
+ ConsumerGroupMember member1 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(30000)
+ .setClassicMemberMetadata(new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSupportedProtocols(protocols))
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .build();
+ ConsumerGroupMember member2 = new ConsumerGroupMember.Builder(memberId2)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar", "zar"))
+ .setServerAssignorName("range")
+ .setRebalanceTimeoutMs(30000)
+ .setAssignedPartitions(mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .build();
+
+ // Consumer group with two members.
+ // Member 1 uses the classic protocol and member 2 uses the consumer protocol.
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConsumerGroupMigrationPolicy(ConsumerGroupMigrationPolicy.DOWNGRADE)
+ .withAssignors(Collections.singletonList(assignor))
+ .withMetadataImage(new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .addTopic(barTopicId, barTopicName, 3)
+ .addTopic(zarTopicId, zarTopicName, 1)
+ .addRacks()
+ .build())
+ .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
+ .withMember(member1)
+ .withMember(member2)
+ .withAssignment(memberId1, mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)))
+ .withAssignment(memberId2, mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5),
+ mkTopicAssignment(barTopicId, 2)))
+ .withAssignmentEpoch(10))
+ .build();
+
+ context.replay(RecordHelpers.newGroupSubscriptionMetadataRecord(groupId, new HashMap<String, TopicMetadata>() {
+ {
+ put(fooTopicName, new TopicMetadata(fooTopicId, fooTopicName, 6, mkMapOfPartitionRacks(6)));
+ put(barTopicName, new TopicMetadata(barTopicId, barTopicName, 3, mkMapOfPartitionRacks(3)));
+ put(zarTopicName, new TopicMetadata(zarTopicId, zarTopicName, 1, mkMapOfPartitionRacks(1)));
+ }
+ }));
+
+ context.commit();
+
+ // Prepare the new assignment.
+ assignor.prepareGroupAssignment(new GroupAssignment(
+ new HashMap<String, MemberAssignment>() {
+ {
+ put(memberId1, new MemberAssignment(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2),
+ mkTopicAssignment(barTopicId, 0, 1)
+ )));
+ put(memberId2, new MemberAssignment(mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5)
+ )));
+ }
+ }
+ ));
+
+ // Member 2 heartbeats with a different subscribedTopicNames. The assignor computes a new assignment
+ // where member 2 will need to revoke topic partition bar-2 thus transitions to the REVOKING state.
+ context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId2)
+ .setMemberEpoch(10)
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setTopicPartitions(Arrays.asList(
+ new ConsumerGroupHeartbeatRequestData.TopicPartitions()
+ .setTopicId(fooTopicId)
+ .setPartitions(Arrays.asList(0, 1, 2)),
+ new ConsumerGroupHeartbeatRequestData.TopicPartitions()
+ .setTopicId(barTopicId)
+ .setPartitions(Arrays.asList(0, 1))
+ ))
+ );
+
+ // Verify that there is a rebalance timeout.
+ context.assertRebalanceTimeout(groupId, memberId2, 30000);
+
+ // Advance time past the session timeout.
+ // Member 2 should be fenced from the group, thus triggering the downgrade.
+ context.sleep(30000 + 1);
+
+ verify(context.metrics, times(1)).onConsumerGroupStateTransition(ConsumerGroup.ConsumerGroupState.RECONCILING, null);
+ verify(context.metrics, times(1)).onClassicGroupStateTransition(null, STABLE);
+
+ // The new classic member 1 has a heartbeat timeout.
+ ScheduledTimeout<Void, Record> heartbeatTimeout = context.timer.timeout(
+ classicGroupHeartbeatKey(groupId, memberId1));
+ assertNotNull(heartbeatTimeout);
+ // The new rebalance has a groupJoin timeout.
+ ScheduledTimeout<Void, Record> groupJoinTimeout = context.timer.timeout(
Review Comment:
ditto about the style.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org