KIP-222 - Add Consumer Group operations to Admin API by jeqo · Pull Request #4454 · apache/kafka

jeqo · 2018-01-21T23:18:15Z

KIP: https://cwiki.apache.org/confluence/display/KAFKA/KIP-222+-+Add+Consumer+Group+operations+to+Admin+API

hachikuji · 2018-01-25T18:17:38Z

I did not take a deep look yet, but there are some problems that will probably require changes to the KIP. The group management protocol handles more than just consumer groups. Connect also uses it for its own group management. Connect groups do not have topic partition assignments, so the current MemberAssignment object does not work generally. I see two options:

Get rid of the generic group APIs and only support consumer groups for now. In other words, if the protocol type is not "consumer," we just raise an exception. This would be a bummer, but it's probably the only way to make your current API work.
Change the API to accommodate different group types. This will mean making the MemberAssignment generic so that it does not depend on the consumer assignment.

Additionally, I think there are a few other points to consider:

The describe API has the ability to fetch the assignment currently, but not the consumer's subscription. I think we should consider exposing this.
Users typically expect "simple" consumer groups which do not use group management to be included when listing consumer groups. In this case, the protocol type will be empty.
No fault of your's, but concurrently, there was a KIP to add support for deleting groups. We should consider adding an AdminClient API to support this (it could come later if needed, but there shouldn't be anything too tricky from an API perspective, so we may as well do it here.

hachikuji · 2018-01-26T19:32:54Z

Just to clarify, if we go with the first option that I suggested above, we probably don't need a revote on the KIP. I doubt anyone will object if you just send an email to the discussion thread with the changes. Specifically, we should do the following:

Drop the describeGroups API.
Change the name of listGroups to listConsumerGroups.
Change listGroupOffsets to listConsumerGroupOffsets.

Otherwise, the logic stays the same. We would just need to filter groups which do not have the "consumer" protocol type. Note that I do think it's important to account for simple consumer groups as well. These can be identified as groups with an empty protocol type.

jeqo · 2018-01-28T12:48:44Z

@hachikuji thanks for your feedback!

Putting Group Management (connect and consumer groups) together as part of the changes to the API was one of the important points discussed in this KIP.

As I understand so far: connect group details are different from consumer group details, so we will need to define different APIs for both instead of trying to make them generic as your second option propose.

Following what your first option proposed we can add:

Drop describeGroups API (and change it to describeConsumerGroups?)
Change listGroups to listConsumerGroups
Change listGroupOffsets to listConsumerGroupOffsets.
add listConnectGroups?
I don't know how Connect Groups use Group Management API in detail but if they have offset management also, we can add listConnectGroupOffsets and describeConnectGroups API

Simple Consumer Groups will be added to listConsumerGroupOffsets filtering protocol type = "consumer" or empty right? I can propose it as part of the KIP discussion to check if it make sense to add them as part of the same response.

And about adding DeleteGroups API I would be happy to, but I think we would need to hold until #4479 is merged right?

hachikuji · 2018-01-29T16:40:27Z

@jeqo I'd probably suggest sticking with consumer groups initially if we want to have any hope of getting this into the release. So if we have 1-3 in your list, we can leave general group APIs for future work. The nice thing about sticking with consumer groups is that it lets us expose the inner details from a consumer perspective. Here a couple concrete suggestions:

Change GroupListing to ConsumerGroupListing. If we do that, then we don't need to expose "protocol type" at all. Maybe we can replace it with a simple flag which indicates whether it is a simple group or not.
Change GroupDescription to ConsumerGroupDescription. We can do the same thing with the "protocol type." We should also expose the group "protocol," which for consumers, just means the assignment strategy.

As far as DeleteGroups, we should probably only consider it if we manage to get 1-3 merged.

Thoughts?

jeqo · 2018-01-30T15:23:26Z

@hachikuji I have applied changes mentioned above. Let me know WDYT, and if I should take this back to the KIP discussion to gather some feedback. I could propose to move Connect Group Management and Delete Consumer Group to a next KIP.

hachikuji

@jeqo Thanks for the updates. From an API perspective, this is looking good. However, the request logic needs a bit more work. I would suggest having a look at kafka.admin.AdminClient to understand the logic behind these requests.

I don't think it's reasonable to get this patch in the next day, so let's plan on getting this into the next feature release. With a little more time, we can also consider the delete API and non-consumer groups (potentially in a separate KIP).

hachikuji · 2018-01-30T19:54:24Z

+    private final String groupId;
+    private final boolean isSimpleConsumerGroup;
+    private final List<MemberDescription> members;
+    private final String protocol;


Maybe we can use the consumer-specific term "partitionAssignor"?

hachikuji · 2018-01-30T21:53:12Z

+        }
+        final long now = time.milliseconds();
+        runnable.call(new Call("describeConsumerGroups", calcDeadlineMs(now, options.timeoutMs()),
+            new ControllerNodeProvider()) {


The DescribeGroup API has to be sent to the group coordinator, which is potentially a different node for each group. You use the FindCoordinator API in order to lookup the coordinator for a given group. The logic should be something like this:

For each group in the request, send a FindCoordinator request to any node in the cluster.

Group the results by coordinator id.

Send DescribeGroups to each coordinator from 2.

Ideally, we should also handle retries correctly. It could happen that the coordinator moves to another node by the time we send DescribeGroups. In this case, the error code will be NOT_COORDINATOR. We should handle this by looking up the coordinator again.

OK, I have pushed draft following this description.

hachikuji · 2018-01-30T21:55:51Z

+    }
+
+    @Override
+    public ListConsumerGroupsResult listConsumerGroups(ListConsumerGroupsOptions options) {


Unfortunately, the consumer groups are not aggregated in the same way that topic metadata is. To get all the groups in the cluster, you have to send the ListGroups request to all nodes.

I assume this is the same as DescribeConsumerGroup API. If changes are fine, I will updated this an listConsumerGroupOffsets

The logic is different for ListGroups. We have to send a separate request to every broker in the cluster and then aggregate the results.

@hachikuji trying to test this I find out that the result from this operation has to change (similar to how the Scala API is doing now) to respond with Map<Node, KafkaFuture<List<ConsumerGroupListing>>> instead, to list all responses from nodes. Does this make sense? I can update the KIP regarding this.

I'd vote for doing the flattening internally than expose the node information in the results, as this is supposed to be internal implementation details that is better not leaking out. The Scala API returning Map[Node, List[GroupOverview]] was not a well designed one in my hind-sight.

hachikuji · 2018-01-30T21:56:25Z

+    }
+
+    @Override
+    public ListConsumerGroupOffsetsResult listConsumerGroupOffsets(final String groupId, final ListConsumerGroupOffsetsOptions options) {


Like DescribeGroups, we need to find the coordinator for the group to send the OffsetFetch request to.

asfgit · 2018-02-06T01:01:38Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-test-coverage/349/

…-describe-consumer-group

hachikuji

Apologies for the review delay. I will look in more depth once the 1.1 release has been finalized. In the meantime, I noticed there were no test cases. It would be a good idea to add both unit and integration tests for the new functionality.

guozhangwang · 2018-04-09T17:55:02Z

Hmm. Why does there have to be a futures map for listConsumerGroups in the first place? listConsumerGroups just returns a future which contains the list of all the consumer groups in the cluster, right? I don't see why a map needs to be involved. On the other hand, if you want to learn more about a specific consumer group, you use describeConsumerGroup, which does know the groups it is getting information about ahead of time. This is analogous to listTopics versus describeTopics. listTopics doesn't have a map. describeTopics does.

Just to clarify, we will not expose the map inside the future to user; the map will be maintained internally only during the second round-trip as we need to check if we have received the responses from all found broker nodes, but as I suggested we should flatten it before exposing to the users. So the returned type of listConusmer will be KafkaFuture<Collection<ConsumerGroupListing>>, while the returned type of describeConsumer will be Map<String, KafkaFuture<ConsumerGroupDescription>> (note for the latter, it is because users already specifically give the list of consumer group ids, like the deleteRecords api)

…uests (apache#4842) This patch fixes an edge case in producer shutdown which prevents `close()` from completing due to a pending request which will never be sent due to shutdown initiation. I have added a test case which reproduces the scenario. Reviewers: Apurva Mehta <apurva@confluent.io>, Ismael Juma <ismael@juma.me.uk>

Added consumer only workload to Trogdor. The topics must already be pre-populated. The spec lets the user request topic pattern and range of partitions to assign to [startPartition, endPartition]. Reviewers: Colin P. Mccabe <cmccabe@confluent.io>, Rajini Sivaram <rajinisivaram@googlemail.com>

…e#4845) The toString() for ConfigResource was using { } instead of ( ) which is inconsistent with the existing toStrings in the code, while toString for Resource was using a mix of ( and }.

guozhangwang

@jeqo I left some more comments on the return types of other Result classes.

Also the jenkins failures seem consistent, need to fix them before merging as well.

guozhangwang · 2018-04-10T17:12:51Z

+        this.futures = futures;
+    }
+
+    public KafkaFuture<Map<String, KafkaFuture<Void>>> values() {


I think we can just have one function between values and groups here. I'd suggest we use

public Map<TopicPartition, KafkaFuture<Void>> deletedGroups()

guozhangwang · 2018-04-10T17:15:41Z

+        return futures;
+    }
+
+    public KafkaFuture<Collection<String>> names() {


Where is this function used?

I'd suggest we only keep one function, i.e.

public Map<TopicPartition, KafkaFuture< ConsumerGroupDescription >> DescribeConsumerGroupsResult#values()

guozhangwang · 2018-04-10T17:16:29Z

+        });
+    }
+
+    public KafkaFuture<Collection<KafkaFuture<Void>>> all() {


For all() function, its returned type should be KafkaFuture<Void>; ditto for other two Results as well.

guozhangwang · 2018-04-10T17:21:13Z

+    /**
+     * Return a future which yields a map of topic partitions to OffsetAndMetadata objects.
+     */
+    public KafkaFuture<Map<TopicPartition, OffsetAndMetadata>> partitionsToOffsetAndMetadata() {


I'd suggest only keep partitionsToOffsetAndMetadata here.

…che#4848) Move creation of quota callback instance out of KafkaConfig constructor to QuotaFactory.instantiate to avoid creating a callback instance for every KafkaConfig since we create temporary KafkaConfigs during dynamic config updates. Reviewers: Jun Rao <junrao@gmail.com>

…trol flag (apache#4849)

guozhangwang · 2018-04-11T15:53:38Z

retest this please

Reviewers: Jun Rao <junrao@gmail.com>

…249) (apache#4427) Reviewers: Jun Rao <junrao@gmail.com>

guozhangwang · 2018-04-11T18:01:57Z

@jeqo could you rebase your PR?

guozhangwang · 2018-04-11T20:10:13Z

@jeqo could you rebase your PR?

NVM, I've done the rebase, and will merge i once the tests passed locally.

guozhangwang · 2018-04-11T21:19:31Z

Unit test passed locally, merging now

jeqo · 2018-04-11T22:15:55Z

@guozhangwang thanks!! :)

KIP: https://cwiki.apache.org/confluence/display/KAFKA/KIP-222+-+Add+Consumer+Group+operations+to+Admin+API Author: Jorge Quilcate Otoya <quilcate.jorge@gmail.com> Author: Jorge Esteban Quilcate Otoya <quilcate.jorge@gmail.com> Author: Guozhang Wang <wangguoz@gmail.com> Reviewers: Colin Patrick McCabe <colin@cmccabe.xyz>, Guozhang Wang <wangguoz@gmail.com> Closes apache#4454 from jeqo/feature/admin-client-describe-consumer-group

jeqo added 10 commits November 4, 2017 23:44

draft describeConsumerGroups on KafkaAdminClient

3c327ed

document changes

5cfab0b

add listConsumerGroups

5a4b5a5

consumer group and instance class documentation

6468069

renaming to support groups (including connect groups)

f888167

merge

4eac264

add list groups/consumer groups and list group offsets

6483c59

increase class fan out, add test method unsupported exceptions

d0f8596

refactor listGroupOffsets

61e4780

add support for offsetandmetadata as part of listGroupOffsets

9cd4c51

SivakumarHaridas approved these changes Jan 22, 2018

View reviewed changes

jeqo added 5 commits January 22, 2018 12:22

refactor member description assignment to support extensibility

9fd47ee

refactor member description assignment to support extensibility

6db76dd

fix checkstyle

624fffa

improve documentation

3d51d69

fix bug: self assignment variable. typo

f0bcca5

jeqo added 2 commits January 30, 2018 13:15

change scope of groups to consumergroups

ee9c074

fix removed method

1baa9e5

hachikuji reviewed Jan 30, 2018

View reviewed changes

jeqo added 2 commits February 5, 2018 22:48

rename protocol to partitionassignor

6558697

lookup for coordinator before describe consumer group

9d135cd

jeqo added 2 commits February 6, 2018 23:42

Merge remote-tracking branch 'origin/trunk' into feature/admin-client…

82d3fab

…-describe-consumer-group

lookup for coordinator before list consumer group and offset

98782cb

hachikuji reviewed Feb 22, 2018

View reviewed changes

hachikuji and others added 3 commits April 9, 2018 15:39

Make [Config]Resource.toString() consistent with existing code (apach…

e490a90

…e#4845) The toString() for ConfigResource was using { } instead of ( ) which is inconsistent with the existing toStrings in the code, while toString for Resource was using a mix of ( and }.

guozhangwang reviewed Apr 10, 2018

View reviewed changes

rajinisivaram and others added 8 commits April 10, 2018 20:42

KAFKA-4883: handle NullPointerException while parsing login modue con…

5e277e5

…trol flag (apache#4849)

MINOR: Add NullPayloadGenerator to Trogdor (apache#4844)

4223ef6

fix test with updated method firm

cb36215

remove redundant access methods

9fb5f82

remove redundant access methods

f410834

remove unused imports

055c557

ignoring test to be fixed on follow-up PR

175ff6d

omkreddy added 2 commits April 11, 2018 10:30

KAFKA-6752: Enable unclean leader election metric (apache#4838)

e29fa9a

Reviewers: Jun Rao <junrao@gmail.com>

KAFKA-6447: Add Delegation Token Operations to KafkaAdminClient (KIP-…

47918f2

…249) (apache#4427) Reviewers: Jun Rao <junrao@gmail.com>

rebase from trunk

1d832b3

guozhangwang closed this in 6a99da8 Apr 11, 2018

jeqo deleted the feature/admin-client-describe-consumer-group branch August 8, 2020 09:20

gharris1727 mentioned this pull request Feb 14, 2024

MINOR: Make Checkstyle more strict, restore global code quality checks to 2018 #15367

Closed

3 tasks

Conversation

jeqo commented Jan 21, 2018

Uh oh!

hachikuji commented Jan 25, 2018

Uh oh!

hachikuji commented Jan 26, 2018

Uh oh!

jeqo commented Jan 28, 2018

Uh oh!

hachikuji commented Jan 29, 2018

Uh oh!

jeqo commented Jan 30, 2018

Uh oh!

hachikuji 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!

asfgit commented Feb 6, 2018

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

guozhangwang commented Apr 9, 2018

Uh oh!

guozhangwang 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!

guozhangwang commented Apr 11, 2018

Uh oh!

guozhangwang commented Apr 11, 2018

Uh oh!

guozhangwang commented Apr 11, 2018

Uh oh!

guozhangwang commented Apr 11, 2018

Uh oh!

jeqo commented Apr 11, 2018

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

17 participants