...

The behavior of groups in Apache Kafka is more complicated and subtle than it first appears. To most users of Kafka, groups are synonymous with consumer groups. However, the "classic" consumer group protocol was extensible and there are several well-known extensions in use. For example, distributed workers in Kafka Connect also use groups as a coordination mechanism, and some applications such as schema registries have also built upon the consumer group protocol in interesting ways. These are all groups. Then,

KIP-848 introduced the new consumer group protocol and modern consumer groups use this new protocol. KIP-932 introduces share groups, KIP-1071 (not yet adopted) introduces streams groups, and there may well be additional types of group in the future.

All of these types of groups share a namespace for group IDs, but the manner in which you administer a group depends and the operations you can perform upon it depend upon its type.

Here’s an example of the complexity. If you start up a distributed Kafka Connect worker using the default configuration, it creates a group called "connect-cluster" . This is a group, but it’s not a consumer group. You can’t see this group in the list of consumer groups with the kafka-consumer-groups.sh  tool, but if you try to describe a consumer group called "connect-cluster"  or even use this group ID with a consumer, you get an error.

...

First, the admin client uses the ConsumerGroupDescribe RPC which responds with error code GROUP_ID_NOT_FOUND (69)  and an empty error message. Next, the admin client falls back to the pre-KIP-848 DescribeGroups RPC in case it's a classic consumer group.  This RPC succeeds and responds with error code NONE (0)  and returns the group with a status of Dead . It looks like a dead consumer groupconsumer group. There is no option of an error message in the protocol in this case, because the RPC doesn't support it. So, the Finally, this dead group is translated into the error message Error: Consumer group 'MYSHARE' does not exist . The output seems kind of acceptable, but the tool actually thinks it's dealing with a dead consumer group. It would be better if the ConsumerGroupDescribe RPC failed in a straightforward way.

This KIP tries to resolve some of these situations and make it easier to work out what’s going on with the groups on a cluster.

Public Interfaces

Client API changes

AdminClient

Add the following methods on the org.apache.kafka.client.admin.AdminClient  interface.

...

Proposed Changes

This KIP introduces a command-line tool for displaying all of the groups and their types.

Finally, in situations where command-line tools are used to administer a group of the wrong type, you’ll now be told the group type is wrong, rather than the group does not exist.

Listing groups

The KIP introduces a new tool called kafka-groups.sh to show all of the groups in a cluster, their types and the protocols they use. This lets you see consumer groups, share groups, Kafka Connect cluster groups, and any other custom groups all together. It doesn't replace the specific tools for the different types of group, but it does shed light on what's actually going on for administrators. Note that this does not require any changes to the Kafka protocol. The information is already available, but not directly accessible by the administrator.

The ListGroups RPC response returns three pieces of information for each group: group ID, type and protocol. For the common types of group, here is what they mean:

The new kafka-groups.sh tool makes all of this information available.

Describing groups

The behavior of AdminClient.describeConsumerGroups(Collection<String>) seems a little unusual. You can describe a collection of group IDs, some of which might exist and others might not. Here's how the response is built:

1. If the group is a consumer group and the client is authorized to describe the group and there was no error, the group information is returned, along with the authorized operations if requested.
2. If the group is not a consumer group (either does not exist or wrong type) and the client is authorized to describe the group and there was no error, the group information for a dead group is returned, along with the authorized operations if requested.
3. If the client is not authorized to the describe the group, the group information error code is set to GROUP_AUTHORIZATION_FAILED . 
4. If there was an error describing the group, the group information error code is set.

In cases (1) and (2), the admin client considers the operation a success, and this means the KafkaFuture  for this group completes successfully. In cases (3) and (4), the admin client considers the operation unsuccessful, and this means the KafkaFuture for this group completes exceptionally.

The behavior of AdminClient.describeShareGroups(Collection<String>)  was modelled on this for consistency.

This gives a problem though because even though it's not difficult to determine which groups have the incorrect type, it would take a breaking change to the admin client to discover this situation if using the admin client.

As a result, this KIP introduces a new option on DescribeConsumerGroupOptions  called validateGroupType  which changes the behavior in the case where a group ID is the wrong group type. For case (2) above, the GROUP_ID_NOT_FOUND  error in the ConsumerGroupDescribe response is translated into a GroupIdNotFoundException containing the error message from the RPC response. Previously, the error was swallowed meaning there was no opportunity to obtain an error message from the broker.

Public Interfaces

Client API changes

AdminClient

Add the following methods on the org.apache.kafka.client.admin.AdminClient  interface.

Here are the method signatures:

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

Here are the method signatures:

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

ListGroupOptions

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;
    }
}

AbstractListGroupsResult

package org.apache.kafka.clients.admin;
   
/**
 * This class 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 listings.
     */
    public KafkaFuture<Collection<T>> valid() {
    }
   
    /**
     * Returns a future which yields just the errors which occurred.
     */
    public KafkaFuture<Collection<Throwable>> errors() {
    }
}

ListGroupsResult

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) {
       return super(futurelistGroups(new ListGroupsOptions());
   }
 
   }
}

ListConsumerGroupsResult

This is changed to extends AbstractListGroupsResult<ConsumerGroupListing> .

ListShareGroupsResult

This is changed to extend AbstractListGroupsResult<ShareGroupListing> .

GroupListing

package org.apache.kafka.client.admin;
  
import org.apache.kafka.common.ShareGroupState;
  
/**
 * A listing of a group /**
    * List the groups available in the cluster.
  * <p>
 *
 The API of * this@param classoptions isThe evolving,options seeto {@linkuse Admin}when forlisting details.
 */
@InterfaceStability.Evolving
public class GroupListing {
  public GroupListing(String groupId, String protocol);
  public GroupListing(String groupId, GroupType type, String protocol);
  public GroupListing(String groupId, Optional<GroupType> type, String protocol);the groups.
    * @return The ListGroupsResult.
    */
   ListGroupsResult listGroups(ListGroupsOptions options);

ListGroupOptions

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

  

  /**
   * The id of the group.
   */
  public String groupId();
  
  /**
   * The group type.
   */
  public Optional<GroupType> type();
  
  /**
     * TheIf group protocol type.
types is set, only groups of these types will be returned by listGroups().
     * Otherwise, all groups are returned.
     */
    public StringListGroupsOptions protocolwithTypes(Set<GroupType> types);
}

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.

The error message in the RPCs gives more information about the failure.

Kafka protocol changes

Error codes

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
• ConsumerGroupHeartbeat
• ShareGroupDescribe
• ShareGroupHeartbeat

These RPCs are used by administrative tools and the new error code will help with the usability of the tools. Assuming that this KIP is delivered in the same release as KIP-848, no new RPC versions will be introduced to support the new error code. If this is not true, a new version of ConsumerGroupDescribe and ConsumerGroupHeartbeat would be required.

The remaining RPCs which work with consumer groups, such as ListOffsets and TxnOffsetCommit, continue to fail with GROUP_ID_NOT_FOUND  if used against a group of the wrong type.

Command-line tools

kafka-groups.sh

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

...

--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 based on group type and protocol in order to show consumer groups.

...

--describe

...

Describe the details of the groups.

...

--group-type <String: type>

...

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

...

--help

...

Print usage information.

...

--list

...

List all groups.

...

--protocol <String: protocol>

...

Filters the groups based on protocol type.

...

--version

...

Display Kafka version.

 {
        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;
    }
}

AbstractListGroupsResult

package org.apache.kafka.clients.admin;
   
/**
 * This class 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 listings.
     */
    public KafkaFuture<Collection<T>> valid() {
    }
   
    /**
     * Returns a future which yields just the errors which occurred.
     */
    public KafkaFuture<Collection<Throwable>> errors() {
    }
}

ListGroupsResult

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

ListConsumerGroupsResult

This is changed to extends AbstractListGroupsResult<ConsumerGroupListing> .

ListShareGroupsResult

This is changed to extend AbstractListGroupsResult<ShareGroupListing> .

GroupListing

package org.apache.kafka.client.admin;
  
import org.apache.kafka.common.ShareGroupState;
  
/**
 * 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, String protocol);
  public GroupListing(String groupId, GroupType type, String protocol);
  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();
}

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 .

DescribeConsumerGroupOptions

The following methods are added to this class.

/**
 * Set to true if describing a group which is not a consumer group fails.
 */
public DescribeConsumerGroupsOptions validateGroupType(boolean validateGroupType);

/**
 * Set to true if describing a group which is not a consumer group fails.
 */
public boolean validateGroupType();

If validateGroupType  is set, when the ConsumerGroupDescribe RPC response contains the error code INCONSISTENT_GROUP_TYPE , the describe fails with InconsistentGroupTypeException . If it is not set, the error code is treated the same as GROUP_ID_NOT_FOUND  and the describe succeeds by returning a ConsumerGroupDescription  in Dead  state.

DescribeShareGroupOptions

The following methods are added to this class.

/**
 * Set to true if describing a group which is not a share group fails.
 */
public DescribeConsumerGroupsOptions validateGroupType(boolean validateGroupType);

/**
 * Set to true if describing a group which is not a share group fails.
 */
public boolean validateGroupType();

If validateGroupType  is set, when the ShareGroupDescribe RPC response contains the error code INCONSISTENT_GROUP_TYPE , the describe fails with InconsistentGroupTypeException . If it is not set, the error code is treated the same as GROUP_ID_NOT_FOUND  and the describe succeeds by returning a ShareGroupDescription  in Dead  state.

Exceptions

The following new exception is added to the org.apache.kafka.common.error  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.

The error message in the RPCs gives more information about the failure.

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. Assuming that this KIP is delivered later than KIP-848, a new version of ConsumerGroupDescribe will be required to support the new error code.

The remaining RPCs which work with groups, such as ListOffsets and TxnOffsetCommit, continue to fail with GROUP_ID_NOT_FOUND  if used against a group of the wrong type.

Command-line tools

kafka-groups.sh

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

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:

$ bin/kafka-

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:

$ bin/kafka-groups.sh --bootstrap-server localhost:9092 --list
old-consumer-group
new-consumer-group
connect-cluster
share-group
schema-registry
simple-consumer-group

To describe all of the groups and their types:

$ bin/kafka-groups.sh --bootstrap-server localhost:9092 --describe
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:

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

To describe all of the consumer groups:

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

To describe all of the share groups:

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

kafka-consumer-groups.sh

For all operations which act on a single group, if that group exists but is not a consumer group, the command fails with a message indicating that the group type is incorrect, rather than the existing message that the group does not exist.

For example, if you try to describe a share group, the output will look like this:

$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group SG1
Error: Group 'SG1' is not a consumer group.

kafka-share-groups.sh

A new option --create  is added to this tool to create a share group. If the share group exists, the command succeeds. If the group does not exist, the share group is created. The output in these cases is the same. If the group exists but it's not a share group, the command fails.

$ bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --create list
old-consumer-group NewShareGroup
Share group 'NewShareGroup' created.

$ bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --create --group ExistingShareGroup
Share group 'ExistingShareGroup' created.


new-consumer-group
connect-cluster
share-group
schema-registry
simple-consumer-group

To describe all of the groups and their types:

$ bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --create --group ConsumerGroup
Error: Group 'ConsumerGroup' is not a share group.

Under the covers, it uses the AlterShareGroupOffsets RPC with an empty Topics  array.

Also, for all operations which act on a single group, if that group exists but is not a share group, the command fails with a message indicating that the group type is incorrect, rather than the existing messages that the group does not exist.

For example, if you try to describe a consumer group, the output will look like this:

$ bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --describe --group CG1
Error: Group 'CG1' is not a share group.

Proposed Changes

This KIP introduces a command-line tool for displaying all of the groups and their types.

Next, it introduces a way to create a share group administratively. If you create Kafka resources such as topics as part of deploying an application, you can now create share groups in the same way.

Finally, in situations where command-line tools are used to administer a group of the wrong type, you’ll now be told the group type is wrong, rather than the group does not exist.

Listing groups

The KIP introduces a new tool called kafka-groups.sh to show all of the groups in a cluster, their types and the protocols they use. This lets you see consumer groups, share groups, Kafka Connect cluster groups, and any other custom groups all together. It doesn't replace the specific tools for the different types of group, but it does shed light on what's actually going on for administrators. Note that this does not require any changes to the Kafka protocol. The information is already available, but not directly accessible by the administrator.

The ListGroups RPC response returns three pieces of information for each group: group ID, type and protocol. For the common types of group, here is what they mean:

...

Type

...

Protocol

...

Meaning

...

Classic

...

"consumer" 

...

Consumer group with the "classic" consumer group protocol

...

Classic

...

""

...

"Simple" consumer group that has committed offsets only

...

Consumer

...

"consumer" 

...

Consumer group with the KIP-848 consumer group protocol

...

Share

...

"share" 

...

Share group

...

Classic

...

"connect" 

...

Kafka Connect distributed worker cluster group

...

Classic

...

Any other string

...

Other customization of "classic" consumer group protocol, such as a schema registry

The new kafka-groups.sh tool makes all of this information available.

Creating groups

Groups are created dynamically on first use. For example, when you connect the first consumer in a consumer group, the group coordinator creates the group as a consumer group. This is convenient, but it does mean that you need to ensure that different types of group use distinct group IDs or the results will be unpredictable. This is because the group type depends upon how the group was created, whether it was a consumer group member, a distributed Kafka Connect cluster, or whatever.

Today, you can create a consumer group administratively before the first member joins by resetting the offsets, such as like this:

...

describe
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:

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

To describe all of the consumer groups:

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

To describe all of the share groups:

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

kafka-consumer-groups.sh

For all operations which act on a single group, if that group exists but is not a consumer group, the command fails with a message indicating that the group type is incorrect, rather than the existing message that the group does not exist.

For example, if you try to describe a share group, the output will look like this:

$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --

...

That’s a slightly contorted way to create a consumer group, but it does already exist and it is known. As a result, this KIP does not introduce a new way to create consumer groups.

...

describe --group SG1
Error: Group 'SG1' is not a consumer group.

kafka-share-groups.sh

...

For all operations which act on a single group, if that group exists but is not a share group, the command fails with a message indicating that the group type is incorrect, rather than the existing messages that the group does not exist.

For example, if you try to describe a consumer group, the output will look like this:

Code Block
$

...

bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --

...

describe --

...

group CG1 Error: Group 'CG1' is not a share group.

Compatibility, Deprecation, and Migration Plan

...