Status

Current state: Under Discussion

...

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

Motivation

The capability to inspect consumer group metadata such as the list of members in the group and their partition assignments is crucial for debugging, monitoring, and administration, but there is currently no API which exposes this information for the new consumer. In the initial design of the group management protocol for the new consumer, it was assumed that group metadata would be persisted in Zookeeper by group coordinators. This would have allowed tooling to inspect Zookeeper directly to obtain this metadata in the same way that it does for the old consumer. However, the alternative of storing group metadata in Kafka was suggested in KAFKA-2017, which has several advantages, such as reducing overall load on Zookeeper and maintaining a clean separation of client state (stored in Kafka) from broker state (stored in Zookeeper). At the time of writing, no general agreement has been reached on this issue, so to avoid forcing a hasty decision, the consensus seems to be to revisit the question after the 0.9 release. 

With this issue momentarily tabled, we still have the problem of how group metadata can be viewed on an active cluster. Inspecting broker logs to find current group membership and assignments is not really a viable solution, even in the short term. To address this problem, we propose to modify the GroupMetadataRequest (formerly known as the ConsumerMetadataRequest) to support returning group metadata. This request is currently used by clients to find the group coordinator. Basically, the idea is to allow this request to return group metadata when it is received by the group's coordinator. For tooling, the workflow would be to send one GroupMetadata request to find the coordinator for the group, and one further request to retrieve the metadata from the coordinator. Additionally, we extend this request to support returning multiple groups so that it can be used to get a list of the groups hosted by each coordinator. add two new requests: ListGroups and DescribeGroup. The purpose of the ListGroups API is to get a simple list of the groups that are managed by each broker. To get a list of all groups in the cluster, tools will have to query each broker separately and combine the results.

Public Interfaces

The specific changes to the GroupMetadata request/response schemas are given below. Briefly, the schemas are modified to support querying multiple groups, and the response is modified to return group and member metadata. The request includes a flag to return the group metadata for all groups managed by the coordinator. We could alternatively use an empty list to indicate this case, but that pattern seems to be frowned upon after experience from the TopicMetadata request.

We also add a flag to the request which indicates whether member metadata is desired (although we have listed them separately, these two flags could be combined into a single Option field). The purpose of this field is to allow the retrieval of member-specific information such as topic subscriptions and partition assignments for the new consumer. Since this data can be large, however, the flag can be disabled for use cases which do not want the overhead. For example, clients which are only trying to discover the current coordinator would disable the flag.

The second API, DescribeGroup, is used to get detailed information about an individual group. This includes a list of the current members and their subscriptions/assignments. To use this API, tools will have to send on GroupMetadataRequest to locate the coordinator for the group, then use DescribeGroup to get the detailed information.

Public Interfaces

Below we show the proposed schemas for ListGroups and DescribeGroup. The ListGroups API takes no arguments and returns a simple list of the groups and their protocol types (e.g. "consumer" or "copycat"). The DescribeGroup API takes a single groupId and returns information on the current status of the group and its members. Below we describe these details in more detail.

ListGroups

ListGroupsRequest => 
 
ListGroupsResponse => [GroupId ProtocolType]

GroupMetadataRequest => IncludeAllGroups IncludeMemberMetadata Groups
  IncludeAllGroups => int8
  IncludeMemberMetadata => int8
  Groups => [GroupId]
    GroupId => String
 
GroupMetadataResponse => [ErrorCode GroupId Coordinator GroupMetadata MemberMetadata]
  ErrorCode => int16
  GroupId => string
  CoordinatorProtocolType => Id Host Port
    string

DescribeGroup

DescribeGroupRequest Id => int32GroupId
    HostGroupId => string
    Port
DescribeGroupResponse => int32
  GroupMetadata =>ErrorCode State ProtocolType Generation Protocol
 Leader Members
  StateErrorCode => int8string
    ProtocolTypeState => string
    GenerationProtocolType => int32string
    Protocol => string
  MemberMetadata => Leader Members
    Leader => string
    Members => [MemberId MemberIpAddressHost ClientId MemberMetadata MemberAssignment]
      MemberId => string
      MemberHostHost => string
      ClientId => string
      MemberMetadata => Bytesbytes
      MemberAssignment => Bytesbytes

Proposed Changes

We propose to implement the above request/response schema as version 1 of the group metadata request with the following semantics:

...

The ListGroups API is straightforward, so we focus on DescribeGroup below. The request is simple, but the response contains several fields worthy of further detail:

Group States: Below we list the possible group states and how it affects the associated metadata.

• Dead: There are no active members in the group. All other group metadata fields will be set to empty.
• Initializing: The group is loading metadata from its storage. All other group metadata fields will be set to empty and no member metadata will be returned.
• Rebalancing: The group is undergoing a rebalance. The generation, protocolType, and protocol will be set according to the prior generation. No member metadata will be returned.
• Stable: The group has a valid generation. Group and member metadata will be set based on the active generation.

To summarize, member metadata is only returned in the Stable state.

Member Metadata: Since the memberId is randomly generated, we must include additional information to help users identify the group member. The client host can be obtained from the session of the member's JoinGroup request and the clientId from the JoinGroup request itself. These fields will stay fixed until the next rebalance.

...

Error Codes: The following error codes are possible with this the DescribeGroup request:

• COORDINATOR_NOT_AVAILABLE: The broker could not determine the coordinator for the associated groupId. Under the current implementation, this would happen if the leader of the consumer offsets topic associated with the groupId were unavailable.
• NOT_COORDINATOR_FOR_GROUP: The broker is not the coordinator for the associated group. The Coordinator field will be set to the host information of the coordinator. Clients should use this to determine if they need an additional query to satisfy their request.
• NONE: The request was satisfied successfully the the group's coordinator.

Compatibility, Deprecation, and Migration Plan

Version 0 of the GroupMetadata request supports only a single groupId and returns only coordinator host information in the response. Since this is a subset of the functionality provided by this request, there should be no problem continuing to support it. One notable difference between the two versions is the use of the NOT_COORDINATOR_FOR_GROUP error. The new version uses this to indicate to clients that the broker is not the coordinator for the group, while the old request would return no error (since the request was only used to locate the coordinator and that was successfully accomplished). An alternative would be to keep the old behavior and have clients check the brokerId of the coordinator in the response to know whether an additional request is needed. It might also be possible to use an UNKNOWN group state to indicate the same.

One thing worth noting for future compatibility is that this request assumes that the coordinator will always have group and member metadata available to it. While it may not always be necessary to store this information in memory, the broker must have some way to load it on demand. Relaxing this requirement would generally mean deprecating and removing this API.These are new requests, so obviously it will not be possible to use them on an older broker. The easiest way for tooling (such as kafka-consumer-groups.sh) would be to include an option specifying whether the group is for the old consumer or new consumer (for copycat, there is no problem). As far as we can tell, that appears to already be the plan in KAFKA-2490.

Rejected Alternatives

• If the persistence question were settled, it may be possible to query the storage system directly. For example, if group metadata were stored in Zookeeper, then tools could query it directly in the same way that they currently do. However, in line with KIP-4, it seems preferable to have tools depend only on the Kafka API since this decouples them from the storage implementation and allows for simpler access control.
• Instead of extending the group metadata request, it would also be possible to use a new request type. The only real drawback is, well, that it requires a new request type. The preference in the Kafka community appears to be keeping the number of request types minimal and extending the GroupMetadata request to return group metadata seems reasonableThe initial version of this KIP proposed to extend the GroupMetadata request instead of adding the new request types. The main issue with this is that it leads to a complex request type which simultaneously tries to handle coordinator discovery, group listing, and group description. Separating this use cases into separate requests simplifies the design.