...

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.

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.

...

Here's another example. In Apache Kafka 3.8, if you use kafka-consumer-groups.sh --describe --group MYSHARE  where the group is a share group, the output is Error: Consumer group 'MYSHARE' does not exist . That's probably OK, but it's interesting to see how it gets there.

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 responds with error code NONE (0)  and returns the group status of Dead . It looks like a dead consumer group. There is no option of an error message in the protocol in this case, because the RPC doesn't support it.

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

Here are the method signatures:

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 with the default options.
    *
    * @param options The options to use when listing the groups<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(ListGroupsOptions options);

ListGroupOptions

package org.apache.kafka.client.admin;
 
import org.apache.kafka.common.GroupType;
 
) {
       return listGroups(new ListGroupsOptions());
   }
 
   /**
    * Options for {@link Admin#listGroups(ListGroupsOptions)}.
 List the groups available in the cluster.
    *
 * The API of* this@param classoptions isThe evolving,options seeto {@linkuse Admin}when forlisting details.
 */
@InterfaceStability.Evolving
public class ListGroupsOptions extends AbstractOptions<ListGroupsOptions> {

    /**
 the groups.
    * @return The ListGroupsResult.
    */
 If types 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> {

    /**is set, only groups of these types will be returned by listGroups().
     * Otherwise, all groups are returned.
     */
    public ListGroupsOptions withTypes(Set<GroupType> types) {
     * If types this.types = (types == null || types.isEmpty()) ? Collections.emptySet() : new HashSet<>(types);is set, only groups of these types will be returned by listGroups().
     * Otherwise, all return this;groups are returned.
    }

    /** */
    public *ListGroupsOptions Returns the list of group types that are requested or empty if no types have been specified.
     */
    public Set<GroupType> types() {withTypes(Set<GroupType> types) {
        this.types = (types == null || types.isEmpty()) ? Collections.emptySet() : new HashSet<>(types);
        return typesthis;
    }

}

AbstractListGroupsResult

package org.apache.kafka.clients.admin;
   
/**
 * This class implements the* commonReturns the APIslist thatof aregroup sharedtypes bythat resultsare classes
requested *or forempty variousif AdminClientno commandstypes forhave listingbeen groupsspecified.
 *   <p>
 */
 The API of thispublic classSet<GroupType> is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class AbstractListGroupsResult<T extends GroupListing> {

types() {
        return types;
    }
}

AbstractListGroupsResult

package org.apache.kafka.clients.admin;AbstractListGroupsResult(KafkaFuture<Collection<Object>> future);

    
/**
 * This class implements *the Returnscommon a futureAPIs that yieldsare eithershared anby exception,results or the full set of group listingsclasses
 * for various AdminClient commands for listing groups.
 * <p>
 * The */
API of this class publicis KafkaFuture<Collection<T>> all()evolving, see {
@link Admin} for  }
  
    /**details.
 */
@InterfaceStability.Evolving
public class AbstractListGroupsResult<T extends GroupListing> {

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

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

ListGroupsResult

package org.apache.kafka.clients.admin;
   
    
/**
     * The result of Returns a future which yields just the {@link Admin#listGroups(ListGroupsOptions)} callerrors which occurred.
 * <p>
 * The API*/
 of this class ispublic evolving, seeKafkaFuture<Collection<Throwable>> errors() {@link
 Admin} for details.
 }
}

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)*/
@InterfaceStability.Evolving
public class ListGroupsResult extends AbstractListGroupsResult<GroupListing> {
    ListGroupsResult(KafkaFuture<Collection<Object>> future) {
        super(future);
    }
}

ListConsumerGroupsResult

...

This class is modified to extend org.apache.kafka.clients.admin.GroupListing .

ShareGroupListing

.

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.This class is modified to extend org.apache.kafka.clients.admin.GroupListing .

Command-line tools

kafka-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 --group NewShareGroup
Share group 'NewShareGroup' existscreated.

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

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

...