Table of Contents

Status

Current state: Under discussionAccepted

Discussion thread: here

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

...

In situations where command-line tools are used to administer a group of the wrong type, the error message now indicates when the group type is wrong, rather than saying the group does not exist. This is achieved using a new error code message in the Kafka protocol and a slight change in behavior of the admin client.

...

Describing groups using the admin client

A new exception InconsistentGroupTypeExceptionThe 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 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.

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.
    *
    * @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,

Here are the method signatures:

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>
    * <p>ThisThis is a convenience method for {@link #listGroups(ListGroupsOptions#describeClassicGroups(Collection, DescribeClassicGroupsOptions)}
    * with default options.
    * See the overload for more details.
    *
    * @param @returngroupIds The ListGroupsResult.
IDs of the groups to describe.
    * @return The DescribeClassicGroupsResult.
    */
   default ListGroupsResultDescribeClassicGroupsResult listGroupsdescribeClassicGroups(Collection<String> groupIds) {
       return listGroupsdescribeClassicGroups(groupIds, new ListGroupsOptionsDescribeClassicGroupsOptions());
   }
 
    /**
    * List the groups available in the cluster with the default options.
    *
    * @param<p>This is optionsa Theconvenience optionsmethod tofor use when listing the groups{@link #listGroups(ListGroupsOptions)} with default options.
    * @return The ListGroupsResultSee the overload for more details.
    */
   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

* @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

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 DescribeClassicGroupsResultDescribeClassicGroupsOptions extends AbstractOptions<DescribeClassicGroupsOptions> {

    publicprivate DescribeClassicGroupsResult(final Map<String, KafkaFuture<ClassicGroupDescription>> futures)boolean includeAuthorizedOperations;

    /**
public DescribeClassicGroupsOptions includeAuthorizedOperations(boolean includeAuthorizedOperations) {
 * Return a map from group id tothis.includeAuthorizedOperations futures which yield group descriptions.= includeAuthorizedOperations;
     */
    public Map<String, KafkaFuture<ClassicGroupDescription>> describedGroups()return this;

    /**}

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

ClassicGroupDescription

}
}

DescribeClassicGroupsResult

Code Block

package org

Code Block

package org.apache.kafka.client.admin;

/**
 * A detailed description of a single classic group in the cluster 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 ClassicGroupDescriptionDescribeClassicGroupsResult {

    public ClassicGroupDescriptionDescribeClassicGroupsResult(Stringfinal groupIdMap<String,
 KafkaFuture<ClassicGroupDescription>>  futures);

    /**
     * Return a map from group id to futures which yield group descriptions.
     */
    public Map<String, String protocol,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  Collection<MemberDescription> members,
                                   String partitionAssignor,
                                   ClassicGroupStateString stateprotocol,
                                   Node coordinator);

    public ClassicGroupDescription(String groupIdCollection<MemberDescription> members,
                                   String protocolpartitionAssignor,
                                   Collection<MemberDescription>GroupState membersgroupState,
                                   String partitionAssignor,Node coordinator);

      public ClassicGroupDescription(String groupId,
                                 ClassicGroupState  String stateprotocol,
                                   NodeCollection<MemberDescription> coordinatormembers,
                                   Set<AclOperation> authorizedOperations);

String partitionAssignor,
    /**
     * The id of the classic group.
     */
    public String groupId();

    /**
     *GroupState ThegroupState,
 group protocol type.
     */
    public String protocol();

    /**
     * If the group is a simple consumer group or not.
  Node coordinator,
  */
    public boolean isSimpleConsumerGroup();

    /**
     * A list of the members of the classic group.
     */
    public Collection<MemberDescription>Set<AclOperation> members(authorizedOperations);

    /**
     * The group partition assignorid of the classic group.
     */
    public String partitionAssignorgroupId();

    /**
     * The classic group state, or UNKNOWN if the state is too new for us to parseprotocol type.
     */
    public ClassicGroupStateString stateprotocol();

    /**
     * TheIf the group coordinator,is ora nullsimple ifconsumer thegroup coordinatoror is not known.
     */
    public Nodeboolean coordinatorisSimpleConsumerGroup();

    /**
     * authorizedOperationsA forlist thisof group,the ormembers nullof ifthe that information is not knownclassic group.
     */
    public Set<AclOperation>Collection<MemberDescription> authorizedOperationsmembers();
}

ClassicGroupState

Code Block

package org.apache.kafka.common;


    /**
     * The classicgroup grouppartition stateassignor.
     */
public enum ClassicGroupState {
 public String  UNKNOWNpartitionAssignor("Unknown"),;

    PREPARING_REBALANCE("PreparingRebalance"),/**
    COMPLETING_REBALANCE("CompletingRebalance"),
    STABLE("Stable"),
    DEAD("Dead"),
    EMPTY("Empty");

    ClassicGroupState(String name);

    /** * The group state, or UNKNOWN if the state is too new for us to parse.
     */
    public * Case-insensitive classic group state lookup by string nameGroupState groupState();

    /**
     * The group coordinator, or null if the coordinator is not known.
     */
    public staticNode ClassicGroupState parse(String namecoordinator();

    /**
 public  String toString();
}

ListGroupsOptions

 * 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;
 
/**
 * 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> {

    /**
     * Only consumer groups will be returned by listGroups().
     * This operation sets filters on group type and protocol type which select consumer groups

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>static ListGroupsOptions typesforConsumerGroups() {
        return types; new 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 extends AbstractListGroupsResult<GroupListing> {
    ListGroupsResult(KafkaFuture<Collection<Object>> future) {        .withTypes(Set.of(GroupType.CLASSIC, GroupType.CONSUMER))
            .withProtocolTypes(Set.of("", ConsumerProtocol.PROTOCOL_TYPE));
    }

    /**
     * Only share groups will be returned by listGroups().
     * This operation sets a filter on group type which select share groups.
     */
    public superstatic ListGroupsOptions forShareGroups(future); {
     }

   return  /**new ListGroupsOptions()
     * Returns a future that yields either an exception, or the full set of group listings. .withTypes(Set.of(GroupType.SHARE));
    }

    /**
     */
    public KafkaFuture<Collection<GroupListing>> all() {
    }
  
    /**
     * Returns a future which yields just the valid listings Only streams groups will be returned by listGroups().
     * This operation sets a filter on group type which select streams groups.
     */
    public static KafkaFuture<Collection<GroupListing>>ListGroupsOptions validforStreamsGroups() {
    }
    return new ListGroupsOptions()
            .withTypes(Set.of(GroupType.STREAMS));
    }

    /**
     * Returns a future which yields just the errors which occurredIf groupStates is set, only groups in these states will be returned by listGroups().
     */
 Otherwise, all groups publicare KafkaFuture<Collection<Throwable>> errors() {
returned.
      }
}

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

    /**
     * TheIf groupprotocol type.
types is  */
  public Optional<GroupType> type();
  
  /**
   * The group protocol type.
set, only groups of these protocol types will be returned by listGroups().
     * Otherwise, all groups are returned.
     */
    public StringListGroupsOptions protocolwithProtocolTypes();
Set<String> protocolTypes) {
  /**
   * If the groupthis.protocolTypes is= a(protocolTypes simple== consumernull group or not.
   */
  public boolean isSimpleConsumerGroup();
}

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 .

Exceptions

The following new exception is added to the org.apache.kafka.common.errors package corresponding to the new error code in the Kafka protocol.

InconsistentGroupTypeException - Indicates that the group exists but the group type is inconsistent with the operation.

Kafka protocol changes

This KIP adds the following error code to the Kafka protocol.

INCONSISTENT_GROUP_TYPE (value TBD) - Indicates that the group exists but the group type is inconsistent with the operation.

This error code is used when the following RPCs are used with an existing group of the wrong type:

ConsumerGroupDescribe
ShareGroupDescribe

These RPCs are used by administrative tools and the new error code will help with the usability of the tools. A new version of ConsumerGroupDescribe with no schema change will be required to support the new error code. Because ShareGroupDescribe is still unstable, this error code can be added to the v0 RPC.

...

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

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.common;

/**
 * The group state.
 */
public enum GroupState {
    UNKNOWN("Unknown"),
    PREPARING_REBALANCE("PreparingRebalance"),
    COMPLETING_REBALANCE("CompletingRebalance"),
    STABLE("Stable"),
    DEAD("Dead"),
    EMPTY("Empty"),
    ASSIGNING("Assigning"),
    RECONCILING("Reconciling");

    GroupState(String name);

    /**
     * Case-insensitive group state lookup by string name.
     */
    public static GroupState parse(String name);

    public String toString();
}

Kafka protocol changes

This KIP introduces a new version for the DescribeGroups API.

DescribeGroups API

The KIP introduces version 6. This changes the error behavior so that if a classic group is described and the group ID either refers to a group of a different type, or the group ID is not found, the error code GROUP_ID_NOT_FOUND is returned. Previously, the error code NONE was used with a group state of DEAD .

Request schema

Version 6 is the same as version 5.

Code Block

{
  "apiKey": 15,
  "type": "request",
  "listeners": ["zkBroker", "broker"],
  "name": "DescribeGroupsRequest",
  // Versions 1 and 2 are the same as version 0.
  //
  // Starting in version 3, authorized operations can be requested.
  //
  // Starting in version 4, the response will include group.instance.id info for members.
  //
  // Version 5 is the first flexible version.
  //
  // Version 6 returns error code GROUP_ID_NOT_FOUND if the group ID is not found (KIP-1043).
  "validVersions": "0-6",
  "flexibleVersions": "6+",
  "fields": [
    { "name": "Groups", "type": "[]string", "versions": "0+", "entityType": "groupId",
      "about": "The names of the groups to describe" },
    { "name": "IncludeAuthorizedOperations", "type": "bool", "versions": "3+",
      "about": "Whether to include authorized operations." }
  ]
}

Response schema

Version 6 adds the ErrorMessage to the DescribedGroup to enable the reasons for a non-existent group to be understood more clearly.

Code Block

{
  "apiKey": 15,
  "type": "response",
  "name": "DescribeGroupsResponse",
  // Version 1 added throttle time.
  //
  // Starting in version 2, on quota violation, brokers send out responses before throttling.
  //
  // Starting in version 3, brokers can send authorized operations.
  //
  // Starting in version 4, the response will optionally include group.instance.id info for members.
  //
  // Version 5 is the first flexible version.
  //
  // Version 6 returns error code GROUP_ID_NOT_FOUND if the group ID is not found (KIP-1043).
  "validVersions": "0-6",
  "flexibleVersions": "6+",
  "fields": [
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "1+", "ignorable": true,
      "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
    { "name": "Groups", "type": "[]DescribedGroup", "versions": "0+",
      "about": "Each described group.", "fields": [
      { "name": "ErrorCode", "type": "int16", "versions": "0+",
        "about": "The describe error, or 0 if there was no error." },
      { "name": "ErrorMessage", "type": "string", "versions": "6+", "nullableVersions": "6+", "default": "null",
        "about": "The describe error message, or null if there was no error." },
      { "name": "GroupId", "type": "string", "versions": "0+", "entityType": "groupId",
        "about": "The group ID string." },
      { "name": "GroupState", "type": "string", "versions": "0+",
        "about": "The group state string, or the empty string." },
      { "name": "ProtocolType", "type": "string", "versions": "0+",
        "about": "The group protocol type, or the empty string." },
      // ProtocolData is currently only filled in if the group state is in the Stable state.
      { "name": "ProtocolData", "type": "string", "versions": "0+",
        "about": "The group protocol data, or the empty string." },
      // N.B. If the group is in the Dead state, the members array will always be empty.
      { "name": "Members", "type": "[]DescribedGroupMember", "versions": "0+",
        "about": "The group members.", "fields": [
        { "name": "MemberId", "type": "string", "versions": "0+",
          "about": "The member ID assigned by the group coordinator." },
        { "name": "GroupInstanceId", "type": "string", "versions": "4+", "ignorable": true,
          "nullableVersions": "4+", "default": "null",
          "about": "The unique identifier of the consumer instance provided by end user." },
        { "name": "ClientId", "type": "string", "versions": "0+",
          "about": "The client ID used in the member's latest join group request." },
        { "name": "ClientHost", "type": "string", "versions": "0+",
          "about": "The client host." },
        // This is currently only provided if the group is in the Stable state.
        { "name": "MemberMetadata", "type": "bytes", "versions": "0+",
          "about": "The metadata corresponding to the current group protocol in use." },
        // This is currently only provided if the group is in the Stable state.
        { "name": "MemberAssignment", "type": "bytes", "versions": "0+",
          "about": "The current assignment provided by the group leader." }
      ]},
      { "name": "AuthorizedOperations", "type": "int32", "versions": "3+",  "default": "-2147483648",
        "about": "32-bit bitfield to represent authorized operations for this group." }
    ]}
  ]
}

Command-line tools

kafka-groups.sh

...

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

ListGroups v5 introduces the TypesFilter in the request and the GroupType in the response. This KIP does not need to introduce a new version of the ListGroups RPC, because:

...

After this KIP, if you use Admin.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 INCONSISTENTGROUP_ID_GROUPNOT_TYPEFOUND . In the admin client, this means the future for this group completes exceptionally with InconsistentGroupTypeException GroupIdNotFoundException rather than succeeding with a dead consumer group.

...

It would be possible to preserve the current behavior of Admin.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.unnecessary complication with little benefit.

It was proposed to add a The error code INCONSISTENT_GROUP_PROTOCOL already exists and is used with the classic consumer group protocol if a member attempts to join the group with an inconsistent sub-protocol. You would see this if you started up a distributed Kafka connect worker with the default configuration and then tried to use kafka-console-consumer.sh with the group ID connect-cluster . Rather than overloading this error code, this KIP introduces the new error code INCONSISTENT_GROUP_TYPE which is similar but applies to to indicate that the group existed but the group type (consumer, share, classic) rather than the sub-protocol within the classic consumer group protocolwas inconsistent with the operation. Instead, the existing error code GROUP_ID_NOT_FOUND is used and the protocol is enhanced so that an error message is returned along with this error code.

Space shortcuts

Child pages

Versions Compared

Old Version 13

New Version Current

Key

Status

Describing groups using the admin client

Public Interfaces

Client API changes

Admin

Unified group state

Public Interfaces

Client API changes

Admin

DescribeClassicGroupsOptions

DescribeClassicGroupsResult

DescribeClassicGroupsOptions

ClassicGroupDescription

DescribeClassicGroupsResult

ClassicGroupDescription

ClassicGroupState

ListGroupsOptions

ConsumerGroupDescription

ListGroupsOptions

ListGroupsResult

GroupListing

ConsumerGroupListing

ShareGroupListing

Exceptions

Kafka protocol changes

ListGroupsResult

GroupListing

Kafka protocol changes

DescribeGroups API

Request schema

Response schema

Command-line tools

kafka-groups.sh

Space shortcuts

Child pages

Page History

Versions Compared

Old Version 13

New Version Current

Key

Status

Describing groups using the admin client

Public Interfaces

Client API changes

Admin

Unified group state

Public Interfaces

Client API changes

Admin

DescribeClassicGroupsOptions

DescribeClassicGroupsResult

DescribeClassicGroupsOptions

ClassicGroupDescription

DescribeClassicGroupsResult

ClassicGroupDescription

ClassicGroupState

ListGroupsOptions

ConsumerGroupDescription

ListGroupsOptions

ListGroupsResult

GroupListing

ConsumerGroupListing

ShareGroupListing

Exceptions

Kafka protocol changes

ListGroupsResult

GroupListing

Kafka protocol changes

DescribeGroups API

Request schema

Response schema

Command-line tools

kafka-groups.sh