DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
The behavior of groups in Apache Kafka is more complicated and subtle than it first appears. To most users of Kafka, groups are synonymous with consumer groups. However, the "classic" consumer group protocol was extensible and there are several well-known extensions in use. For example, distributed workers in Kafka Connect also use groups as a coordination mechanism, and some applications such as schema registries have also built upon the consumer group protocol in interesting ways. These are all groups. Then, KIP-848 introduced the new consumer group protocol and modern consumer groups use this new protocol. KIP-932 introduces share groups, KIP-1071 (not yet adopted) introduces streams groups, and there may well be additional types of group in the future.
...
Add the following methods on the org.apache.kafka.client.admin.AdminClient interface.
| Method signature | Description |
|---|---|
ListGroupsResult listGroups() | List the groups available in the cluster. |
ListGroupsResult listGroups(ListGroupsOptions options) | List the groups available in the cluster. |
Here are the method signatures:
...
| Code Block |
|---|
package org.apache.kafka.client.admin;
import org.apache.kafka.common.GroupType;
/**
* Options for {@link Admin#listGroups(ListGroupsOptions)}.
*
* The API of this class is evolving, see {@link Admin} for details.
*/
@InterfaceStability.Evolving
public class ListGroupsOptions extends AbstractOptions<ListGroupsOptions> {
} |
ListGroupsResult
| Code Block |
|---|
package org.apache.kafka.clients.admin; /** * The result of the* {@link Admin#listGroups(ListGroupsOptions)} call. * <p> * The API of this class is evolving, see {@link Admin} for details. */ @InterfaceStability.Evolving public class ListGroupsResult {If types is set, only groups of these types will be returned by listGroups(). * Otherwise, all groups are returned. */** public *ListGroupsOptions Returns a future that yields either an exception, or the full set of group listings. */ public KafkaFuture<Collection<GroupListing>> all() { } withTypes(Set<GroupType> types) { this.types = (types == null || types.isEmpty()) ? Collections.emptySet() : new HashSet<>(types); return this; } /** * Returns a future which yields just the valid listings the list of group types that are requested or empty if no types have been specified. */ public KafkaFuture<Collection<GroupListing>>Set<GroupType> validtypes() { } return types; } } |
ListGroupsResult
| Code Block |
|---|
package org.apache.kafka.clients.admin;/** /** * ReturnsThe a future which yields justresult of the errors which occurred{@link Admin#listGroups(ListGroupsOptions)} call. * <p> */ The API of publicthis KafkaFuture<Collection<Throwable>>class errors() { } } |
GroupListing
| Code Block |
|---|
package org.apache.kafka.client.admin; import org.apache.kafka.common.ShareGroupState; is evolving, see {@link Admin} for details. */ @InterfaceStability.Evolving public class ListGroupsResult { /** * A listing of a* groupReturns ina thefuture cluster. that *yields <p> either *an Theexception, APIor ofthe thisfull classset isof evolving, see {@link Admin} for details. group listings. */ @InterfaceStability.Evolving public class GroupListing { public KafkaFuture<Collection<GroupListing>> GroupListingall(String) groupId,{ GroupType type, String protocolType);} /** * The id of the groupReturns a future which yields just the valid listings. */ public StringKafkaFuture<Collection<GroupListing>> groupIdvalid(); { /**} * The group type. */ /** public GroupType type(); * Returns a /** future which yields *just the Theerrors protocolwhich typeoccurred. */ public StringKafkaFuture<Collection<Throwable>> protocolTypeerrors(); { } } |
Command-line tools
kafka-groups.sh
A new tool called kafka-groups.sh is introduced for listing and describing groups of any kind. It has the following options:
...
--bootstrap-server <String: server to connect to>
...
REQUIRED: The server(s) to connect to.
...
--command-config <String: command config property file>
...
Property file containing configs to be passed to Admin Client.
...
--describe
...
Describe the details of the groups.
...
--group-type <String: type>
...
Filters the groups based on group type. Valid types are: 'consumer' (consumer groups) and 'share' (share groups).
...
--help
...
Print usage information.
...
--list
...
List all groups.
...
--version
...
Display Kafka version.
GroupListing
| Code Block |
|---|
package org.apache.kafka.client.admin;
import org.apache.kafka.common.ShareGroupState;
/**
* A listing of a group in the cluster.
* <p>
* The API of this class is evolving, see {@link Admin} for details.
*/
@InterfaceStability.Evolving
public class GroupListing {
public GroupListing(String groupId, String protocol);
public GroupListing(String groupId, GroupType type, String protocol);
public GroupListing(String groupId, Optional<GroupType> type, String protocol);
/**
* The id of the group.
*/
public String groupId();
/**
* The group type.
*/
public Optional<GroupType> type();
/**
* The group protocol type.
*/
public String protocol();
} |
ConsumerGroupListing
This class is modified to extend org.apache.kafka.clients.admin.GroupListing .
ShareGroupListing
This class is modified to extend org.apache.kafka.clients.admin.GroupListing .
Command-line tools
kafka-groups.sh
A new tool called kafka-groups.sh is introduced for listing and describing groups of any kind. It has the following options:
| Option | Description |
|---|---|
--bootstrap-server <String: server to connect to> | REQUIRED: The server(s) to connect to. |
--command-config <String: command config property file> | Property file containing configs to be passed to Admin Client. |
--consumer | Filters the groups based on group type and protocol in order to show consumer groups. |
--describe | Describe the details of the groups. |
--group-type <String: type> | Filters the groups based on group type. Valid types are: 'consumer' (consumer groups) and 'share' (share groups). |
--help | Print usage information. |
--list | List all groups. |
--protocol <String: protocol> | Filters the groups based on protocol type. |
--version | Display Kafka version. |
Note that --consumer actually matches all groups whose type is Consumer , and groups whose type is Classic and protocol type is "consumer". The filtering is done in the kafka-groups.sh tool.
Here are some examples.
To list all of the groups:
| Code Block |
|---|
$ bin/kafka-groups.sh --bootstrap-server localhost:9092 --list
old-consumer-group
new-consumer-group
connect-cluster
share-group |
To describe all of the groups and their types:
| Code Block |
|---|
$ bin/kafka-groups.sh --bootstrap-server localhost:9092 --describe
GROUP TYPE PROTOCOL
old-consumer-group Classic consumer
new-consumer-group Consumer consumer
connect-cluster Classic connect
share-group Share share |
To list all of the consumer groups:
| Code Block |
|---|
$ bin/kafka-groups.sh --bootstrap-server localhost:9092 --list --consumer
old-consumer-group
new-consumer-group |
To describe all of the consumer groups:
| Code Block |
|---|
$ bin/kafka-groups.sh --bootstrap-server localhost:9092 --describe --consumer
GROUP TYPE PROTOCOL
old-consumer-group Classic consumer
new-consumer-group Consumer consumer |
To describe all of the share groups:
| Code Block |
|---|
$ bin/kafka-groups.sh --bootstrap-server localhost:9092 --describe --group-type share
GROUP TYPE PROTOCOL
share-group Share share |
kafka-consumer-groups.sh
For all operations which act on a single group, if that group exists but is not a consumer group, the command fails with a message indicating that the group type is incorrect, rather than the existing message that the group does not exist.
For example, if you try to describe a share group, the output will look like this:
| Code Block |
|---|
$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group SG1
Error: Group 'SG1' is not a consumer group. |
kafka-share-groups.sh
A new option --create is added to this tool to create a share group. If the share group exists, the command succeeds. If the group does not exist, the share group is created. If the group exists but it's not a share group, the command fails.
Note that --group-type consumer actually matches all groups whose protocol is "consumer" , and --group-type share matches all groups whose type is Share. The filtering is done in the kafka-groups.sh tool.
Here are some examples.
To list all of the groups:
| Code Block |
|---|
$ bin/kafka-groups.sh --bootstrap-server localhost:9092 --list
old-consumer-group
new-consumer-group
connect-cluster
share-group |
To describe all of the groups and their types:
| Code Block |
|---|
$ bin/kafka-groups.sh --bootstrap-server localhost:9092 --describe
GROUP TYPE PROTOCOL
old-consumer-group Classic consumer
new-consumer-group Consumer consumer
connect-cluster Classic connect
share-group Share share |
To list all of the consumer groups:
| Code Block |
|---|
$ bin/kafka-groups.sh --bootstrap-server localhost:9092 --list --group-type consumer
old-consumer-group
new-consumer-group |
To describe all of the consumer groups:
| Code Block |
|---|
$ bin/kafka-groups.sh --bootstrap-server localhost:9092 --describe --group-type consumer
GROUP TYPE PROTOCOL
old-consumer-group Classic consumer
new-consumer-group Consumer consumer |
kafka-consumer-groups.sh
For all operations which act on a single group, if that group exists but is not a consumer group, the command fails with a message indicating that the group type is incorrect, rather than the existing message that the group does not exist.
For example, if you try to describe a share group, the output will look like this:
| Code Block |
|---|
$ bin/kafka-consumershare-groups.sh --bootstrap-server localhost:9092 --describecreate --group SG1NewShareGroup Error:Share Groupgroup 'SG1NewShareGroup' is not a consumer group.exists. $ bin/kafka-share-groups.sh |
...
A new option --create is added to this tool to create a share group. If the share group exists, the command succeeds. If the group does not exist, the share group is created. If the group exists but it's not a share group, the command fails.
| Code Block |
|---|
--bootstrap-server localhost:9092 --create --group ExistingShareGroup Share group 'ExistingShareGroup' exists. $ bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --create --group NewShareGroup Share group 'NewShareGroup' exists. $ bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --create --group ExistingShareGroup Share group 'ExistingShareGroup' exists. --create --group ConsumerGroup Error: Group 'ConsumerGroup' is not a share group. |
Under the covers, it uses the AlterShareGroupOffsets RPC with an empty Topics array.
Also, for all operations which act on a single group, if that group exists but is not a share group, the command fails with a message indicating that the group type is incorrect, rather than the existing messages that the group does not exist.
For example, if you try to describe a consumer group, the output will look like this:
| Code Block |
|---|
$ bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --createdescribe --group ConsumerGroupCG1 Error: Group 'ConsumerGroupCG1' is not a share group. |
Under the covers, it uses the AlterShareGroupOffsets RPC with an empty Topics array.
Also, for all operations which act on a single group, if that group exists but is not a share group, the command fails with a message indicating that the group type is incorrect, rather than the existing messages that the group does not exist.
For example, if you try to describe a consumer group, the output will look like this:
| Code Block |
|---|
$ bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --describe --group CG1
Error: Group 'CG1' is not a share group. |
Proposed Changes
This KIP introduces a command-line tool for displaying all of the groups and their types.
Next, it introduces a way to create a share group administratively. If you create Kafka resources such as topics as part of deploying an application, you can now create share groups in the same way.
Finally, in situations where command-line tools are used to administer a group of the wrong type, you’ll now be told the group type is wrong, rather than the group does not exist.
Listing groups
The KIP introduces a new tool called kafka-groups.sh to show all of the groups in a cluster, their types and the protocols they use. This lets you see consumer groups, share groups, Kafka Connect cluster groups, and any other custom groups all together. It doesn't replace the specific tools for the different types of group, but it does shed light on what's actually going on for administrators. Note that this does not require any changes to the Kafka protocol. The information is already available, but not directly accessible by the administrator.
The ListGroups RPC response returns three pieces of information for each group: group ID, type and protocol. For the common types of group, here is what they mean:
...
Type
...
Protocol
...
Meaning
...
Classic
...
"consumer"
...
Consumer group with the "classic" consumer group protocol
...
Consumer
...
"consumer"
...
Consumer group with the KIP-848 consumer group protocol
...
Share
...
"share"
...
Share group
...
Classic
...
"connect"
...
Kafka Connect distributed worker cluster group
...
Classic
...
Any other string
...
Other customization of "classic" consumer group protocol, such as a schema registry
Proposed Changes
This KIP introduces a command-line tool for displaying all of the groups and their types.
Next, it introduces a way to create a share group administratively. If you create Kafka resources such as topics as part of deploying an application, you can now create share groups in the same way.
Finally, in situations where command-line tools are used to administer a group of the wrong type, you’ll now be told the group type is wrong, rather than the group does not exist.
Listing groups
The KIP introduces a new tool called kafka-groups.sh to show all of the groups in a cluster, their types and the protocols they use. This lets you see consumer groups, share groups, Kafka Connect cluster groups, and any other custom groups all together. It doesn't replace the specific tools for the different types of group, but it does shed light on what's actually going on for administrators. Note that this does not require any changes to the Kafka protocol. The information is already available, but not directly accessible by the administrator.
The ListGroups RPC response returns three pieces of information for each group: group ID, type and protocol. For the common types of group, here is what they mean:
Type | Protocol | Meaning |
|---|---|---|
Classic |
| Consumer group with the "classic" consumer group protocol |
Consumer |
| Consumer group with the KIP-848 consumer group protocol |
Share |
| Share group |
Classic |
| Kafka Connect distributed worker cluster group |
Classic | Any other string | Other customization of "classic" consumer group protocol, such as a schema registry |
The new kafka-groups.sh tool makes all of this information available.
Creating groups
Groups are created dynamically on first use. For example, when you connect the first consumer in a consumer group, the group coordinator creates the group as a consumer group. This is convenient, but it does mean that you need to ensure that different types of group use distinct group IDs or the results will be unpredictable. This is because the group type depends upon how the group was created, whether it was a consumer group member, a distributed Kafka Connect cluster, or whatever.
Today, you can create a consumer group administratively before the first member joins by resetting the offsets, such as like this:
$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --reset-offsets --to-latest --group CG1 --topic T1 --execute
That’s a slightly contorted way to create a consumer group, but it does already exist and it is known. As a result, this KIP does not introduce a new way to create consumer groups.
For share groups, a new --create option is added to the kafka-share-groups.sh tool to create a share group. You can use --reset-offsets in a similar way as consumer groups, but if you want to create a share group without specifying any topics, use --create instead. Note that the starting offset of a share group comes from the group's group.share.auto.offset.reset configuration property as introduced in KIP-932. This defaults to "latest" .
$ bin/kafka-share
The new kafka-groups.sh tool makes all of this information available.
Creating groups
Groups are created dynamically on first use. For example, when you connect the first consumer in a consumer group, the group coordinator creates the group as a consumer group. This is convenient, but it does mean that you need to ensure that different types of group use distinct group IDs or the results will be unpredictable. This is because the group type depends upon how the group was created, whether it was a consumer group member, a distributed Kafka Connect cluster, or whatever.
Today, you can create a consumer group administratively before the first member joins by resetting the offsets, such as like this:
$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --reset-offsets -create -to-latest --group CG1 --topic T1 --execute
That’s a slightly contorted way to create a consumer group, but it does already exist and it is known. As a result, this KIP does not introduce a new way to create consumer groups.
group SG1
Compatibility, Deprecation, and Migration Plan
A change to existing behavior is the error messages issued by the kafka-consumer-groups.sh and For share groups, a new --create option is added to the kafka-share-groups.sh tool to create a share group. You can use --reset-offsets in a similar way as consumer groups, but if you want to create a share group without specifying any topics, use --create instead. Note that the starting offset of a share group comes from the group's group.share.auto.offset.reset configuration property as introduced in KIP-932. This defaults to "latest" .
$ bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --create --group SG1
Compatibility, Deprecation, and Migration Plan
...
tools when working with groups of the wrong type.
We also need to consider compatibility for the ListGroups RPC and the AdminClient.listGroups() method. ListGroups v5 introduces the TypesFilter in the request and the GroupType in the response. This KIP does not introduce a new version of the ListGroups RPC.
- If the broker does not support ListGroups v5, setting a types filter using
ListGroupOptions.withTypes()results inUnsupportedVersionExceptionwhen the ListGroups request is serialized. (This is KIP-932 behavior.) - If the broker receives a group type in
TypesFilterthat it does not support in a ListGroups v5 (or later) request, the filter is treated as unknown group type which thus matches no groups. (This is KIP-932 behavior.) - If the client receives a group type that it does not understand in a ListGroups v5 (or later) response, the group listing is
GroupType.UNKNOWN. This is because the group type in the ListGroups RPC response cannot be parsed into a value of theGroupTypeenumeration, and the default value ofUNKNOWNis used in these situations.
Test Plan
The feature will be thoroughly tested with unit and integration tests.
...