Versions Compared

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.
Comment: Describe classic groups, table from --list

...

A new exception InconsistentGroupTypeException  is used by AdminClientAdmin.describeConsumerGroups(Collection<String>) and AdminClientAdmin.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.AdminClientAdmin  interface.

Method signatureDescription
DescribeClassicGroupsResult describeClassicGroups(Collection<String> groupIds) Describe some classic groups 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.
ListGroupsResult listGroups(ListGroupsOptions options) List the groups available in the cluster.

Here are the method signatures:

Code Block
   /**
    * ListDescribe some theclassic groups available in the cluster with the default options.
    *
    * <p>This@param isgroupIds aThe convenienceIDs methodof forthe {@link #listGroups(ListGroupsOptions)} with default optionsgroups to describe.
    * @param Seeoptions the overloadThe foroptions moreto details.
use when describing the *groups.
    * @return The ListGroupsResultDescribeClassicGroupsResult.
    */
   defaultDescribeClassicGroupsResult ListGroupsResult listGroupsdescribeClassicGroups()Collection<String> {groupIds,
       return listGroups(new ListGroupsOptions());
   }
 
   /**
    * List the groups available in the cluster.
    *
    * @param options The options to use when listing the groups.
    * @return The ListGroupsResult.
    */
   ListGroupsResultDescribeClassicGroupsOptions listGroups(ListGroupsOptions options);

ListGroupsOptions

Code Block
package org.apache.kafka.client.admin;

 
import org.apache.kafka.common.GroupType;
 
/**
 * Options for {@link Admin#listGroups(ListGroupsOptions)}.
 *
 *Describe Thesome APIclassic ofgroups thisin classthe is evolvingcluster, seewith {@linkthe Admin} for detailsdefault options.
 */
@InterfaceStability.Evolving
public class ListGroupsOptions extends* AbstractOptions<ListGroupsOptions> {<p>

    /**
 This is a convenience *method Iffor types is set, only groups of these types will be returned by listGroups().
     * Otherwise, all groups are returned.
 {@link #describeClassicGroups(Collection, DescribeClassicGroupsOptions)}
    * with default options. See the overload for more details.
    *
    */
 @param groupIds The publicIDs of ListGroupsOptionsthe withTypes(Set<GroupType> types) {groups to describe.
    * @return The  thisDescribeClassicGroupsResult.types
 = (types == null || types.isEmpty()) ? Collections.emptySet() : new HashSet<>(types);
  */
   default DescribeClassicGroupsResult describeClassicGroups(Collection<String> groupIds) {
       return describeClassicGroups(groupIds,  return thisnew DescribeClassicGroupsOptions());
    }

     /**
     * ReturnsList the listgroups ofavailable groupin typesthe thatcluster arewith requestedthe ordefault emptyoptions.
 if no types have been specified. *
     */
 <p>This is a publicconvenience Set<GroupType>method types()for {
@link #listGroups(ListGroupsOptions)} with default options.
    return* types;
See the overload  }
}

AbstractListGroupsResult

Code Block
package org.apache.kafka.clients.admin;
for more details.
    
/**
 * This class * @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);

DescribeClassicGroupsOptions

Code Block
/**
 * Options for {@link Admin#describeClassicGroups(Collection, DescribeClassicGroupsOptions)}.
 * <p>
 * The API of this class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class DescribeClassicGroupsOptions extends AbstractOptions<DescribeClassicGroupsOptions> {
    private boolean includeAuthorizedOperations;

    public DescribeClassicGroupsOptions includeAuthorizedOperations(boolean includeAuthorizedOperations) {
        this.includeAuthorizedOperations = includeAuthorizedOperations;
        return this;
    }

    public boolean includeAuthorizedOperations() {
        return includeAuthorizedOperations;
    }
}

DescribeClassicGroupsResult

Code Block
package org.apache.kafka.client.admin;

/**
 * The result of the {@link KafkaAdminClient#describeClassicGroups(Collection, DescribeClassicGroupsOptions)}} call.
 *
 * The API of this class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class DescribeClassicGroupsResult {

    public DescribeClassicGroupsResult(final Map<String, KafkaFuture<ClassicGroupDescription>> futures);

    /**
     * Return a map from group id to futures which yield group descriptions.
     */
    public Map<String, KafkaFuture<ClassicGroupDescription>> describedGroups();

    /**
     * Return a future which yields all ClassicGroupDescription objects, if all the describes succeed.
     */
    public KafkaFuture<Map<String, ClassicGroupDescription>> all();
}

ClassicGroupDescription

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,
                                   ClassicGroupState state,
                                   Node coordinator);

    public ClassicGroupDescription(String groupId,
                                   String protocol,
                                   Collection<MemberDescription> members,
                                   String partitionAssignor,
                                   ClassicGroupState state,
                                   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 ClassicGroupState state();

    /**
     * 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();
}

ClassicGroupState

Code Block
package org.apache.kafka.common;

/**
 * The classic group state.
 */
public enum ClassicGroupState {
    UNKNOWN("Unknown"),
    PREPARING_REBALANCE("PreparingRebalance"),
    COMPLETING_REBALANCE("CompletingRebalance"),
    STABLE("Stable"),
    DEAD("Dead"),
    EMPTY("Empty");

    ClassicGroupState(String name);

    /**
     * Case-insensitive classic group state lookup by string name.
     */
    public static ClassicGroupState parse(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> {

    /**
     * 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 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 extends AbstractListGroupsResult<GroupListing> {
    ListGroupsResult(KafkaFuture<Collection<Object>> future) {
        super(future);
    }
implements the common APIs that are shared by results classes
 * for various AdminClient commands for listing groups.
 * <p>
 * The API of this class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class AbstractListGroupsResult<T extends GroupListing> {

    AbstractListGroupsResult(KafkaFuture<Collection<Object>> future);

    /**
     * Returns a future that yields either an exception, or the full set of group listings.
     */
    public KafkaFuture<Collection<T>> all() {
    }
  
    /**
     * Returns a future which yields just the valid a future that yields either an exception, or the full set of group listings.
     */
    public KafkaFuture<Collection<T>>KafkaFuture<Collection<GroupListing>> validall() {
    }
   
    /**
     * Returns a future which yields just the errorsvalid which occurredlistings.
     */
    public KafkaFuture<Collection<Throwable>> errors() {
    }
}

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 extends AbstractListGroupsResult<GroupListing> {
    ListGroupsResult(KafkaFuture<Collection<Object>> future) {
        super(future); */
    public KafkaFuture<Collection<GroupListing>> valid() {
    }
   
    /**
     * Returns a future which yields just the errors which occurred.
     */
    public KafkaFuture<Collection<Throwable>> errors() {
    }
}

ListConsumerGroupsResult

...

ListShareGroupsResult

This is changed to extend AbstractListGroupsResult<ShareGroupListing> .

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 protocolgroupId();
  
  /**
   * The id of the groupgroup type.
   */
  public StringOptional<GroupType> groupIdtype();
  
  /**
   * The group protocol type.
   */
  public Optional<GroupType>String typeprotocol();
  
  /**
   * TheIf the group is a simple consumer group protocolor typenot.
   */
  public Stringboolean protocolisSimpleConsumerGroup();
}

ConsumerGroupListing

...

A new tool called kafka-groups.sh  is introduced for listing and describing groups of any kind. It has the following options:

OptionDescription

--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 to show all kinds of consumer groups, including classic and simple consumer groups. This matches group type 'consumer', and group type 'classic' where the protocol type is 'consumer' or empty.

--describe

Describe the details of the groups.

--group-type <String: type>

Filters the groups based on group type. Valid types are: 'classic', 'consumer' and 'share'.

--help

Print usage information.

--list

List all groups.

--protocol <String: protocol>

Filters the groups based on protocol typetype.

--share

Filters the groups to show share groups.

--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", and also "simple" consumer groups whose type is Classic and protocol type is "" . 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
schema-registry
simple-consumer-group

. The filtering is done in the kafka-groups.sh  tool.

Here are some examples.

To list all of the groupsTo describe all of the groups and their types:

Code Block
$ bin/kafka-groups.sh --bootstrap-server localhost:9092 --describelist
GROUP                   TYPE          PROTOCOL
old-consumer-group      Classic       consumer
new-consumer-group      Consumer      consumer
connect-cluster         Classic       connect
share-group             Share         share
schema-registry         Classic       sr
simple-consumer-group   Classic

To list all of the consumer groups, silently merging together all of the different kinds of consumer group:

Code Block
$ bin/kafka-groups.sh --bootstrap-server localhost:9092 --list --consumer
old-consumer-group
new-consumer-group
 sr
simple-consumer-group   Classic

To describe list all of the consumer groups, silently merging together all of the different kinds of consumer group:

Code Block
$ bin/kafka-groups.sh --bootstrap-server localhost:9092 --describelist --consumer
GROUP                   TYPE          PROTOCOL
old-consumer-group      Classic       consumer
new-consumer-group      Consumer      consumer
simple-consumer-group   Classic

To describe list all of the KIP-848 consumer groups:

Code Block
$ bin/kafka-groups.sh --bootstrap-server localhost:9092 --describelist --group-type consumer
GROUP                   TYPE          PROTOCOL
new-consumer-group      Consumer      consumer

To describe list all of the share groups:

Code Block
$ bin/kafka-groups.sh --bootstrap-server localhost:9092 --describelist --group-type share
GROUP                   TYPE          PROTOCOL
share-group             Share         share

...

When writing this KIP, it seemed that perhaps the ListGroups RPC would need to be enhanced to create the AdminClientAdmin.listGroups()  method. This turned out not to be necessary.

...

Prior to this KIP, the behavior of AdminClientAdmin.describeConsumerGroups(Collection<String>) seems a little unusual when used with groups which are not consumer groups. You can describe a collection of group IDs, some of which might exist and others might not. Here's how the response is built:

...

After this KIP, if you use AdminClientAdmin.describeConsumerGroup(Collection<String>)  to describe a group which is not a consumer group, case (2) above will result in the group information error code set to INCONSISTENT_GROUP_TYPE . In the admin client, this means the future for this group completes exceptionally with InconsistentGroupTypeException  rather than succeeding with a dead consumer group.

...

It would be possible to preserve the current behavior of AdminClientAdmin.describeConsumerGroups(Collection<String>)  when used with a group which is not a consumer group and introduce an option to ask it to validate the group type rather than converting any indescribable group into a dead consumer group. This seems like an unnecessary complication with little benefit.

...