KAFKA-9130: KIP-518 Allow listing consumer groups per state #8238

mimaison · 2020-03-06T16:31:31Z

Committer Checklist (excluded from commit message)

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

dajac · 2020-03-16T10:41:28Z

@mimaison Is it already ready to be reviewed? I will help reviewing it.

mimaison · 2020-03-16T11:00:33Z

@dajac Thanks for offering your help! Yes it's pretty much complete and can start do be reviewed. I'm just planning to add a few more tests, early this week (hopefully today!)

dajac

Thanks for the PR. Overall, it looks pretty good. I have left a bunch of comments/questions.

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

clients/src/main/resources/common/message/ListGroupsRequest.json

clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java

core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala

core/src/main/scala/kafka/server/KafkaApis.scala

clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java

core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala

core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala

dajac

@mimaison I have made another pass over it and I have left few small comments and one question.

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

core/src/test/scala/unit/kafka/admin/DescribeConsumerGroupTest.scala

core/src/test/scala/unit/kafka/admin/ListConsumerGroupTest.scala

clients/src/main/resources/common/message/ListGroupsRequest.json

core/src/test/scala/unit/kafka/server/ListGroupsRequestTest.scala

clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java

dajac · 2020-04-16T10:06:24Z

@mimaison Let me know when I can do another pass on this PR.

mimaison · 2020-04-16T12:40:57Z

@dajac Thanks a lot for the feedback!
I've pushed an update addressing all your points. Go ahead if you can make another pass

dajac

@mimaison Thanks for the update. I just made a second pass over it and left few comments.

core/src/test/scala/unit/kafka/server/KafkaApisTest.scala

core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala

core/src/test/scala/unit/kafka/admin/DescribeConsumerGroupTest.scala

Co-authored-by: Mickael Maison <mickael.maison@gmail.com> Co-authored-by: Edoardo Comar <ecomar@uk.ibm.com>

mimaison · 2020-04-30T22:04:43Z

Thanks @dajac for the great feedback! I believe I've addressed all your comments now.

@cmccabe @rajinisivaram As you voted on the KIP, could you take a look?

dajac

@mimaison Thanks for the update. LGTM!

hachikuji

Thanks, left some comments.

hachikuji · 2020-05-18T20:04:11Z

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

+    private Optional<Set<ConsumerGroupState>> states = Optional.empty();
+
+    /**
+     * Only groups in these states will be returned by listConsumerGroups()


Probably worth adding a comment about broker compatibility with this API.

Can you address this comment?

hachikuji · 2020-05-18T20:11:43Z

clients/src/main/resources/common/message/ListGroupsRequest.json

  "flexibleVersions": "3+",
  "fields": [
+    { "name": "States", "type": "[]string", "versions": "4+", "tag": 0, "taggedVersions": "4+",


Sorry I missed this from the discussion, but why are we bumping the version if we are only adding tagged fields? Is it so that we can detect whether the capability is supported? If so, then I wonder why we don't make this a regular field.

Yes the version bump is necessary to detect if this field was supported.

As we're bumping the version and as you said in #8238 (comment) the overhead of the extra field on this API is not a concern, it's probably simpler to use a regular field.

hachikuji · 2020-05-18T20:13:50Z

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

+     * All groups with their states will be returned by listConsumerGroups()
+     */
+    public ListConsumerGroupsOptions inAnyState() {
+        this.states = Optional.of(EnumSet.allOf(ConsumerGroupState.class));


Hmm.. We have an UNKNOWN state in ConsumerGroupState in case the group coordinator adds a new state that the client isn't aware of. Currently we're going to pass this through the request, which is a bit odd. Furthermore, if the coordinator does add new states, we will be unable to see them using this API. I think it might be better to use a null list of states in the request to indicate that any state is needed.

That's a good point so I agree, it makes sense to return all states when null (or an empty list) is used.

This can also be argued for the state value in the response. Currently ConsumerGroupDescription stores the state as ConsumerGroupState so states the client isn't aware of are mapped to UNKNOWN so I'm doing the same in ConsumerGroupListing.

hachikuji · 2020-05-18T20:15:54Z

core/src/main/scala/kafka/server/KafkaApis.scala

+                val listedGroup = new ListGroupsResponseData.ListedGroup()
+                  .setGroupId(group.groupId)
+                  .setProtocolType(group.protocolType)
+                if (!states.isEmpty)


Why don't we always return the state? I don't think overhead is a huge concern for an api like this.

- Switch to use regular fields instead of tagged fields - Empty/null state filter now means all states

mimaison · 2020-05-22T09:21:30Z

Thanks @hachikuji for the feedback. You brought some interesting points that overall simplify the KIP/logic a bit.
While initially, it seemed a perfect use case for tagged fields, the need to bump the version to check for compatibility made it a bit confusing.

I've pushed an update following your suggestions to use a regular field and default to see all groups. I think it's easier to reason about and keeps the API simple. Let me know what you think. I'll update the KIP accordingly if we think that's the best option.

hachikuji

@mimaison The change makes sense to me if you are ok with it. I will review the patch again shortly.

mimaison · 2020-05-26T20:03:02Z

@hachikuji Yes I think it's better this way. I'll update the KIP

hachikuji

Thanks, left a few small comments.

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

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

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

hachikuji · 2020-05-28T05:53:44Z

clients/src/main/resources/common/message/ListGroupsResponse.json

@@ -34,7 +36,9 @@
      { "name": "GroupId", "type": "string", "versions": "0+", "entityType": "groupId",
        "about": "The group ID." },
      { "name": "ProtocolType", "type": "string", "versions": "0+",
-        "about": "The group protocol type." }
+        "about": "The group protocol type." },
+      { "name": "GroupState", "type": "string", "versions": "4+", "nullableVersions": "0+", "ignorable": true, "default": "null",


Is it intentional to use nullable versions 0+? I'm surprised the generator doesn't fail.

core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala

core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala

clients/src/main/resources/common/message/ListGroupsRequest.json

hachikuji · 2020-05-28T18:53:26Z

@mimaison fyi, if you can get the feedback addressed by tomorrow, I think we can still get this into 2.6. cc @rhauch

mimaison · 2020-05-28T20:58:19Z

@hachikuji Thanks for the feedback. I've pushed updates addressing your comments.

I've also updated the KIP accordingly and I'll send a message to the VOTE thread on the mailing list.

hachikuji

Left a few nitpicks. This is ready to merge once addressed.

hachikuji · 2020-05-28T23:29:53Z

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

        this.groupId = groupId;
        this.isSimpleConsumerGroup = isSimpleConsumerGroup;
+        this.state = state;


nit: usually we would write this is this.state = requireNonNull(state);

hachikuji · 2020-05-28T23:31:24Z

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

+    private Optional<Set<ConsumerGroupState>> states = Optional.empty();
+
+    /**
+     * Only groups in these states will be returned by listConsumerGroups()


Can you address this comment?

hachikuji · 2020-05-28T23:33:38Z

core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala


 object ConsumerGroupCommand extends Logging {

+  val allStates = ConsumerGroupState.values.toList


Is this used?

hachikuji · 2020-05-28T23:40:50Z

core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala

          }, s"Expected to be able to list $testGroupId")

+          TestUtils.waitUntilTrue(() => {
+            val options = new ListConsumerGroupsOptions()


This seems the same as the previous case. It might be more interesting if we request only groups that are stable? Then we have covered both successful and unsuccessful matching.

hachikuji · 2020-05-28T23:41:24Z

core/src/test/scala/unit/kafka/admin/ListConsumerGroupTest.scala

+    val service = getConsumerGroupService(cgcArgs)
+
+    val expectedListing = Set(
+        new ConsumerGroupListing(simpleGroup, true, Optional.of(ConsumerGroupState.EMPTY)),


nit: looks misaligned

hachikuji · 2020-05-28T23:42:50Z

core/src/test/scala/unit/kafka/admin/ListConsumerGroupTest.scala

+    assertEquals(Set(ConsumerGroupState.DEAD, ConsumerGroupState.COMPLETING_REBALANCE), result)
+
+    try {
+      ConsumerGroupCommand.consumerGroupStatesFromString("bad, wrong")


nit: we could use assertThrows or intercept for all of these

mimaison · 2020-05-29T09:40:07Z

Thanks @hachikuji. I've pushed another update addressing your comments.

I also sent an update to the thread on the mailing list.

hachikuji

LGTM

hachikuji · 2020-05-29T18:28:43Z

Note I am also picking this into 2.6.

Implementation of KIP-518: https://cwiki.apache.org/confluence/display/KAFKA/KIP-518%3A+Allow+listing+consumer+groups+per+state. Reviewers: David Jacot <djacot@confluent.io>, Jason Gustafson <jason@confluent.io> Co-authored-by: Mickael Maison <mickael.maison@gmail.com> Co-authored-by: Edoardo Comar <ecomar@uk.ibm.com>

* 'trunk' of github.com:apache/kafka: (36 commits) Remove redundant `containsKey` call in KafkaProducer (apache#8761) KAFKA-9494; Include additional metadata information in DescribeConfig response (KIP-569) (apache#8723) KAFKA-10061; Fix flaky `ReassignPartitionsIntegrationTest.testCancellation` (apache#8749) KAFKA-9130; KIP-518 Allow listing consumer groups per state (apache#8238) KAFKA-9501: convert between active and standby without closing stores (apache#8248) KAFKA-10056; Ensure consumer metadata contains new topics on subscription change (apache#8739) MINOR: Log the reason for coordinator discovery failure (apache#8747) KAFKA-10029; Don't update completedReceives when channels are closed to avoid ConcurrentModificationException (apache#8705) MINOR: remove unnecessary timeout for admin request (apache#8738) MINOR: Relax Percentiles test (apache#8748) MINOR: regression test for task assignor config (apache#8743) MINOR: Update documentation.html to refer to 2.6 (apache#8745) MINOR: Update documentation.html to refer to 2.5 (apache#8744) KAFKA-9673: Filter and Conditional SMTs (apache#8699) KAFKA-9971: Error Reporting in Sink Connectors (KIP-610) (apache#8720) KAFKA-10052: Harden assertion of topic settings in Connect integration tests (apache#8735) MINOR: Slight MetadataCache tweaks to avoid unnecessary work (apache#8728) KAFKA-9802; Increase transaction timeout in system tests to reduce flakiness (apache#8736) KAFKA-10050: kafka_log4j_appender.py fixed for JDK11 (apache#8731) KAFKA-9146: Add option to force delete active members in StreamsResetter (apache#8589) ... # Conflicts: # core/src/main/scala/kafka/log/Log.scala

* apache-github/2.6: (32 commits) KAFKA-10083: fix failed testReassignmentWithRandomSubscriptionsAndChanges tests (apache#8786) KAFKA-9945: TopicCommand should support --if-exists and --if-not-exists when --bootstrap-server is used (apache#8737) KAFKA-9320: Enable TLSv1.3 by default (KIP-573) (apache#8695) KAFKA-10082: Fix the failed testMultiConsumerStickyAssignment (apache#8777) MINOR: Remove unused variable to fix spotBugs failure (apache#8779) MINOR: ChangelogReader should poll for duration 0 for standby restore (apache#8773) KAFKA-10030: Allow fetching a key from a single partition (apache#8706) Kafka-10064 Add documentation for KIP-571 (apache#8760) MINOR: Code cleanup and assertion message fixes in Connect integration tests (apache#8750) KAFKA-9987: optimize sticky assignment algorithm for same-subscription case (apache#8668) KAFKA-9392; Clarify deleteAcls javadoc and add test for create/delete timing (apache#7956) KAFKA-10074: Improve performance of `matchingAcls` (apache#8769) KAFKA-9494; Include additional metadata information in DescribeConfig response (KIP-569) (apache#8723) KAFKA-10056; Ensure consumer metadata contains new topics on subscription change (apache#8739) KAFKA-10029; Don't update completedReceives when channels are closed to avoid ConcurrentModificationException (apache#8705) KAFKA-10061; Fix flaky `ReassignPartitionsIntegrationTest.testCancellation` (apache#8749) KAFKA-9130; KIP-518 Allow listing consumer groups per state (apache#8238) KAFKA-9501: convert between active and standby without closing stores (apache#8248) MINOR: Relax Percentiles test (apache#8748) MINOR: regression test for task assignor config (apache#8743) ...

mimaison marked this pull request as ready for review March 16, 2020 17:23

dajac reviewed Mar 17, 2020

View reviewed changes

mimaison force-pushed the kip-518 branch from 37e744e to 5ffcfc5 Compare March 26, 2020 17:02

dajac reviewed Mar 27, 2020

View reviewed changes

mimaison force-pushed the kip-518 branch from 13f5051 to 05f3fd5 Compare April 16, 2020 10:35

dajac reviewed Apr 21, 2020

View reviewed changes

mimaison and others added 8 commits April 30, 2020 15:28

KAFKA-9130: KIP-518 Allow listing consumer groups per state

88e7a0c

Co-authored-by: Mickael Maison <mickael.maison@gmail.com> Co-authored-by: Edoardo Comar <ecomar@uk.ibm.com>

Add test in admin e2e

143514c

First pass at addressing feedback

badbc1a

Addressed comments

b243a61

addressed a couple of comments and CLI's list state formatting

145b689

Address latest feedback

a0247b0

testDescribeWithMultipleSubActions

4bed4d3

Address feedback

d6102e8

mimaison force-pushed the kip-518 branch from b1f1d73 to d6102e8 Compare April 30, 2020 14:52

Improve error reporting

5854e56

dajac reviewed May 6, 2020

View reviewed changes

hachikuji reviewed May 18, 2020

View reviewed changes

Addressed feedback

1f032e7

- Switch to use regular fields instead of tagged fields - Empty/null state filter now means all states

hachikuji reviewed May 26, 2020

View reviewed changes

hachikuji reviewed May 28, 2020

View reviewed changes

mimaison added 2 commits May 28, 2020 13:38

Address some of the feedback

499b15b

Address some of the feedback

56f839a

Address comments

5e18e5c

hachikuji reviewed May 28, 2020

View reviewed changes

Address last round of reviews

acb64f3

hachikuji approved these changes May 29, 2020

View reviewed changes

hachikuji merged commit fe948d3 into apache:trunk May 29, 2020

mimaison deleted the kip-518 branch May 29, 2020 18:48


		object ConsumerGroupCommand extends Logging {

		val allStates = ConsumerGroupState.values.toList

KAFKA-9130: KIP-518 Allow listing consumer groups per state #8238

KAFKA-9130: KIP-518 Allow listing consumer groups per state #8238

Conversation

mimaison commented Mar 6, 2020

Committer Checklist (excluded from commit message)

dajac commented Mar 16, 2020

mimaison commented Mar 16, 2020 • edited Loading

dajac left a comment

Choose a reason for hiding this comment

dajac left a comment

Choose a reason for hiding this comment

dajac commented Apr 16, 2020

mimaison commented Apr 16, 2020

dajac left a comment

Choose a reason for hiding this comment

mimaison commented Apr 30, 2020

dajac left a comment

Choose a reason for hiding this comment

hachikuji left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mimaison commented May 22, 2020

hachikuji left a comment

Choose a reason for hiding this comment

mimaison commented May 26, 2020

hachikuji left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hachikuji commented May 28, 2020

mimaison commented May 28, 2020

hachikuji left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mimaison commented May 29, 2020

hachikuji left a comment

Choose a reason for hiding this comment

hachikuji commented May 29, 2020

mimaison commented Mar 16, 2020 •

edited

Loading