Versions Compared

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.
Comment: Details of deprecation of listConsumerGroups

Table of Contents

Status

Current stateUnder discussionAccepted

Discussion thread: here

JIRA: https://issues.apache.org/jira/browse/KAFKA-16891

...

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 signatureDescription
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 signatureDescriptionListConsumerGroupsResult listConsumerGroups() List the consumer groups available
in the cluster, with the default options.
ListConsumerGroupsResult listConsumerGroups(ListConsumerGroupsOptions
DescribeClassicGroupsResult describeClassicGroups(Collection<String> groupIds, DescribeClassicGroupsOptions options) Describe some classic groups in the cluster.
ListGroupsResult listGroups() List the
consumer
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 signatureDescription
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();
}

ClassicGroupState

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 
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"),
    STABLE("Stable"),
    DEAD("Dead"),
    EMPTY("Empty");

    ClassicGroupState(String name);is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class ListGroupsOptions extends AbstractOptions<ListGroupsOptions> {

    /**
     * Only Case-insensitiveconsumer classicgroups groupwill statebe lookupreturned by string namelistGroups().
     */
 This operation sets filters publicon staticgroup ClassicGroupStatetype 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> {

    /**and protocol type which select consumer groups.
     */
    public static ListGroupsOptions forConsumerGroups() {
        return new ListGroupsOptions()
            .withTypes(Set.of(GroupType.CLASSIC, GroupType.CONSUMER))
     * If types is set, only groups .withProtocolTypes(Set.of these types will be returned by listGroups().("", ConsumerProtocol.PROTOCOL_TYPE));
    }

    /**
     * Otherwise,Only allshare groups will arebe returned by listGroups().
     */
 This operation sets publica ListGroupsOptionsfilter withTypes(Set<GroupType> types) {
    on group type which select share groups.
    this.types = (types == null || types.isEmpty()) ? Collections.emptySet() : new HashSet<>(types); */
    public static ListGroupsOptions forShareGroups() {
        return this;
    }

new ListGroupsOptions()
    /**
     * Returns the list of group types that are requested or empty if no types have been specified .withTypes(Set.of(GroupType.SHARE));
    }

    /**
     * Only streams groups will be returned by listGroups().
     * This operation sets a filter on group type which select streams groups.
     */
    public static Set<GroupType>ListGroupsOptions typesforStreamsGroups() {
        return types; new ListGroupsOptions()
    }
}

ListGroupsResult

Code Block
package org.apache.kafka.clients.admin;
   
        .withTypes(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 listingsIf 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 KafkaFuture<Collection<GroupListing>>ListGroupsOptions allinGroupStates(Set<GroupState> groupStates) {
       }
 this.groupStates = 
(groupStates == null  /**
     * Returns a future which yields just the valid listings.
     */
|| groupStates.isEmpty()) ? Collections.emptySet() : new HashSet<>(groupStates);
      public KafkaFuture<Collection<GroupListing>> valid() {return this;
      }
   
    /**
     * ReturnsIf aprotocol futuretypes whichis yieldsset, justonly thegroups errors which occurredof these protocol types will be returned by listGroups().
     * Otherwise, all groups are returned.
     */
    public KafkaFuture<Collection<Throwable>> errors() {
    }
}

...


    public ListGroupsOptions withProtocolTypes(Set<String> protocolTypes) {
        this.protocolTypes = (protocolTypes == null || protocolTypes.isEmpty()) ? Set.of() : Set.copyOf(protocolTypes);
        return this;
    }

     /**
     * 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.

StateClassic groupClassic consumer groupModern consumer groupShare group
UNKNOWNYesYesYesYes
PREPARING_REBALANCEYesYes

COMPLETING_REBALANCEYesYes

STABLEYesYesYesYes
DEADYesYesYesYes
EMPTYYesYesYesYes
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

...