Versions Compared

Key

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

...

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.

List the consumer groups available ListConsumerGroupsResult listConsumerGroups(ListConsumerGroupsOptions consumer
Method signatureDescription
ListConsumerGroupsResult listConsumerGroups() 
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 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();
}

...

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.

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

...