Status

Current state: Under discussion

Discussion thread: here [Change the link from the KIP proposal email archive to your own email thread]

JIRA: KAFKA-9130 [Change the link from KAFKA-1 to your own ticket]

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

At the moment, the ListGroups API always returns all groups. There are metrics that track the number of groups in several states but the only way to know the state of a group is to describe it using the DescribeGroups API. In large clusters with hundreds or thousands of groups, it can be hard for a cluster administrator to identify which groups are stable or which groups are dead and can be deleted for example. Basically this requires describing a large number of groups which is inefficiency in terms of time for the administrator and load for the cluster.

To improve that use case, I propose to allow listing groups by state. For example listing stable groups allows to immediately tell if several applications are healthy if their groups are all returned without needing to describe each group. This would allow building tools and user interfaces that can easily and quickly display informations about live groups.

Public Interfaces

ListGroups API

Update ListGroupsRequest to include a new flexible field "states".

{
  "apiKey": 16,
  "type": "request",
  "name": "ListGroupsRequest",
  // Version 1 and 2 are the same as version 0.
  //
  // Version 3 is the first flexible version.
  // KIP-518 adds the States flexible field
  "validVersions": "0-3",
  "flexibleVersions": "3+",
  

  ]
}

Update ListGroupsResponse to include a new field "group_state".

{
  "apiKey": 16,
  "type": "response",
  "name": "ListGroupsResponse",
  // Version 1 adds the throttle time.
  //
  // Starting in version 2, on quota violation, brokers send out responses before throttling.
  //
  // Version 3 is the first flexible version.
  "validVersions": "0-3",
  "flexibleVersions": "3+",
  "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": "ErrorCode", "type": "int16", "versions": "0+",
      "about": "The error code, or 0 if there was no error." },
    { "name": "Groups", "type": "[]ListedGroup", "versions": "0+",
      "about": "Each group in the response.", "fields": [
      { "name": "GroupId", "type": "string", "versions": "0+", "entityType": "groupId",
        "about": "The group ID." },
      { "name": "ProtocolType", "type": "string", "versions": "0+",
        "about": "The group protocol type." },
      

    ]}
  ]
}

If State is not specified or if it's empty in the request, from version 3, all groups will be returned with their states. At the moment, consumer group state is already serialized as a STRING in DescribeGroupsResponse, so serializing it the same way here.

AdminClient API

To expose this feature in the AdminClient API, ListConsumerGroupsOptions will be updated.

public class ListConsumerGroupsOptions extends AbstractOptions<ListConsumerGroupsOptions> {

    /**
     * Only groups in these states will be returned by listConsumerGroups()
     * By default, all groups are returned
     */
    public void inStates(List<ConsumerGroupState> states) {}

    public List<ConsumerGroupState> states() {}
}

Similarly, the state field will be exposed in ConsumerGroupListing:

public class ConsumerGroupListing {
    ...
    private final ConsumerGroupState groupState;

    /**
     * Consumer Group State
     */
    public ConsumerGroupState groupState() {
        return groupState;
    }
}

The state will be UNKNOWN when ListGroupRequest < 3 is used, for example when connecting to an older broker.

Command line tools

To expose this feature in the command line tools, ConsumerGroupCommand tool will be updated to support filtering by group states. A new argument "--states" will allow to specify a list of group states. This new option will only be valid when used with "–list". When specified, the state of each group will be printed next to the group id.

For example:

./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list
group1
group2

./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list --states
group1 stable
group2 empty

./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list --states stable
group1 stable

Compatibility, Deprecation, and Migration Plan

The default behaviour of ListGroups will stay the same and it will still list all groups if no state is specified.

With this KIP, having Describe authorization on the Cluster resource allows to retrieve the state of all groups. Previously this required having Describe authorization on each Group.

There is nothing to deprecate or migrate.

Rejected Alternatives

• Allow specifying a filter in ListGroups to retrieve groups by name or by state using wildcard
  
  • Unclear if it's a common use case