dajac commented on code in PR #18224:
URL: https://github.com/apache/kafka/pull/18224#discussion_r1908411206
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/classic/ClassicGroupTest.java:
##########
@@ -1590,7 +1590,7 @@ public void testFromConsumerGroupWithoutJoiningMember() {
ClassicGroup classicGroup = ClassicGroup.fromConsumerGroup(
consumerGroup,
- memberId2,
+ Collections.singleton(memberId2),
Review Comment:
nit: You could use `List.of` now.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -3001,17 +3010,52 @@ private <T> CoordinatorResult<T, CoordinatorRecord>
consumerGroupFenceMember(
ConsumerGroupMember member,
T response
) {
+ return consumerGroupFenceMembers(group, Set.of(member), response);
+ }
+
+ /**
+ * Fences members from a consumer group and maybe downgrade the consumer
group to a classic group.
+ *
+ * @param group The group.
+ * @param members The members.
+ * @param response The response of the CoordinatorResult.
+ *
+ * @return The CoordinatorResult to be applied.
+ */
+ private <T> CoordinatorResult<T, CoordinatorRecord>
consumerGroupFenceMembers(
+ ConsumerGroup group,
+ Set<ConsumerGroupMember> members,
+ T response
+ ) {
+ if (members.isEmpty()) {
+ // No members to fence. Don't bump the group epoch.
+ return new CoordinatorResult<>(Collections.emptyList(), response);
+ }
+
+ Set<String> memberIds = new HashSet<String>();
+ for (ConsumerGroupMember member : members) {
+ memberIds.add(member.memberId());
+ }
Review Comment:
Not sure whether it is worth it but have you considered passing
`ConsumerGroupMember`s to `validateOnlineDowngradeWithFencedMembers` and
`convertToClassicGroup`?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/classic/ClassicGroupTest.java:
##########
@@ -1459,7 +1459,7 @@ public void testFromConsumerGroupWithJoiningMember() {
ClassicGroup classicGroup = ClassicGroup.fromConsumerGroup(
consumerGroup,
- memberId2,
+ Collections.singleton(memberId2),
Review Comment:
nit: You could use `List.of` now.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -984,27 +984,36 @@ private boolean
validateOnlineDowngradeWithReplacedMemberId(
* Creates a ClassicGroup corresponding to the given ConsumerGroup.
*
* @param consumerGroup The converted ConsumerGroup.
- * @param leavingMemberId The leaving member that triggers the downgrade
validation.
+ * @param leavingMemberIds The leaving member(s) that triggered the
downgrade validation.
* @param joiningMember The newly joined member if the downgrade is
triggered by static member replacement.
+ * When not null, leavingMemberIds must contain
the single member being replaced.
* @param records The record list to which the conversion
records are added.
*/
private void convertToClassicGroup(
ConsumerGroup consumerGroup,
- String leavingMemberId,
+ Set<String> leavingMemberIds,
ConsumerGroupMember joiningMember,
List<CoordinatorRecord> records
) {
+ if (joiningMember != null && leavingMemberIds.size() != 1) {
+ throw new IllegalArgumentException(
+ String.format("joiningMember is not null, but leavingMemberIds
contains %d members.", leavingMemberIds.size())
+ );
+ }
Review Comment:
I wonder if we could decouple the dependency between the two variables in
order to simplify the code. For instance, we could get the previous member id
of `joiningMember` from `consumerGroup` based on its instance id. Would it be
possible?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
