KAFKA-17750: Extend kafka-consumer-groups command line tool to support new consumer group (part 2) by FrankYang0529 · Pull Request #18034 · apache/kafka

FrankYang0529 · 2024-12-04T12:16:31Z

Add fields groupEpoch and targetAssignmentEpoch to ConsumerGroupDescription.java.
Add fields memberEpoch and upgraded to MemberDescription.java.
Add assertion to PlaintextAdminIntegrationTest#testDescribeClassicGroups to make sure member in classic group returns upgraded as Optional.empty.
Add new case testConsumerGroupWithMemberMigration to PlaintextAdminIntegrationTest to make sure migration member has correct upgraded value. Add assertion for groupEpoch, targetAssignmentEpoch, memberEpoch as well.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

dajac

@FrankYang0529 Thanks for the patch. I left a few suggestions and questions for consideration.

dajac · 2024-12-05T09:29:23Z

+            Optional.empty(), Optional.empty());
    }

    public ConsumerGroupDescription(String groupId,


Was this constructor introduced in 4.0 only? It seems to be the case but we need to double check. If it is not, we would need to introduce a new one in order to not break existing usages.

This constructor is introduced in 4.0. It is not existent in 3.9.

kafka/clients/src/main/java/org/apache/kafka/clients/admin/ConsumerGroupDescription.java

Lines 45 to 81 in 84caaa6

public ConsumerGroupDescription(String groupId,

boolean isSimpleConsumerGroup,

Collection<MemberDescription> members,

String partitionAssignor,

ConsumerGroupState state,

Node coordinator) {

this(groupId, isSimpleConsumerGroup, members, partitionAssignor, state, coordinator, Collections.emptySet());

}

public ConsumerGroupDescription(String groupId,

boolean isSimpleConsumerGroup,

Collection<MemberDescription> members,

String partitionAssignor,

ConsumerGroupState state,

Node coordinator,

Set<AclOperation> authorizedOperations) {

this(groupId, isSimpleConsumerGroup, members, partitionAssignor, GroupType.CLASSIC, state, coordinator, authorizedOperations);

}

public ConsumerGroupDescription(String groupId,

boolean isSimpleConsumerGroup,

Collection<MemberDescription> members,

String partitionAssignor,

GroupType type,

ConsumerGroupState state,

Node coordinator,

Set<AclOperation> authorizedOperations) {

this.groupId = groupId == null ? "" : groupId;

this.isSimpleConsumerGroup = isSimpleConsumerGroup;

this.members = members == null ? Collections.emptyList() :

Collections.unmodifiableList(new ArrayList<>(members));

this.partitionAssignor = partitionAssignor == null ? "" : partitionAssignor;

this.type = type;

this.state = state;

this.coordinator = coordinator;

this.authorizedOperations = authorizedOperations;

}

We can remove the two constructors directly, as doing so does not result in any compatibility issues.

dajac · 2024-12-05T09:30:24Z

+    private final Optional<Integer> memberEpoch;
+    private final Optional<Boolean> isClassic;

    public MemberDescription(String memberId,


Here, I suppose that we must introduce a new overload of the constructor with the new arguments in order to not break compatibility of existing usages.

Thanks for reminder. Updated it.

Have we considered deprecating the other constructors? It seems odd to have so many constructors.

Updated both PR and KIP. Thanks for the suggestion.

dajac · 2024-12-05T09:30:42Z

-    ) {
+        Optional<MemberAssignment> targetAssignment,
+        Optional<Integer> memberEpoch,
+        Optional<Boolean> isClassic) {


nit: Let's keep the closing ) on a new line as it was before.

dajac · 2024-12-05T09:32:15Z

+    }
+
+    /**
+     * The flag indicating whether a member is classic.


nit: How about The flag indicating whether a member with a consumer group uses the classic protocol. It indicates that the member was perhaps not upgraded?

dajac · 2024-12-05T09:36:02Z

+    /**
+     * The flag indicating whether a member is classic.
+     */
+    public Optional<Boolean> isClassic() {


Sorry for coming back on this one but I just thought about an alternative. How about calling it nonUpgraded? We could set it to true when the group is a consumer group and the member uses classic protocol. We could also only set it when the group is a consumer group vs a classic group. This may be even more explicit for users. What do you think? @chia7712 Do you have any thoughts too?

How about calling it nonUpgraded? We could set it to true when the group is a consumer group and the member uses classic protocol.

How about using "upgraded"? it is true only if the group is a consumer group and the member uses the consumer protocol.

That works too. Something as follow?

Classic Group -> Optional.empty

Consumer Group -> Optional.empty if unknown

Consumer Group -> Optional.of(false) if classic

Consumer Group -> Optional.of(true) if consumer

Thanks for the suggestion. Updated PR and KIP.

chia7712

@FrankYang0529 thanks for this patch

chia7712 · 2024-12-05T20:51:56Z

+            Optional.empty(), Optional.empty());
    }

    public ConsumerGroupDescription(String groupId,


We can remove the two constructors directly, as doing so does not result in any compatibility issues.

chia7712 · 2024-12-05T20:54:22Z

-            ", targetAssignment=" + targetAssignment + ")";
+            ", targetAssignment=" + targetAssignment +
+            ", memberEpoch=" + memberEpoch.orElse(null) +
+            ", isClassic=" + upgraded.orElse(null) +


isClassic -> upgraded

chia7712 · 2024-12-05T20:58:18Z

+    private final Optional<Integer> memberEpoch;
+    private final Optional<Boolean> isClassic;

    public MemberDescription(String memberId,


Have we considered deprecating the other constructors? It seems odd to have so many constructors.

dajac

@FrankYang0529 Thanks for the update. I left some more comments for consideration.

dajac · 2024-12-06T08:44:30Z

+    public MemberDescription(String memberId,
+                             Optional<String> groupInstanceId,
+                             String clientId,
+                             String host,
+                             MemberAssignment assignment,
+                             Optional<MemberAssignment> targetAssignment
+    ) {


nit: I think that we are mixing two code styles here. We would usually use:

public MemberDescription( String memberId, Optional<String> groupInstanceId, String clientId, String host, MemberAssignment assignment, Optional<MemberAssignment> targetAssignment ) {

or

public MemberDescription(String memberId, Optional<String> groupInstanceId, String clientId, String host, MemberAssignment assignment, Optional<MemberAssignment> targetAssignment) {

I personally prefer the first one but it is up to you.

dajac · 2024-12-06T08:50:29Z

+     * True for consumer member.
+     * False for classic member
+     * Empty for unknown or members in classic group.


nit: I suggest to reference the group type in the java doc in order to be crystal clear. e.g.

The flag indicating whether a member within a {CONSUMER} group uses the {CONSUMER} protocol. The optional is set to true if it does, to false if it does not, and to empty if it is unknown or if the group is a {CLASSIC} group.

dajac · 2024-12-06T08:54:01Z

-            )))
+            ))),
+            Optional.of(10),
+            Optional.of(true)


Should we cover the unknown and the classic cases in the tests too?

Add classic case to testSuccessfulHandleConsumerGroupResponse. For unknown case, I think it's covered by testSuccessfulHandleClassicGroupResponse. Thanks.

dajac · 2024-12-06T08:57:13Z

+          assertEquals(groupType == GroupType.CLASSIC, testGroupDescription.groupEpoch.isEmpty)
+          assertEquals(groupType == GroupType.CLASSIC, testGroupDescription.targetAssignmentEpoch.isEmpty)


Would it be possible to actually verify the values too?

I would like to verify values here, but the case is only for classic now. It's waiting for https://issues.apache.org/jira/browse/KAFKA-17960. Could we add it after we reopen consumer for this case? Thanks.

Create a minor PR for this: #18203

dajac · 2024-12-06T08:57:34Z

+          members.asScala.foreach(member => {
+            assertEquals(testClientId, member.clientId)
+            assertEquals(if (groupType == GroupType.CLASSIC) Optional.empty else Optional.of(true), member.upgraded)
+          })


nit: foreach(member => { -> foreach { member =>

dajac · 2024-12-06T09:02:16Z

+
+      val classicMember = group.members.asScala.find(_.clientId == testClassicClientId)
+      assertTrue(classicMember.isDefined)
+      assertTrue(classicMember.get.memberEpoch.isPresent && classicMember.get.memberEpoch.get > 0)


assertEquals(Optional.of(2), classicMember.get.memberEpoch).

dajac · 2024-12-06T09:03:00Z

+      val classicMember = group.members.asScala.find(_.clientId == testClassicClientId)
+      assertTrue(classicMember.isDefined)
+      assertTrue(classicMember.get.memberEpoch.isPresent && classicMember.get.memberEpoch.get > 0)
+      assertTrue(classicMember.get.upgraded.isPresent && !classicMember.get.upgraded.get)


assertEquals(Optional.of(false), classicMember.get.upgraded).

dajac · 2024-12-06T09:03:07Z

+      assertTrue(classicMember.get.memberEpoch.isPresent && classicMember.get.memberEpoch.get > 0)
+      assertTrue(consumerMember.get.upgraded.isPresent && consumerMember.get.upgraded.get)


dajac · 2024-12-06T09:04:17Z

    }

+    /**
+     * The epoch of the group member.


nit: Let's also mention that this is only provided in the CONSUMER group case.

dajac · 2024-12-06T09:04:41Z

+    /**
+     * The epoch of the consumer group.
+     */
+    public Optional<Integer> groupEpoch() {
+        return groupEpoch;
+    }
+
+    /**
+     * The epoch of the target assignment.
+     */
+    public Optional<Integer> targetAssignmentEpoch() {
+        return targetAssignmentEpoch;
+    }


Let's also mention here that they are only provided when the type is consumer.

…t new consumer group (part 2) Signed-off-by: PoAn Yang <payang@apache.org>

FrankYang0529 · 2024-12-10T09:08:58Z

Hi @dajac, could you please help me review this PR when you have time? I would like to keep working on last part of KIP-1099 (tool). Thanks.

dajac

lgtm, thanks.

dajac · 2024-12-10T13:02:31Z

@FrankYang0529 Done :)

chia7712 · 2024-12-11T13:11:20Z

@FrankYang0529 Could you please file a minor to fix the deprecation warning introduced by this PR?

> Task :tools:compileTestJava
Note: /home/chia7712/project/kafka/tools/src/test/java/org/apache/kafka/tools/consumer/group/ConsumerGroupServiceTest.java uses or overrides a deprecated API.
Note: Recompile with -Xlint:deprecation for details.

FrankYang0529 · 2024-12-11T13:16:22Z

Could you please file a minor to fix the deprecation warning introduced by this PR?

Thanks for finding this. Create a PR for it.

#18139

…t new consumer group (part 2) (apache#18034) * Add fields `groupEpoch` and `targetAssignmentEpoch` to `ConsumerGroupDescription.java`. * Add fields `memberEpoch` and `upgraded` to `MemberDescription.java`. * Add assertion to `PlaintextAdminIntegrationTest#testDescribeClassicGroups` to make sure member in classic group returns `upgraded` as `Optional.empty`. * Add new case `testConsumerGroupWithMemberMigration` to `PlaintextAdminIntegrationTest` to make sure migration member has correct `upgraded` value. Add assertion for `groupEpoch`, `targetAssignmentEpoch`, `memberEpoch` as well. Reviewers: David Jacot <djacot@confluent.io> Signed-off-by: PoAn Yang <payang@apache.org>

github-actions bot added core Kafka Broker clients labels Dec 4, 2024

FrankYang0529 mentioned this pull request Dec 4, 2024

KAFKA-17750: Extend kafka-consumer-groups command line tool to support new consumer group #17872

Closed

3 tasks

FrankYang0529 requested a review from dajac December 4, 2024 15:48

dajac reviewed Dec 5, 2024

View reviewed changes

FrankYang0529 force-pushed the KAFKA-17750-admin branch 2 times, most recently from bf9ea6b to b61e4b5 Compare December 5, 2024 15:07

chia7712 reviewed Dec 5, 2024

View reviewed changes

FrankYang0529 force-pushed the KAFKA-17750-admin branch from b61e4b5 to 22f9dd6 Compare December 6, 2024 06:47

github-actions bot added the tools label Dec 6, 2024

FrankYang0529 force-pushed the KAFKA-17750-admin branch 2 times, most recently from 1486622 to 9361113 Compare December 6, 2024 07:10

dajac reviewed Dec 6, 2024

View reviewed changes

FrankYang0529 force-pushed the KAFKA-17750-admin branch 3 times, most recently from 263c615 to 69bd0f3 Compare December 6, 2024 13:48

FrankYang0529 requested review from chia7712 and dajac December 7, 2024 05:44

KAFKA-17750: Extend kafka-consumer-groups command line tool to suppor…

9c37588

…t new consumer group (part 2) Signed-off-by: PoAn Yang <payang@apache.org>

FrankYang0529 force-pushed the KAFKA-17750-admin branch from 69bd0f3 to 9c37588 Compare December 9, 2024 13:39

dajac approved these changes Dec 10, 2024

View reviewed changes

dajac merged commit c8380ae into apache:trunk Dec 10, 2024

FrankYang0529 deleted the KAFKA-17750-admin branch December 10, 2024 13:11

FrankYang0529 mentioned this pull request Dec 11, 2024

MINOR: suppress deprecation warnings for MemberDescription #18139

Merged

3 tasks

FrankYang0529 mentioned this pull request Dec 16, 2024

MINOR: add assertion about groupEpoch and targetAssignmentEpoch to testConsumerGroups #18203

Merged

3 tasks

	public ConsumerGroupDescription(String groupId,
	boolean isSimpleConsumerGroup,
	Collection<MemberDescription> members,
	String partitionAssignor,
	ConsumerGroupState state,
	Node coordinator) {
	this(groupId, isSimpleConsumerGroup, members, partitionAssignor, state, coordinator, Collections.emptySet());
	}

	public ConsumerGroupDescription(String groupId,
	boolean isSimpleConsumerGroup,
	Collection<MemberDescription> members,
	String partitionAssignor,
	ConsumerGroupState state,
	Node coordinator,
	Set<AclOperation> authorizedOperations) {
	this(groupId, isSimpleConsumerGroup, members, partitionAssignor, GroupType.CLASSIC, state, coordinator, authorizedOperations);
	}

	public ConsumerGroupDescription(String groupId,
	boolean isSimpleConsumerGroup,
	Collection<MemberDescription> members,
	String partitionAssignor,
	GroupType type,
	ConsumerGroupState state,
	Node coordinator,
	Set<AclOperation> authorizedOperations) {
	this.groupId = groupId == null ? "" : groupId;
	this.isSimpleConsumerGroup = isSimpleConsumerGroup;
	this.members = members == null ? Collections.emptyList() :
	Collections.unmodifiableList(new ArrayList<>(members));
	this.partitionAssignor = partitionAssignor == null ? "" : partitionAssignor;
	this.type = type;
	this.state = state;
	this.coordinator = coordinator;
	this.authorizedOperations = authorizedOperations;
	}

		assertEquals(groupType == GroupType.CLASSIC, testGroupDescription.groupEpoch.isEmpty)
		assertEquals(groupType == GroupType.CLASSIC, testGroupDescription.targetAssignmentEpoch.isEmpty)

		assertTrue(classicMember.get.memberEpoch.isPresent && classicMember.get.memberEpoch.get > 0)
		assertTrue(consumerMember.get.upgraded.isPresent && consumerMember.get.upgraded.get)

Conversation

FrankYang0529 commented Dec 4, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Committer Checklist (excluded from commit message)

Uh oh!

dajac left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

chia7712 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dajac left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

FrankYang0529 commented Dec 10, 2024

Uh oh!

dajac left a comment

Choose a reason for hiding this comment

Uh oh!

dajac commented Dec 10, 2024

FrankYang0529 commented Dec 4, 2024 •

edited

Loading