DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
The error message of the exception GroupIdNotFoundException is used by Admin.describeConsumerGroups(Collection<String>) and Admin.describeShareGroups(Collection<String>) to indicate that the group being described has the wrong type. This small change enables the error messages from the command line kafka-consumer-groups.sh and kafka-share-groups.sh to indicate when the group ID is found, but it has the wrong type.
Public Interfaces
Client API changes
Admin
Add the following methods on the org.apache.kafka.client.admin.Admin interface.
...
Unified group state
The Admin client interfaces to list groups provide the ability to filter by group state. By introducing a method for listing groups of all types, the question arises how to handle group state. Previously, there were separate enums for ConsumerGroupState and ShareGroupState . Actually, ConsumerGroupState contains the set of states for both classic and modern consumer groups, which overlap but are not quite the same. ShareGroupState is a subset of the states in ConsumerGroupState .
This KIP introduces a single enum GroupState which contains all of the states from existing enums. In practice, that means the states are exactly the same as those in ConsumerGroupState .
Then, ConsumerGroupState is deprecated in favour of GroupState , and ShareGroupState is immediately replaced by GroupState because the KIP-932 interfaces have not actually been released yet.
The methods which return ConsumerGroupState will be deprecated and replaced with methods called groupState() which return GroupState instead.
Public Interfaces
Client API changes
Admin
Add the following methods on the org.apache.kafka.client.admin.Admin interface.
| Method signature | Description |
|---|---|
DescribeClassicGroupsResult describeClassicGroups(Collection<String> groupIds) | Describe some classic groups |
It also deprecates the following methods because listGroups with a type filter is preferred for listing groups rather than adding separate methods for each group type over time. The equivalent share groups methods will be removed from KIP-932.
| Method signature | Description | ListConsumerGroupsResult listConsumerGroups() | List the consumer groups available
|---|---|
| in the cluster, with the default options. | |
DescribeClassicGroupsResult describeClassicGroups(Collection<String> groupIds, DescribeClassicGroupsOptions options) | Describe some classic groups in the cluster. |
ListGroupsResult listGroups() | List the |
| groups available in the cluster. |
Also deprecated are the related classes such as ListConsumerGroupsResult and ListConsumerGroupsOptions .
Here are the method signatures of the new methods:
ListGroupsResult listGroups(ListGroupsOptions options) | List the groups available in the cluster. |
It also deprecates the following methods because listGroups with a type filter is preferred for listing groups rather than adding separate methods for each group type over time. The equivalent share groups methods will be removed from KIP-932.
| Method signature | Description |
|---|---|
ListConsumerGroupsResult listConsumerGroups() | List the consumer groups available in the cluster, with the default options. |
ListConsumerGroupsResult listConsumerGroups(ListConsumerGroupsOptions options) | List the consumer groups in the cluster. |
Also deprecated are the related classes such as ListConsumerGroupsResult and ListConsumerGroupsOptions .
Here are the method signatures of the new methods:
| Code Block |
|---|
/**
* Describe some classic groups in the cluster.
*
* |
| Code Block |
/**
* Describe some classic groups in the cluster.
*
* @param groupIds The IDs of the groups to describe.
* @param options The options to use when describing the groups.
* @return The DescribeClassicGroupsResult.
*/
DescribeClassicGroupsResult describeClassicGroups(Collection<String> groupIds,
DescribeClassicGroupsOptions options);
/**
* Describe some classic groups in the cluster, with the default options.
* <p>
* This is a convenience method for {@link #describeClassicGroups(Collection, DescribeClassicGroupsOptions)}
* with default options. See the overload for more details.
*
* @param groupIds The IDs of the groups to describe.
* @return The DescribeClassicGroupsResult.
*/
default DescribeClassicGroupsResult describeClassicGroups(Collection<String> groupIds) {
return describeClassicGroups(groupIds, new DescribeClassicGroupsOptions());
}
/**
* List the groups available in the cluster with the default options.
*
* <p>This is a convenience method for {@link #listGroups(ListGroupsOptions)} with default options.
* See the overload for more details.
*
* @return The ListGroupsResult.
*/
default ListGroupsResult listGroups() {
return listGroups(new ListGroupsOptions());
}
/**
* List the groups available in the cluster.
*
* @param options The options to use when listing the groups.
* @return The ListGroupsResult.
*/
ListGroupsResult listGroups(ListGroupsOptions options); |
...
| Code Block |
|---|
package org.apache.kafka.client.admin;
/**
* A detailed description of a single classic group in the cluster.
*/
@InterfaceStability.Evolving
public class ClassicGroupDescription {
public ClassicGroupDescription(String groupId,
String protocol,
Collection<MemberDescription> members,
String partitionAssignor,
ClassicGroupStateGroupState stategroupState,
Node coordinator);
public ClassicGroupDescription(String groupId,
String protocol,
Collection<MemberDescription> members,
String partitionAssignor,
ClassicGroupStateGroupState stategroupState,
Node coordinator,
Set<AclOperation> authorizedOperations);
/**
* The id of the classic group.
*/
public String groupId();
/**
* The group protocol type.
*/
public String protocol();
/**
* If the group is a simple consumer group or not.
*/
public boolean isSimpleConsumerGroup();
/**
* A list of the members of the classic group.
*/
public Collection<MemberDescription> members();
/**
* The group partition assignor.
*/
public String partitionAssignor();
/**
* The classic group state, or UNKNOWN if the state is too new for us to parse.
*/
public ClassicGroupStateGroupState stategroupState();
/**
* The group coordinator, or null if the coordinator is not known.
*/
public Node coordinator();
/**
* authorizedOperations for this group, or null if that information is not known.
*/
public Set<AclOperation> authorizedOperations();
} |
...
ConsumerGroupDescription
The existing constructors are deprecated and equivalents which use GroupState instead of ConsumerGroupState are added.
The method ConsumerGroupState state() is deprecated and GroupState groupState() is added.
ListGroupsOptions
| Code Block |
|---|
package org.apache.kafka.client.admin; import org.apache.kafka.common.GroupType; /** * TheOptions classicfor group state{@link Admin#listGroups(ListGroupsOptions)}. */ public * enumThe ClassicGroupStateAPI { of this class UNKNOWN("Unknown"), PREPARING_REBALANCE("PreparingRebalance"), COMPLETING_REBALANCE("CompletingRebalance"),is evolving, see {@link Admin} for details. */ @InterfaceStability.Evolving public class ListGroupsOptions extends AbstractOptions<ListGroupsOptions> { STABLE("Stable"), /** DEAD("Dead"), EMPTY("Empty"); ClassicGroupState(String name); * Only consumer groups will be returned by listGroups(). /** This operation sets filters * Case-insensitive classic group state lookup by string nameon group type and protocol type which select consumer groups. */ public static ClassicGroupStateListGroupsOptions parseforConsumerGroups(String name); ) { public String toString(); } |
ListGroupsOptions
| 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> { return new ListGroupsOptions() .withTypes(Set.of(GroupType.CLASSIC, GroupType.CONSUMER)) .withProtocolTypes(Set.of("", ConsumerProtocol.PROTOCOL_TYPE)); } /** * If types is set, only groups of these typesOnly share groups will be returned by listGroups(). * Otherwise,This alloperation groupssets area returned. filter on group type which select share groups. */ public static ListGroupsOptions withTypesforShareGroups(Set<GroupType> types) { this.typesreturn =new (types == null || types.isEmpty()) ? Collections.emptySet() : new HashSet<>(types); ListGroupsOptions() return this.withTypes(Set.of(GroupType.SHARE)); } /** * ReturnsOnly thestreams listgroups ofwill groupbe typesreturned that are requested or empty if no types have been specifiedby listGroups(). * This operation sets a filter on group type which select streams groups. */ public Set<GroupType>static ListGroupsOptions typesforStreamsGroups() { return types; new ListGroupsOptions() } } |
ListGroupsResult
| Code Block |
|---|
package org.apache.kafka.clients.adminwithTypes(Set.of(GroupType.STREAMS)); } /** * 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 { ListGroupsResult(KafkaFuture<Collection<Object>> future) { super(future); } /** * Returns a future that yields either an exception, or the full set of group listings. */ public KafkaFuture<Collection<GroupListing>> all() { } If groupStates is set, only groups in these states will be returned by listGroups(). * Otherwise, all groups are returned. * This operation is supported by brokers with version 2.6.0 or later. */ public ListGroupsOptions inGroupStates(Set<GroupState> groupStates) { this.groupStates = (groupStates == null || groupStates.isEmpty()) ? Collections.emptySet() : new HashSet<>(groupStates); return this; } /** * ReturnsIf aprotocol futuretypes whichis yieldsset, justonly thegroups validof listings. these protocol types will *be returned by listGroups(). * Otherwise, all groups are returned. */ public KafkaFuture<Collection<GroupListing>>ListGroupsOptions validwithProtocolTypes(Set<String> protocolTypes) { } this.protocolTypes = /** * Returns a future which yields just the errors which occurred.(protocolTypes == null || protocolTypes.isEmpty()) ? Set.of() : Set.copyOf(protocolTypes); return this; */ public KafkaFuture<Collection<Throwable>> errors() { } } |
...
}
/**
* If types is set, only groups of these types will be returned by listGroups().
* Otherwise, all groups are returned.
*/
public ListGroupsOptions withTypes(Set<GroupType> types) {
this.types = (types == null || types.isEmpty()) ? Collections.emptySet() : new HashSet<>(types);
return this;
}
/**
* Returns the list of group states that are requested or empty if not states have been specified.
*/
public Set<GroupState> groupStates() {
return groupStates;
}
/**
* Returns the list of protocol types that are requested or empty if no protocol types have been specified.
*/
public Set<String> protocolTypes() {
return protocolTypes;
}
/**
* Returns the list of group types that are requested or empty if no types have been specified.
*/
public Set<GroupType> types() {
return types;
}
} |
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 {
ListGroupsResult(KafkaFuture<Collection<Object>> future) {
super(future);
}
/**
* Returns a future that yields either an exception, or the full set of group listings.
*/
public KafkaFuture<Collection<GroupListing>> all() {
}
/**
* Returns a future which yields just the valid listings.
*/
public KafkaFuture<Collection<GroupListing>> valid() {
}
/**
* Returns a future which yields just the errors which occurred.
*/
public KafkaFuture<Collection<Throwable>> errors() {
}
} |
GroupListing
| Code Block |
|---|
package org.apache.kafka.client.admin;
/**
* 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, 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();
/**
* The group state.
*/
public Optional<GroupState> groupState();
/**
* If the group is a simple consumer group or not.
*/
public boolean isSimpleConsumerGroup();
} |
The new GroupState enum contains the states for all types of groups. The following table shows the correspondence between the group states and types. This table will be included in the javadoc.
| State | Classic group | Classic consumer group | Modern consumer group | Share group |
|---|---|---|---|---|
| UNKNOWN | Yes | Yes | Yes | Yes |
| PREPARING_REBALANCE | Yes | Yes | ||
| COMPLETING_REBALANCE | Yes | Yes | ||
| STABLE | Yes | Yes | Yes | Yes |
| DEAD | Yes | Yes | Yes | Yes |
| EMPTY | Yes | Yes | Yes | Yes |
| ASSIGNING | Yes | |||
| RECONCILING | Yes |
| Code Block |
|---|
package org.apache.kafka.client.admincommon; /** * A listing of a group in the cluster. * <p> * The API of this class is evolving, see {@link Admin} for details group state. */ @InterfaceStability.Evolving public classenum GroupListingGroupState { public GroupListing(String groupId, Optional<GroupType> type, String protocol); /** * The id of the group. */ public String groupId(); /** * The group type. */ public Optional<GroupType> type(); UNKNOWN("Unknown"), PREPARING_REBALANCE("PreparingRebalance"), COMPLETING_REBALANCE("CompletingRebalance"), STABLE("Stable"), DEAD("Dead"), EMPTY("Empty"), ASSIGNING("Assigning"), RECONCILING("Reconciling"); GroupState(String name); /** * TheCase-insensitive group protocolstate type. lookup by string */name. public String protocol(); */** * Ifpublic thestatic groupGroupState is a simple consumer group or not.parse(String name); */ public booleanString isSimpleConsumerGrouptoString(); } |
Kafka protocol changes
...