Status

Current state: Under Discussion

Discussion thread: here

JIRA: Unable to render Jira issues macro, execution error.

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

Motivation

The old consumer based group management allows for explicitly deleting a consumer group. This functionality is missing in the new consumer based group management where offsets are stored in a Kafka internal topic. The only way a consumer group can be removed is when its offsets are expired. The default retention time for committed offsets is currently 1 day (KIP-186 aims to increase that to 7 days). The override for this retention time is through the OffsetCommit API. Therefore, the offset cache can easily expand for consumer groups that are just waiting for their offsets to expire. With KIP-211 that proposes removing the retention_time field from the OffsetCommit API, the need for an alternate group deletion mechanism is more eminent than ever.

To mimic the tooling behavior of the old consumer group tool, this KIP proposes supporting the --delete switch for Kafka stored group offsets (new consumer) too. The behavior of --delete for the new consumer will be similar to that of the old consumer. For example, if group has active consumers the deletion will fail. In other words, a group can be deleted only when its state is Empty.

Public Interfaces

While existing APIs remain unchanged a new API DeleteGroups is introduced to remove offsets of the given group(s) from the internal offset cache. A deleteGroups() functionality will also be introduced in AdminClient API.

Proposed Changes

A new API is introduced for deleting consumer groups:

DeleteGroups Request (Version: 0) => [group_id]
  group_id => STRING
 
DeleteGroups Response (Version: 0) => [group_error_codes]
  throttle_time_ms => INT32
  error_code => INT16
  group_error_codes => group error_code 
    group_id => STRING
    error_code => INT16

In addition to this new API, we define the necessary operation/resource permission to use the API as Delete/Group. In other words, when SASL authentication is enabled a user needs to have Delete operation access to a group in order to delete it.

Potential error codes returned by the API (at the top level or per-group level or both) include:

• 0: None
• 15: COORDINATOR_NOT_AVAILABLE
• 24: INVALID_GROUP_ID
• 30: GROUP_AUTHORIZATION_FAILED

Similar to how the consumer group command line tool supports describing a single group at a time, the deletion functionality will be implemented the same way (deleting one group at a time). The protocol and the corresponding AdminClient API, however, support deleting multiple groups in one request. The new AdminClient API signature in kafka.admin.AdminClient would be like this:

def deleteConsumerGroups(groupIds: List[String]): DeleteGroupsResult

DeleteGroupsResult will include the top level as well as per group error codes. It will be derived from a DeleteGroupsResponse class that will be created in org.apache.kafka.common.requests to fully capture the outcome of a DeleteGroups API call from the broker.

Compatibility, Deprecation, and Migration Plan

The changes proposed by this KIP apply to the new Java-based consumer only, and are backward compatible because the suggested API does not already exist. New clients that work with older brokers will not be able to make use of this new API.

Rejected Alternatives

1. Using the newly introduced offset reset functionality of the consumer group command tool by injecting a retention_time of 0. This alternative was rejected in favor of KIP-211, that aims at removing the retention_time field from the OffsetCommit API.