Status

Current state: Under DiscussionAccepted

Discussion thread: here

JIRA: here

...

To assign tasks among Streams clients, the group coordinator uses topology metadata that is initialized using a dedicated StreamsGroupInitialize RPC members joins the group and persisted in the consumer offsets topic. To trigger topology initialization and/or detect topology inconsistencies, each heartbeat contains a unique ID of the topology (which can be derived automatically from the topology via hashing), and the group coordinator can ask a Streams client to trigger an initialize request in the heartbeat response. 

Streams groups

We will introduce a new group type called streams to the group coordinator. For each Streams group, we define new record key and value types for the group metadata, topology metadata and group member metadata. The records are persisted in the __consumer_offsets topic. A group can either be a streams group, a share group or a consumer group, defined by the first heartbeat request using the corresponding groupId GroupId. 

Joining, leaving and staying in a Streams group

During the complete lifetime of a Streams client, it its members continuously sends a send StreamsGroupHeartbeatRequests to the group coordinator. The request contains requests contain the following information:

• All non-assignment metadata as defined in the ConsumerGroupHeartbeatRequest such as the group ID, member ID, member epoch, instance ID, rack ID and rebalance timeout.

• The currently assigned active, standby and warm-up tasks. Each task is identified by a subtopology ID (which can be an arbitrary string in protocol for future extendabilityextensibility, even though subtopologies are currently numbered) and a partition number.

• The process ID to identify multiple Streams clients members running in the same process (i.e. the same Streams client).

• The user-defined endpoint to be used for running interactive queries.

• Topology metadata needed for creating internal topics and computing the assignment.

• The topology epoch, which A unique ID of the processing topology. The ID is defined by the client, and will be hash 0 by default. The epoch can be bumped by the client to deploy a new version of the topology representation by defaultmetadata.

• For each task, the current sum of task changelog changelog offsets and task changelog end offsets. These are used to determine whether a warm-up task is below acceptable.recovery.lag and can be used to optimize the assignment in the assignor. For more details on these offsets see "Cumulative task changelog offsets and task changelog end offsets" section below

• A flag to request shutdown of the whole application (only if the member epoch is the group leave epoch).

Group ID, member ID, member epoch, and instance ID (if defined by the client) are sent with each heartbeat request. Any other information that has not changed since the last heartbeat can be omitted, unless an error occurred.

...

• All non-assignment metadata as defined in the ConsumerGroupHeartbeatResponse such as the member ID (when joining as a new member), new member epoch and heartbeat interval ms.

• The current target assignment for this member, as sets of active, standby and warm-up task IDs.

• For each client in the group, the user-defined endpoint and the assigned tasks for IQA flag to indicate whether this client should initiate a StreamsGroupInitialize request to (re-)initialize the topology metadata

Again, fields that don’t change will be omitted from the heartbeat response.

Initializing and updating topology metadata on the broker

Whenever a Streams client receives a heartbeat response indicating that it should initialize or update the topology metadata on the broker, it initiates a StreamsGroupInitializeRequest on the group coordinator. The request member joins the Streams group, the first heartbeat request contains metadata of the topology. The metadata describes the topology as a set of subtopologies, each identified by a unique string identifier and containing the following metadata that is relevant for creation of internal topics and assignment:

• The set of source topics. Here, the source topics can be defined through a Google RE2/J regular expression.

• The set of repartition sink topics. We do not include non-repartition sink topics or output topics that the topology produces in using dynamic routing.

• The set of changelog topics, replication factors and other topic-level configurations.

• The set of repartitions topics, the number of partitions for each of them (optional), replication factors and other topic-level configurations.

• The set of copartition groups, that is, the set of source topics (user source topics, or internal repartition source topics) that have to have the same number of partitions, e.g. because they are being used inside a join.

Note that the metadata of topology does not specify the number of input partitions a source topology has, and similarly also does not specify the number of partitions for changelog topics, and may omit the number of partitions for some repartition topics (if it can be inferred). By leaving the number of partitions parametric, the broker can allow changing the number of partitions for certain topologies, without reinitializing the group. In particular, stateless topologies without repartitioning steps, can support any number of input partitions without reinitializing the broker-side topology.

The metadata sent to the broker via the heartbeat is data sent to the broker via the StreamsGroupInitialize RPC is persisted as a new record type in the __consumer_offsets topic and can be retrieved is part of the group coordinator state in subsequent requests. If one or more source topics or internal topics do not exist, the group coordinator invokes an asynchronous initializationwill enter state NOT_READY  and will attempt to create any missing internal topics (see more details below). Only after the topology metadata is initialized and all source topics and the internal topics are createdexist, the group coordinator starts assigning tasks to the members of the group. The StreamsGroupInitializeResponse only indicates whether an error occurred or not.

If internal topics exist, but no consistent set of topics can be created without changing the number of partitions of the existing topics, an error is returned. The topic-level configurations are not validated and only used for initialization.

Topology updates

Topology updates

To avoid unintentional topology updates or rollbacks, updating the broker-side topology after it has been initialized for a group requires bumping the client-side configuration topology.epoch, which defaults to 0. By default, whenever a Streams member joins the group with a topology that is different from Topology updates can be initiated by any member upon joining. Whenever a streams member joins the group (whether a new member, or a static member after a restart) with a topology ID that is different from the current topology of the streams group, and the topology epoch is not bumped, the group coordinator will ask one of the new members to reinitialize respond with an error, indicating that the member is incorrectly updating the broker-side topology. 

Any member can update the topology by sending a topology with a bumped topology epoch when joining.  The group coordinator then updates the broker-side topology with the topology metadata sent by the member. If the topology ID epoch of an existing member does not correspond to the current topology ID epoch of the group , this will anymore (so the topology of the member is stale w.r.t. the group's topology), this will be indicated in the heartbeat response (status STALE_TOPOLOGY)sent to that member (for informational purposes, e.g., logging a warning). 

The topology ID epoch of the group and the topology ID epoch of each member are provided to the assignor, which will make sure that members with a topology ID epoch that does not correspond to the group's topology ID epoch will not get any new tasks assigned, avoiding problems caused by topology incompatibilities.

It is up to the client implementation how topology IDs are defined. The protocol and the broker only requires that two clients with the same topology ID should have the same topology. In the initial implementation, the topology ID will be derived from the topology by deterministic hashing. However, it would also be possible to derive an implementation where topology IDs are explicitely defined through a configuration. This would, for example, allow the topologies to be manually versioned or topology IDs be derived from a git commit hash.

We also allow the StreamsGroupInitialize to be invoked by a client that is not a member of the group and was not asked to initialize the group. This can also be used to create an empty initialized group. However, default mode is that a group is initialized by one of the clients after joining. Adding a strict mode, where topology updates are requested explicitly, any implicit updates of the topology are rejected and members with incompatible topologies are fenced out of the group can be built on top of this protocol, but will be postponed to a follow-up KIP.

Describing and listing streams groups

Streams groups will be returned in ListGroupsResponse with GroupType equal to the string streams. The group ID of a streams group can be used in OffsetFetchRequest and OffsetCommitRequest as usual. Sending a ConsumerGroupHeartbeatRequest to a streams group will return an GROUP_ID_NOT_FOUND error. Sending a StreamsGroupHeartbeatRequest to a consumer group will similarly return an GROUP_ID_NOT_FOUND error.

There will be a new RPC DescribeStreamsGroup that returns, given the group ID, all metadata related to the streams group, such as

• The topology metadata that was provided through the StreamsGroupInitialize API.

• The latest member metadata that each member provided through the StreamsGroupHeartbeat API.

• The current target assignment generated by the assignor.

We are delaying extensions of the Admin API and the command-line tools to describe and modify streams groups to a follow-up KIP. 

Rebalance Process

Similar as in KIP-848 consumer groups, the rebalance process is entirely driven by the group coordinator, and based on the same three epochs: group epoch, assignment epoch, and member epoch. The main difference is the assigned resource - instead of assigning partitions to consumers, we assign tasks consisting of a subtopology ID and a partition number, and each task can be assigned to a client in three roles: as an active task, as a standby task or as a warm-up task.

We only explain the main differences in the rebalance process in regular consumer groups here.

• The group epoch will be bumped every time the topology is updated through the StreamsGroupInitialize API.

• The partition metadata for the group, which is tracked by the group coordinator, is the partition metadata of all input topics (i.e., user source topics and internal repartition topics we read from).

• The group epoch is bumped every time a member:

  • When a client with an assigned warm-up task reports a task offset and task end offset whose difference is less that acceptable.recovery.lag.

  • Updates its topology ID, rack ID, client tags or process ID. Note: Typically, these do not change within the lifetime of a Streams client, so this only happens when a member with static membership rejoins with an updated configuration.

Assignment

Every time the group epoch is bumped, a new task assignment is computed by calling the task assignor, which consists of assigning tasks as active, standby or warm-up tasks to members of the group. The task assignor is configured on the broker side both through a global static configuration that defined the default assingor for all groups, and a dynamic group-level configuration that sets the assignor for a specific group.

The AK implementation will provide the following assignors

• highly_available - Like the current HighAvailabilityTaskAssignor

• sticky - Like the current StickyTaskAssignor

Which assignors are provided by the group coordinator is not part of the protocol and is specific to each implementation. In this KIP, we do not introduce a pluggable interface for a task assignors, but we leave it open to do so in the future.

If no task assignor is configured globally or on the group-level, the broker picks the assignor. The broker is free to make a dynamic decision here - the AK implementation will choose highly_available for stateful topologies and sticky for stateless topologies.

Each task assignor must make sure to fulfill the following invariants:

• Each warm-up task that has reached acceptable.recovery.lag must be turned into an active task or a standby task, or be unassigned from the member.

• A stateful task (in the various roles) cannot be assigned to two clients with the same processId.

Cumulative task changelog offsets and task changelog end offsets

Whenever a stateful task is added or removed to/from a Streams client, or when warm-up task reaches acceptable.recovery.lag and in regular intervals, defined by a broker-side configuration task.offset.interval.ms, each client reports two sets of offsets, the sums of task changelog offsets and the sums of task changelog end offsets. These sets can be used by the assignors, in particular the highly_available assignor, to determine which tasks are caught-up to acceptable.recovery.lag and optimize the assignment when multiple clients with a partial copy of the state exist.

Cumulative task changelog offsets

The cumulative changelog offset of a task is the sum of all positions in the changelog topics of the local state stores belonging to the task. A streams client reports the cumulative changelog offsets for all tasks with local state. That is:

• For each processing (active) task, the sum of the offsets last fully replicated to the broker in the changelog topic of each local state. Under at-least-once guarantees, this is going to correspond to the high watermark, under exactly-once semantics, it’s going to be the last stable offset.

• For each restoring (active, standby or warm-up), task the sum of the positions in the changelog topic and checkpointed to the local state directory.

• For each dormant task (task which is not owned, but which has state locally on disk) the sum of the checkpointed positions in the changelog topic present in its state directory

Cumulative task end offsets

Similarly, the client reports the sum of the end offsets for the currently owned tasks, if available. This simplifies the broker-side task assignment, since the broker doesn’t need to fetch the current end-offsets. Specifically:

• For active processing tasks, the cumulative task end offset is the same as the cumulative task offset

• For all restoring tasks (active, standby or warm-up), the cumulative end-offset is the sum of the last end offsets cached by the restore consumer. If an end offset of a topic partition is unknown, no end-offset is reported.

• For dormant standby tasks, no end-offset is reported.

Using cumulative task offsets and cumulative task end offsets

By reporting cumulative task offsets and cumulative task end offsets instead of the cumulative task lag, we can determine most task lags in the group coordinator, without the group coordinator or the streams client having to do additional requests. Even if the end offsets of the topic partition is not known on that client, as long as it is known on another client in the group. However, it may happen that no end offset is known in the group coordinator - for example, for tasks that are dormant on at least one client, and no client currently owns that task as active, standby or warm-up tasks. In these cases, the task with unknown end-offsets are never considered to be caught-up, however, the cumulative task offset can still be used when deciding on where to place a new active, standby or warm-up task - typically, by selecting the client with the maximal cumulative task offset that fulfils all other requirements.

Reconciliation of the group

Once a new target assignment is installed, each member will independently reconcile their current assignment with their new target assignment. Ultimately, each member will converge to their target epoch and assignment. The reconciliation process handles active tasks and standby/warm-up tasks differently:

• Active tasks are reconciled like topic partitions in the KIP-848 consumer group protocol, that is, for reconciling the target assignment, the reconciliation follows three phases:

  • The group coordinator first asks the member to revoke any active tasks that are not assigned to that member in the target assignment any more, by removing those active tasks from the set of tasks sent to the member in the heartbeat response

  • The member must first confirm revocation of these active tasks by removing them from the set of active tasks sent in the heartbeat request

  • Now, active tasks will be incrementally assigned to the member. An active task is assigned as soon as no other client owns it anymore, that is, once the previous owner (if any) has confirmed the revocation of the corresponding active task.

• Standby and warm-up tasks are reconciled in parallel with active tasks, but following slightly different logic:

  • As for active tasks, standby and warm-up tasks removed from the members target assignment are removed from the assignment sent in the heartbeat response immediately, that is, with the next heartbeat. The member confirms the revocation of the tasks as soon as the state directory (or any other resource related to the task) is closed.

  • Newly assigned standby and warm-up tasks await that any member with the same processId owning that task (as active or standby) confirms revocation of the task by removing it from the corresponding set in a heartbeat response. That is, a task that is not an active task or a standby task on any other member with the same processId can be assigned immediately. Otherwise, it is assigned as soon as the blocking task is revoked. Target assignments that assign a task to two members with the same processId are invalid.

Streams Group States

The possible states of the streams group are EMPTY, ASSIGNING, RECONCILING , STABLE, DEAD as for consumer groups, and additionally INITIALIZING for streams groups that are pending initialization with a topology through the StreamsGroupInitialize RPC.

Global & Dynamic Group Configuration

We make core assignment options configurable centrally on the broker, without relying on each clients configuration. This allows tuning a streams group without redeploying the streams application. The three core assignment options will be introduced on the broker-side: acceptable.recovery.lag, max.warmup.replicas and num.standby.replicas. They can be configured both globally on the broker, and dynamically for specific streams groups through the IncrementalAlterConfigs and DescribeConfigs RPCs.

Public Interfaces

This section lists the changes impacting the public interfaces.

KRPC

New Errors

The conditions in which these errors are returned are stated further down.

• STREAMS_INVALID_TOPOLOGY - The supplied topology is invalid. Returned if the client sends a topology that does not fulfill the expected invariants, see below in the sections "Request Validation".

• STREAMS_MISSING_SOURCE_TOPICS - There are source topics missing for a topology that is supposed to be initialized. Also returned if the source topic regular expression matched no topics.

• STREAMS_INCONSISTENT_INTERNAL_TOPICS - There are internal topics present on the broker that are not consistent with the internal topic requirements of the provided topology.

StreamsGroupInitialize

The StreamsGroupInitialize API is called to initialize the topology information on the broker side, that will be reused by all members of the group. The StreamsGroupInitialize RPC is normally called only by a single application to explicitly initialize the group, after the broker requested the initialization from that member. It can, however, also be used to create and initialize an empty group.

Request Schema

{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker"],
  "name": "StreamsGroupInitializeRequest",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "GroupId", "type": "string", "versions": "0+", "entityType": "groupId",
      "about": "The group identifier." },
    { "name":  "Topology", "type": "[]Subtopology", "versions": "0+",
      "about": "The sub-topologies of the streams application.",
      "fields": [
        { "name": "Subtopology", "type": "string", "versions": "0+",
          "about": "String to uniquely identify the sub-topology. Deterministically generated from the topology" },
        { "name": "SourceTopics", "type": "[]string", "versions": "0+",
          "about": "The topics the topology reads from." },
        { "name": "SourceTopicRegex", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
           "about": "The regular expressions identifying topics the sub-topology reads from. null if not provided." },
        { "name": "StateChangelogTopics", "type": "[]TopicInfo", "versions": "0+",
          "about": "The set of state changelog topics associated with this sub-topology. Created automatically." },
        { "name": "RepartitionSinkTopics", "type": "[]string", "versions": "0+",
          "about": "The repartition topics the sub-topology writes to." }, 
        { "name": "RepartitionSourceTopics", "type": "[]TopicInfo", "versions": "0+",
          "about": "The set of source topics that are internally created repartition topics. Created automatically." }
      ]
    }
  ],
  "commonStructs": [
    { "name": "TopicConfig", "versions": "0+", "fields": [
        { "name": "key", "type": "string", "versions": "0+",
          "about": "The key of the topic-level configuration." },
        { "name": "value", "type": "string", "versions": "0+",
          "about": "The value of the topic-level configuration," }
      ]
    },
    { "name": "TopicInfo", "versions": "0+", "fields": [
      { "name": "Name", "type": "string", "versions": "0+",
        "about": "The name of the topic." },
      { "name": "Partitions", "type": "int32", "versions": "0+",
        "about": "The number of partitions in the topic. Can be 0 if no specific number of partitions is enforced. Always 0 for changelog topics." },
      { "name": "TopicConfigs", "type": "[]TopicConfig", "versions": "0+", "nullableVersions": "0+", "default": "null",
        "about": "Topic-level configurations as key-value pairs."
      }
    ]}
  ]
}

Required ACL

• READ on the group

• CREATE on cluster resource, or CREATE on all topics in StateChangelogTopics and RepartitionSourceTopics

• DESCRIBE_CONFIGS on all topics included in the message

Request Validation

INVALID_REQUEST is returned should the request not obey to the following invariants:

• GroupId must be non-empty.

STREAMS_INVALID_TOPOLOGY is returned should the request not obey the following invariants:

• A stateChangelogTopic topics must not have a defined partition number.

• A repartitionSourceTopic cannot be in sourceTopics or stateChangelogTopics of any subtopology.

• A stateChangelogTopic cannot be in sourceTopics or repartitionSinkTopic or repartitionSourceTopics of any subtopology.

• A repartitionSourceTopic of one subtopology must be a repartitionSinkTopic of at least one other subtopology.

Request Handling

When the group coordinator handles a StreamsGroupInitialize request:

• It performs request validation.
• Looks up or creates the group. GROUP_ID_NOT_FOUND is returned if the group ID is associated with a group type that is not streams or classic (the latter will be allowed for migration).

• Creates all required internal topics.

  • STREAMS_MISSING_SOURCE_TOPICS is returned if there are source topics missing for a topology that is supposed to be initialized. Also returned if the source topic regular expression matched no topics.

  • STREAMS_INCONSISTENT_INTERNAL_TOPICS is returned if there are internal topics present on the broker that are not consistent with the internal topic requirements of the provided topology.

• Writes the topology, keyed with the groupId to the consumer offset topic. Existing records will be overwritten.

Response Schema

{
  "apiKey": TBD,
  "type": "response",
  "name": "StreamsGroupInitializeResponse",
  "validVersions": "0",
  "flexibleVersions": "0+",
  // Supported errors:
  // - GROUP_AUTHORIZATION_FAILED (version 0+)
  // - GROUP_ID_NOT_FOUND (version 0+)
  // - NOT_COORDINATOR (version 0+)
  // - COORDINATOR_NOT_AVAILABLE (version 0+)
  // - COORDINATOR_LOAD_IN_PROGRESS (version 0+)
  // - INVALID_REQUEST (version 0+)
  // - TOPIC_AUTHORIZATION_FAILED (version 0+) 
  // - CLUSTER_AUTHORIZATION_FAILED (version 0+)
  // - STREAMS_INVALID_TOPOLOGY (version 0+)
  // - STREAMS_MISSING_SOURCE_TOPICS (version 0+)
  // - STREAMS_INCONSISTENT_INTERNAL_TOPICS (version 0+)
  "fields": [
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
      "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 top-level error code, or 0 if there was no error" },
    { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "The top-level error message, or null if there was no error." }
  ]
}

StreamsGroupHeartbeat

The StreamsGroupHeartbeat API is the new core API used by streams application to form a group. The API allows members to advertise their state, and their owned tasks. The group coordinator uses it to assign/revoke tasks to/from members. This API is also used as a liveness check.

Request Schema

The member must set all the (top level) fields with the exception of RackId and InstanceId when it joins for the first time or when an error occurs (e.g. request timed out). Otherwise, it is expected to only fill in the fields which have changed since the last heartbeat.

broker will reject members that attempt to join with a stale topology epoch, or with the current topology epoch but different topology metadata.

Handling topic topology mismatches

It can happen that the group is initialized to a topology, but source / sink or internal topics required by the topology do not exist or differ in their configuration from what is required for the topology to successfully execute. This is typically detected during the handling of the streams group heartbeat in the group coordinator, where we detect changes in either the topology or the topic metadata on the broker, triggering "topology configuration" process, in which the group coordinator performs the following steps:

• Check that all source topics exists, resolve source topic regular expressions and check that each of them resolve to at least one topic.
• Check that "copartition groups" are satisfied, that is, all source topics that are supposed to be copartitioned are indeed copartitioned.
• Derive the required number of  partitions for all internal topics from the source topic configuration.
• Check that all internal topics exist with the right configuration.

If any source topics or internal topics are missing, the group enters a state NOT_READY. In NOT_READY, all heartbeats will be handled as usual (so they typically should not fail), but in the heartbeat response, the status will indicate which kind of problem exists - all members will get an empty assignment when the group is in NOT_READY  state. Below, we describe the behavior of the protocol when any mismatch is detected during topic/topology configuration. The behaviors are ordered by precedence, for example, if source topics and internal topics are missing, then the groups takes on the behavior for missing source topics, not the behavior for missing internal topics.

1. Source topics missing
   Condition: A source topic is missing or a source topic regex resolves to zero topics.
   Behavior: The group will enter/remain in state NOT_READY.  Heartbeat responses will indicate status MISSING_SOURCE_TOPICS .  In the status detail, we specify all missing source topics and all regular expressions matching zero topics.
2. Topics incorrectly partitioned
   Condition: Some topics are incorrectly partitioned, if two topics are supposed to be copartitioned according to the topology, but in the current topic metadata on the broker, the number of partitions for the two topics is different, or the number of partitions in a changelog topic does not correspond to the maximal number of source topic partition for that subtopology.
   Behavior: The group will enter/remain in state NOT_READY.  Heartbeat responses will indicate status INCORRECTLY_PARTITIONED_TOPICS . In the status detail, we specify at least one incorrectly partitioned topic.
3. Internal topics are missing
   Condition: One or more internal topics are missing.
   Behavior: If discovered during a heartbeat, the group coordinator will attempt to create the internal topics by sending a corresponding topic create request will be sent to the Kraft coordinator. There can only be one such request in-flight at a time, an appropriate back-off mechanism will be used to prevent too many attempts to create the topics. If the appropriate ACL for topic creation are not assigned to the principle executing the heartbeat, no such attempt will be made. The group will enter/remain in state NOT_READY .  Heartbeat responses will indicate status MISSING_INTERNAL_TOPICS. In the status detail, we specify whether an attempt to create the topics was made, whether and why a previous attempt failed, whether sufficient ACLs to create the topics are available.
4. Topic configuration mismatches
   If an internal topic exists, but does not have the same configuration as defined in the topology (all parameters of the topic beside number of partitions, that is, replication factor, retention time, etc.), this will be logged on the broker, but otherwise be ignored.

Describing and listing streams groups

Streams groups will be returned in ListGroupsResponse with GroupType equal to the string streams. The group ID of a streams group can be used in OffsetFetchRequest and OffsetCommitRequest as usual. Sending a ConsumerGroupHeartbeatRequest to a streams group will return an GROUP_ID_NOT_FOUND error. Sending a StreamsGroupHeartbeatRequest to a consumer group will similarly return an GROUP_ID_NOT_FOUND error.

There will be a new RPC DescribeStreamsGroup that returns, given the group ID, all metadata related to the streams group, such as

• The topology metadata of the group. This topology metadata is the result of the above "topology configuration" process, so it contains the topology metadata as initialized by one of the streams group members, but with a concrete number of partitions for each topic, and with source topic regular expressions resolved to a specific set of topics.

• The topology epoch of the group.

• The latest member metadata that each member provided through the StreamsGroupHeartbeat API.

• The current target assignment generated by the assignor.

We also include extensions of the Admin API and the command-line tools to describe and modify streams groups.

Rebalance Process

Similar as in KIP-848 consumer groups, the rebalance process is entirely driven by the group coordinator, and based on the same three epochs: group epoch, assignment epoch, and member epoch. The main difference is the assigned resource - instead of assigning partitions to consumers, we assign tasks consisting of a subtopology ID and a partition number, and each task can be assigned to a client in three roles: as an active task, as a standby task or as a warm-up task.

We only explain the main differences in the rebalance process in regular consumer groups here.

• The partition metadata for the group, which is tracked by the group coordinator, is the partition metadata of all input topics (i.e., user source topics and internal repartition topics we read from).

• The group epoch is bumped:

  • When a member joins or leaves the group.
  • When a member is fenced or removed from the group by the group coordinator.
  • When the partition metadata is updated. For instance when a new partition is added or a new topic matching the subscribed topics is created.
  • When a member with an assigned warm-up task reports a task changelog offset and task changelog end offset whose difference is less than acceptable.recovery.lag.
  • When a member updates its topology metadata, rack ID, client tags or process ID. Note: Typically, these do not change within the lifetime of a Streams client, so this only happens when a member with static membership rejoins with an updated configuration.
  • When an assignment configuration for the group is updated.

Assignment

Every time the group epoch is bumped, a new task assignment is computed by calling the task assignor, which consists of assigning tasks as active, standby or warm-up tasks to members of the group. The task assignor is configured on the broker side both through a global static configuration that defined the default assignor for all groups, and a dynamic group-level configuration that sets the assignor for a specific group.

The AK implementation will provide the following assignors

• highly_available - Similar to the current HighAvailabilityTaskAssignor

• sticky - similar to the current StickyTaskAssignor

Which assignors are provided by the group coordinator is not part of the protocol and is specific to each implementation. In this KIP, we do not introduce a pluggable interface for a task assignors, but we leave it open to do so in the future.

If no task assignor is configured globally or on the group-level, the broker picks the assignor. The broker is free to make a dynamic decision here - the AK implementation will choose highly_available for stateful topologies and sticky for stateless topologies.

Each task assignor must make sure to fulfill the following invariants:

• Each warm-up task that has reached acceptable.recovery.lag must be turned into an active task or a standby task, or be unassigned from the member.

• A stateful task (in the various roles) cannot be assigned to two clients with the same processId.

Cumulative task changelog offsets and task changelog end offsets

Whenever a stateful task is added or removed to/from a Streams client, or when warm-up task reaches acceptable.recovery.lag and in regular intervals, defined by a broker-side configuration task.offset.interval.ms, each client reports two sets of offsets, the sums of task changelog offsets and the sums of task changelog end offsets. These sets can be used by the assignors, in particular the highly_available assignor, to determine which tasks are caught-up to acceptable.recovery.lag and optimize the assignment when multiple clients with a partial copy of the state exist.

Cumulative task changelog offsets

The cumulative changelog offset of a task is the sum of all positions in the changelog topics of the local state stores belonging to the task. A Streams member reports the cumulative changelog offsets for all tasks with local state. That is:

• For each processing (active) task, the sum of the offsets last fully replicated to the broker in the changelog topic of each local state. Under at-least-once guarantees, this is going to correspond to the high watermark, under exactly-once semantics, it’s going to be the last stable offset.

• For each restoring (active, standby or warm-up), task the sum of the positions in the changelog topic and checkpointed to the local state directory.

• For each dormant task (task which is not owned, but which has state locally on disk), the sum of the checkpointed positions in the changelog topic present in its state directory. The client only reports the offsets for dormant tasks that we manage to acquire the lock to the state directory for.

Members that run in the same process (and use the same state directory) may report offsets for overlapping sets of dormant tasks. These offsets can conflict (since they are recorded at different points in time), but these conflicts can easily be resolved by taking the most recently received offsets. The current assignors will be updated to take this into account.

Cumulative task changelog end offsets

Similarly, the member reports the sum of the end offsets in all changelog topics for the currently owned tasks, if available. This simplifies the broker-side task assignment, since the broker doesn’t need to fetch the current end-offsets. Specifically:

• For active processing tasks, the cumulative task changelog end offset is the same as the cumulative task changelog offset

• For all restoring tasks (active, standby or warm-up), the cumulative end-offset is the sum of the last end offsets cached by the restore consumer. If an end offset of a topic partition is unknown, no end-offset is reported.

• For dormant tasks, no end-offset is reported.

Using cumulative task changelog offsets and cumulative task changelog end offsets

By reporting cumulative task changelog offsets and cumulative task changelog end offsets instead of the cumulative task lag, we can determine most task lags in the group coordinator, without the group coordinator or the streams client having to do additional requests. Even if the end offsets of the topic partition is not known to that member, as long as it is known to another member in the group. However, it may happen that no end offset is known in the group coordinator - for example, for tasks that are dormant on at least one client, and no member currently owns that task as active, standby or warm-up tasks. In these cases, the task with unknown end-offsets are never considered to be caught-up, however, the cumulative task changelog offset can still be used when deciding on where to place a new active, standby or warm-up task - typically, by selecting the client with the maximal cumulative task changelog offset that fulfils all other requirements.

Reconciliation of the group

Once a new target assignment is installed, each member will independently reconcile their current assignment with their new target assignment. Ultimately, each member will converge to their target epoch and assignment. The reconciliation process handles active tasks and standby/warm-up tasks differently:

• Active tasks are reconciled like topic partitions in the KIP-848 consumer group protocol, that is, for reconciling the target assignment, the reconciliation follows three phases:

  • The group coordinator first asks the member to revoke any active tasks that are not assigned to that member in the target assignment any more (if either no warmup tasks are used at all, or if the target member has a hot standby/warmup task), by removing those active tasks from the set of tasks sent to the member in the heartbeat response

  • The member must first confirm revocation of these active tasks by removing them from the set of active tasks sent in the heartbeat request

  • Now, active tasks will be incrementally assigned to the member. An active task is assigned as soon as no other client owns it anymore, that is, once the previous owner (if any) has confirmed the revocation of the corresponding active task.

• Standby and warm-up tasks are reconciled in parallel with active tasks, but following slightly different logic:

  • As for active tasks, standby and warm-up tasks removed from the members target assignment are removed from the assignment sent in the heartbeat response immediately, that is, with the next heartbeat. The member confirms the revocation of the tasks as soon as the state directory (or any other resource related to the task) is closed.

  • Newly assigned standby and warm-up tasks await that any member with the same processId owning that task (as active or standby) confirms revocation of the task by removing it from the corresponding set in a heartbeat response. That is, a task that is not an active task or a standby task on any other member with the same processId can be assigned immediately. Otherwise, it is assigned as soon as the blocking task is revoked. Target assignments that assign a task to two members with the same processId are invalid.

Streams Group States

The possible states of the streams group are EMPTY, ASSIGNING, RECONCILING , STABLE, NOT_READY, DEAD. The states are similar as consumer groups, but contain an extra stat NOT_READY, which is used when a group is not empty and has a topology set, but not all topics are present to start processing using the topology.

Global & Dynamic Group Configuration

We make core assignment options configurable centrally on the broker, without relying on each clients configuration. This allows tuning a streams group without redeploying the streams application. The three core assignment options will be introduced on the broker-side: acceptable.recovery.lag, num.warmup.replicas, num.standby.replicas, and rack.aware.assignment.tags. They can be configured both globally on the broker, and dynamically for specific streams groups through the IncrementalAlterConfigs and DescribeConfigs RPCs. 

We store the last used assignment configuration in the group metadata on the broker. This way, we can detect if an assignment configuration is dynamically changed, and we can trigger reassignment immediately.

Online Migration from a Classic Consumer Group to a Streams Group

As in KIP-848, upgrading from a classic consumer group currently used in Kafka Streams to the new streams group is possible by rolling the Streams clients, assuming the Streams protocol is enabled on the brokers. When the first consumer of a Streams client using the new Streams rebalance protocol joins the group, the group is converted from a classic group to a streams group. When the last consumer of an Streams client using the new Streams rebalance protocol leaves the group, the group is converted back to a classic group. Note that the group epoch starts at the current group generation. During the migration, all the JoinGroup, SyncGroup and Heartbeat calls from the non-upgraded consumers are translated to the new Streams protocol.

Let's recapitulate how the classic rebalance protocol works in Streams. First, the consumers in a Streams client join or re-join the group with the JoinGroup API. The JoinGroup request contains the subscribed topics, the owned partitions, the generation ID, the Streams-specific subscription info, and some other fields. The Streams-specific subscription info contains the process ID of the Streams client, the owned active and standby tasks, the task offset sums, the user endpoints for Interactive Query, and some more fields. When all the consumers of the Streams clients of the Streams application have joined, the group coordinator picks a leader for the group out of the consumers and sends back the JoinGroup response to all the members of the group (i.e., consumers that joined). The JoinGroup response contains the member ID, the generation ID, and the member ID of the leader (to make the leader aware of leadership) as well as all the consumer and Streams-specific metadata about subscribed topics, owned partitions, tasks, etc. The leader uses the data in the JoinGroup response for computing the assignment. Second, all the members in the Streams clients collect their assignment—computed by the leader—by using the SyncGroup API. The leader sends the computed assignment with the SyncGroup request to the group coordinator, and the group coordinator distributes the assignment to the members through the SyncGroup response. In parallel, the members heartbeat with the Heartbeat API in order to maintain their session. The Heartbeat API is also used by the group coordinator to inform the members about an ongoing rebalance. All those interactions are synchronized on the generation of the group. It is important to note that the consumer does not make any assumption about the generation ID. It basically uses what it receives from the group coordinator. The classic rebalance protocol used in Streams supports two modes: Eager and Cooperative. In the eager mode, the consumer revokes all its partitions before rejoining the group during a rebalance. In the cooperative mode, the consumer does not revoke any partitions before rejoining the group. However, it revokes the partitions that it does not own anymore when it receives its new assignment and rejoins immediately if it had to revoke any partitions.

The Streams rebalance protocol relies on the StreamsGroupHeartbeat API to do all the above. Concretely, the API updates the group state, provides the active, standby, and warm-up tasks owned by the Streams client, gets back the assignment, and updates the session. We can remap those to the classic protocol as follows: The JoinGroup API updates the group state and provides the active and standby tasks  owned (warm-ups are standby tasks in the classic protocol), the SyncGroup API gets back the task assignment, and the Heartbeat API updates the session. The main difference here is that the JoinGroup and SyncGroup do not run continuously. The group coordinator has to trigger it when it is needed by returning the REBALANCE_IN_PROGRESS error in the heartbeat response.

The implementation of the new Streams protocol in the group coordinator will handle the JoinGroup and SyncGroup APIs of the classic protocol. The group coordinator will ensure that a rebalance is triggered when the assignment of a Streams client on the classic protocol needs to be updated by returning a REBALANCE_IN_PROGRESS error in the heartbeat response.

When the first consumer of a Streams client that uses the new Streams rebalance protocol joins a classic group, the classic group in the group coordinator will be transformed to a streams group. The generation ID becomes the group epoch. The consumer of the Streams client initializes the group with the topology metadata. A rebalance is triggered by the group coordinator for all consumers still on the classic protocol by sending the REBALANCE_IN_PROGRESS error in the heartbeat. The consumers on the classic protocol send their owned active and standby tasks in the JoinGroup request to the group coordinator. The consumers on the Streams protocol send their owned active, standby, and warm-up tasks (i.e., should be empty since they just joined) through the StreamsGroupHeartbeat request to the group coordinator. In contrast to the classic protocol, the group coordinator does not pick a leader for computing the assignment but computes the assignment itself. That is the target assignment. The group epoch is increased. From the target assignment, the current member assignment is computed depending on the owned tasks. If members do not need to revoke any tasks, their member epoch is increased to the group epoch. If the members need to revoke a task, their member epoch stays the same. The group coordinator sends the new member epoch alongside the current member assignment through the StreamsGroupHeartbeat response to the members on the Streams protocol. The members still on the classic protocol receive the member epoch through the JoinGroup response, but they still need to wait for the SyncGroup response for their current assignment. Basically, the group coordinator translates the JoinGroup and SyncGroup API to the Streams protocol internally and communicates to members still on the classic protocol via JoinGroup, SyncGroup as well as Heartbeat and with the members on the Streams protocol via the StreamsGroupHeartbeat.

The Streams protocol also assigns warm-up tasks. However, the classic protocol does not have any notion of a warm-up task. If the group coordinator assigns a warm-up task to a Streams client on the classic protocol, that warm-up task is translated to a standby task in the assignment for the Streams client on the classic protocol. The group coordinator chooses one of the Streams clients on the classic protocol to trigger a probing rebalance.

A more detailed description of this process can be found in KIP-848 in Section Supporting Online Consumer Group Upgrade.  

Example

• classic group (generation=23)
  
  • A
  • B

• assignment
  • A - active=[0_0, 0_2, 0_4], standby=[0_1, 0_3, 0_5]
  • B - active=[0_1, 0_3, 0_5], standby=[0_0, 0_2, 0_4]

C joins using the Streams protocol. The classic group is transformed to a streams group.

• streams group (group epoch=24)
  • A (classic)
  • B (classic)
  • C (streams)
• target assignment (epoch=24)
  • A - active=[0_0, 0_2, 0_3], standby=[0_1, 0_5], warm-up=[]
  • B - active=[0_1, 0_4, 0_5], standby=[0_2, 0_3], warm-up=[]
  • C - active=[], standby=[0_0, 0_4], warm-up=[0_2, 0_5]
• member assignment
  • A
    • Receives REBALANCE_IN_PROGRESS error in heartbeat response
    • JoinGroupRequest: active=[0_0, 0_2, 0_4], standby=[0_1, 0_3, 0_5]
    • JoinGroupResponse: generation ID=24
    • SyncGroupResponse: active=[0_0, 0_2, 0_3], standby=[0_1, 0_5]
  • B
    • Receives REBALANCE_IN_PROGRESS error in heartbeat response
    • JoinGroupRequest: active=[0_1, 0_3, 0_5], standby=[0_0, 0_2, 0_4]
    • JoinGroupResponse: generation ID=24
    • SyncGroupResponse: active=[0_1, 0_4, 0_5], standby=[0_2, 0_3]
  • C
    • StreamsGroupHeartbeat: epoch=24, active=[], standby=[0_0, 0_4], warm-up=[0_2, 0_5]

    C's warm-up task 0_2 is caught up. A is requested to revoke active task 0_2, thus A does not increase its generation ID.

• streams group (group epoch=25)
  • A (classic)
  • B (classic)
  • C (streams)
• target assignment (epoch=25)
  • A - active=[0_0, 0_3], standby=[0_1, 0_5], warm-up=[]
  • B - active=[0_1, 0_4, 0_5], standby=[0_2, 0_3], warm-up=[]
  • C - active=[0_2], standby=[0_0, 0_4], warm-up=[0_5]
• member assignment
  • A
    • Receives REBALANCE_IN_PROGRESS error in heartbeat response
    • JoinGroupRequest: active=[0_0, 0_2, 0_3], standby=[0_1, 0_5], warm-up=[]
    • JoinGroupResponse: generation ID=24
    • SyncGroupResponse: active=[0_0, 0_3], standby=[0_1, 0_5]
  • B
    • Receives REBALANCE_IN_PROGRESS error in heartbeat response
    • JoinGroupRequest: active=[0_1, 0_4, 0_5], standby=[0_2, 0_3]
    • JoinGroupResponse: generation ID=25
    • SyncGroupResponse: active=[0_1, 0_4, 0_5], standby=[0_2, 0_3]
  • C
    • StreamsGroupHeartbeat: epoch=25, active=[], standby=[0_0, 0_4], warm-up=[0_5]

A follow-up rebalance is triggered so that A can report the revoked active task 0_2. Since A does not need to revoke tasks anymore the generation ID is increased.

• streams group (group epoch=25)
  • A (classic)
  • B (classic)
  • C (streams)
• target assignment (epoch=25)
  • A - active=[0_0, 0_3], standby=[0_1, 0_5], warm-up=[]
  • B - active=[0_1, 0_4, 0_5], standby=[0_2, 0_3], warm-up=[]
  • C - active=[0_2], standby=[0_0, 0_4], warm-up=[0_5]
• member assignment
  • A
    • Receives REBALANCE_IN_PROGRESS error in heartbeat response
    • JoinGroupRequest: active=[0_0, 0_3], standby=[0_1, 0_5], warm-up=[]
    • JoinGroupResponse: generation ID=25
    • SyncGroupResponse: active=[0_0, 0_3], standby=[0_1, 0_5]
  • B
    • Receives REBALANCE_IN_PROGRESS error in heartbeat response
    • JoinGroupRequest: active=[0_1, 0_4, 0_5], standby=[0_2, 0_3]
    • JoinGroupResponse: generation ID=25
    • SyncGroupResponse: active=[0_1, 0_4, 0_5], standby=[0_2, 0_3]
  • C
    • StreamsGroupHeartbeat: epoch=25, active=[0_2], standby=[0_0, 0_4], warm-up=[0_5]

If C leaves the group now, the group coordinator transforms back the group to a classic group and only use JoinGroup, SyncGroup, and Heartbeat to communicate with the members. The records of the Streams protocol are deleted from the __consumer_offsets topic.

Public Interfaces

This section lists the changes impacting the public interfaces.

KRPC

New Errors

The conditions in which these errors are returned are stated further down.

• STREAMS_INVALID_TOPOLOGY - The supplied topology is invalid. Returned if the client sends a topology that does not fulfill the expected invariants, see below in the sections "Request Validation".

• STREAMS_INVALID_TOPOLOGY_EPOCH  - The client provided a invalid topology epoch with respect to the stream group state, for example, the topology was changed by the epoch not bumped.
• STREAMS_TOPOLOGY_FENCED  - When a client attempts to join with an outdated topology epoch.

StreamsGroupHeartbeat

The StreamsGroupHeartbeat API is the new core API used by streams application to form a group. The API allows members to initialize a topology, advertise their state, and their owned tasks. The group coordinator uses it to assign/revoke tasks to/from members. This API is also used as a liveness check.

Request Schema

The member must set all the (top level) fields with the exception of RackId and InstanceId when it joins for the first time or when an error occurs (e.g. request timed out). Otherwise, it is expected to only fill in the fields which have changed since the last heartbeat.

{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker"],
  "name": "StreamsGroupHeartbeatRequest",
  "validVersions": "0", // "0" in 4.2/4.3 release; bumped to "0-1" in 4.4 release
  "flexibleVersions": "0+",
  "fields": [
    { "name": "GroupId

{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker"],
  "name": "StreamsGroupHeartbeatRequest",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "GroupId", "type": "string", "versions": "0+", "entityType": "groupId",
      "about": "The group identifier." },
    { "name": "MemberId", "type": "string", "versions": "0+",
      "about": "The member id generated by the coordinator. The member id must be kept during the entire lifetime of the member." },
    { "name": "MemberEpoch", "type": "int32", "versions": "0+",
      "about": "The current member epoch; 0 to join the group; -1 to leave the group; -2 to indicate that the static member will rejoin." },
    { "name": "InstanceId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "null if not provided or if it didn't change since the last heartbeat; the instance Id otherwise." },
    { "name": "RackId", "type": "string", "versions": "0+",  "nullableVersions": "0+", "default": "null",
      "about": "null if not provided or if it didn't change since the last heartbeat; the rack ID of consumer otherwise." },
    { "name": "RebalanceTimeoutMs", "type": "int32", "versions": "0+", "default": -1,
      "about": "-1 if it didn't change since the last heartbeat; the maximum time in milliseconds that the coordinator will wait on the member to revoke its partitions otherwise." },

    { "name": "TopologyId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": null,
      "about": "The ID of the topology. Only sent when memberEpoch = 0, must be non-empty. Null otherwise. Used to check if topology corresponds to server-topology. " },

    { "name": "ActiveTasks", "type": "[]TaskIds", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "Currently owned active tasks for this client. Null if unchanged since last heartbeat." },
    { "name": "StandbyTasks", "type": "[]TaskIds", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "Currently owned standby tasks for this client. Null if unchanged since last heartbeat." },
    { "name": "WarmupTasks", "type": "[]TaskIds", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "Currently owned warm-up tasks for this client. Null if unchanged since last heartbeat." },

    { "name": "ProcessId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "Identity of the streams instance that may have multiple consumers. Null if unchanged since last heartbeat." },
    { "name": "UserEndpoint", "type": "Endpoint", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "User-defined endpoint for Interactive Queries. Null if unchanged since last heartbeat." },
    { "name": "ClientTags", "type": "[]KeyValue", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "Used for rack-aware assignment algorithm. Null if unchanged since last heartbeat." },

    { "name": "TaskOffsets", "type": "[]TaskOffset", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "Cumulative offsets for tasks. Only updated when a warm up task has caught up, and according to the task offset interval. Null if unchanged since last heartbeat." },
    { "name": "TaskEndOffsets", "type": "[]TaskOffset", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "Cumulative end-offsets for tasks. Only updated when a warm up task has caught up, and according to the task offset interval. Null if unchanged since last heartbeat." },
    { "name": "ShutdownApplication", "type": "bool", "versions": "0+", "default": false,
      "about": "Whether all Streams clients in the group should shut down. Can only be defined if memberEpoch = -1" }
  ],
  "commonStructs": [
    { "name": "Endpoint", "versions": "0+", "fields": [
      { "name": "Host", "type": "string", "versions": "0+",
        "about": "host of the endpoint" },
      { "name": "Port", "type": "int32", "versions": "0+",
        "about": "port of the endpoint" }
    ]},
    { "name": "KeyValue", "versions": "0+", "fields": [
      { "name": "Key", "type": "string", "versions": "0+",
 "entityType": "groupId",
      "about": "keyThe of the configgroup identifier." },
     
    { "name": "ValueMemberId", "type": "string", "versions": "0+",
        "about": "valueThe member ofID thegenerated config"by }
the streams consumer. The ]},
member ID must be { "name": "TaskOffset", "versions": "0+", "fields": [
      kept during the entire lifetime of the streams consumer process." },
    { "name": "SubtopologyMemberEpoch", "type": "stringint32", "versions": "0+",
        "about": "The sub-topology identifier." },
      { "name": "Partition", "type": "int32", "versions": "0+",
        "about": "The partitioncurrent member epoch; 0 to join the group; -1 to leave the group; -2 to indicate that the static member will rejoin." },
      { "name": "OffsetEndpointInformationEpoch", "type": "int64int32", "versions": "0+",
        "about": "The offset." }
    ] current endpoint epoch of this client, represents the latest endpoint epoch this client received"},
    { "name": "TaskIdsInstanceId", "versionstype": "0+string", "fieldsversions": [
      { "name": "Subtopology""0+", "typenullableVersions": "string0+", "versionsdefault": "0+null",
        "about": "The sub-topology identifiernull if not provided or if it didn't change since the last heartbeat; the instance ID for static membership otherwise." },
      { "name": "PartitionsRackId", "type": "[]int32string", "versions": "0+",
    "nullableVersions": "0+", "default": "null",
      "about": "Thenull partitionsif of the input topics processed by this membernot provided or if it didn't change since the last heartbeat; the rack ID of the member otherwise." },
     ]}
  ]
}

Required ACL

• Read Group

Request Validation

INVALID_REQUEST is returned should the request not obey to the following invariants:

• GroupId must be non-empty.

• Either MemberId is non-empty or MemberEpoch is 0.

• MemberEpoch must be >= -2.

• InstanceId, if not null, must be non-empty.

• RebalanceTimeoutMs must be larger than zero in the first heartbeat request.

• ActiveTasks, StandbyTasks and WarmupTasks have to be disjoint sets

• Each element of ActiveTasks, StandbyTasks and WarmupTasks has to be a valid task ID in the topology initialized for the groupID.

• ActiveTasks, StandbyTasks and WarmupTasks have to be non-null and empty when joining (MemberEpoch is 0)

Request Handling

When the group coordinator handles a StreamsGroupHeartbeat request:

1. Looks up the group. GROUP_ID_NOT_FOUND is returned if the group ID does not exist, or the group ID is associated with a group type that is not streams or classic (the latter will be allowed for migration).

2. Creates the member should the member epoch be zero, otherwise checks whether it exists. If it does not exist, UNKNOWN_MEMBER_ID is returned.

3. Checks whether the member epoch matches the member epoch in its current assignment. FENCED_MEMBER_EPOCH is returned otherwise. The member is also removed from the group.

   • There is an edge case here. When the group coordinator transitions a member to its target epoch, the heartbeat response with the new member epoch may be lost. In this case, the member will retry with the member epoch that he knows about and its request will be rejected with a FENCED_MEMBER_EPOCH. This will be handled as in KIP-848.

4. Updates the members informations if any. The group epoch is incremented if there is any change.

5. Reconcile the member assignments as explained earlier in this document. 

Reponse Schema

The group coordinator will only set the ActiveTasks, StandbyTasks and WarmupTasks fields until the member acknowledges that it has converged to the desired assignment. This is done to ensure that the members converge to the target assignment.

{ "name": "RebalanceTimeoutMs", "type": "int32", "versions": "0+", "default": -1,
      "about": "-1 if it didn't change since the last heartbeat; the maximum time in milliseconds that the coordinator will wait on the member to revoke its tasks otherwise." },

    { "name": "Topology", "type": "Topology", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "The topology metadata of the streams application. Used to initialize the topology of the group and to check if the topology corresponds to the topology initialized for the group. Only sent when memberEpoch = 0, must be non-empty. Null otherwise.",
      "fields": [
        { "name": "Epoch", "type": "int32", "versions": "0+",
          "about": "The epoch of the topology. Used to check if the topology corresponds to the topology initialized on the brokers." },
        { "name": "Subtopologies", "type": "[]Subtopology", "versions": "0+",
          "about": "The sub-topologies of the streams application.",
          "fields": [
            { "name": "SubtopologyId", "type": "string", "versions": "0+",
              "about": "String to uniquely identify the subtopology. Deterministically generated from the topology" },
            { "name": "SourceTopics", "type": "[]string", "versions": "0+",
              "about": "The topics the topology reads from." },
            { "name": "SourceTopicRegex", "type": "[]string", "versions": "0+",
              "about": "The regular expressions identifying topics the subtopology reads from." },
            { "name": "StateChangelogTopics", "type": "[]TopicInfo", "versions": "0+",
              "about": "The set of state changelog topics associated with this subtopology. Created automatically." },
            { "name": "RepartitionSinkTopics", "type": "[]string", "versions": "0+",
              "about": "The repartition topics the subtopology writes to." },
            { "name": "RepartitionSourceTopics", "type": "[]TopicInfo", "versions": "0+",
              "about": "The set of source topics that are internally created repartition topics. Created automatically." },
            { "name": "CopartitionGroups", "type": "[]CopartitionGroup", "versions": "0+",
              "about": "A subset of source topics that must be copartitioned.",
              "fields": [
                { "name": "SourceTopics", "type": "[]int16", "versions": "0+",
                  "about": "The topics the topology reads from. Index into the array on the subtopology level." },
                { "name": "SourceTopicRegex", "type": "[]int16", "versions": "0+",
                  "about": "Regular expressions identifying topics the subtopology reads from. Index into the array on the subtopology level." },
                { "name": "RepartitionSourceTopics", "type": "[]int16", "versions": "0+",
                  "about": "The set of source topics that are internally created repartition topics. Index into the array on the subtopology level." }
              ]}
          ]}
      ]
    },

    { "name": "ActiveTasks", "type": "[]TaskIds", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "Currently owned active tasks for this client. Null if unchanged since last heartbeat." },
    { "name": "StandbyTasks", "type": "[]TaskIds", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "Currently owned standby tasks for this client. Null if unchanged since last heartbeat." },
    { "name": "WarmupTasks", "type": "[]TaskIds", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "Currently owned warm-up tasks for this client. Null if unchanged since last heartbeat." },

    { "name": "ProcessId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "Identity of the streams instance that may have multiple consumers. Null if unchanged since last heartbeat." },
    { "name": "UserEndpoint", "type": "Endpoint", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "User-defined endpoint for Interactive Queries. Null if unchanged since last heartbeat or if not defined on the client." },
    { "name": "ClientTags", "type": "[]KeyValue", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "Used for rack-aware assignment algorithm. Null if unchanged since last heartbeat." },

    { "name": "TaskOffsets", "type": "[]TaskOffset", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "Cumulative changelog offsets for tasks. Only updated when a warm-up task has caught up, and according to the task offset interval. Null if unchanged since last heartbeat." },
    { "name": "TaskEndOffsets", "type": "[]TaskOffset", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "Cumulative changelog end-offsets for tasks. Only updated when a warm-up task has caught up, and according to the task offset interval. Null if unchanged since last heartbeat." },
    { "name": "ShutdownApplication", "type": "bool", "versions": "0+", "default": false,
      "about": "Whether all Streams clients in the group should shut down." }
  ],

  "commonStructs": [
    { "name": "KeyValue", "versions": "0+", "fields": [
      { "name": "Key", "type": "string", "versions": "0+",
        "about": "key of the config" },
      { "name": "Value", "type": "string", "versions": "0+",
        "about": "value of the config" }
    ]},
    { "name": "TopicInfo", "versions": "0+", "fields": [
      { "name": "Name", "type": "string", "versions": "0+",
        "about": "The name of the topic." },
      { "name": "Partitions", "type": "int32", "versions": "0+",
        "about": "The number of partitions in the topic. Can be 0 if no specific number of partitions is enforced. Always 0 for changelog topics." },
      { "name": "ReplicationFactor", "type": "int16", "versions": "0+",
        "about": "The replication factor of the topic. Can be 0 if the default replication factor should be used." },
      { "name": "TopicConfigs", "type": "[]KeyValue", "versions": "0+",
        "about": "Topic-level configurations as key-value pairs."
      }
    ]},
    { "name": "Endpoint", "versions": "0+", "fields": [
      { "name": "Host", "type": "string", "versions": "0+",
        "about": "host of the endpoint" },
      { "name": "Port", "type": "uint16", "versions": "0+",
        "about": "port of the endpoint" }
    ]},
    { "name": "TaskOffset", "versions": "0+", "fields": [
      { "name": "SubtopologyId", "type": "string", "versions": "0+",
        "about": "The subtopology identifier." },
      { "name": "Partition", "type": "int32", "versions": "0+",
        "about": "The partition." },
      { "name": "Offset", "type": "int64", "versions": "0+",
        "about": "The offset." }
    ]},
    { "name": "TaskIds", "versions": "0+", "fields": [
      { "name": "SubtopologyId", "type": "string", "versions": "0+",
        "about": "The subtopology identifier." },
      { "name": "Partitions", "type": "[]int32", "versions": "0+",
        "about": "The partitions of the input topics processed by this member." }
    ]}
  ]
}

Required ACL

• READ on group

• CREATE on cluster resource, or CREATE on all topics in StateChangelogTopics and RepartitionSourceTopics

  • Note that this ACLs are only required if the group coordinator should create the internal topics implicitly. If the internal topics are created explicitly, this ACL is not needed for the Streams group heartbeat.

• DESCRIBE on all topics included in the message

Request Validation

INVALID_REQUEST is returned should the request not obey to the following invariants:

• GroupId must be non-empty.

• MemberId must be non-empty.

• MemberEpoch must be >= -2.

• InstanceId, if not null, must be non-empty.

• RebalanceTimeoutMs must be larger than zero in the first heartbeat request.

• ActiveTasks, StandbyTasks and WarmupTasks have to be disjoint sets

• ActiveTasks, StandbyTasks and WarmupTasks have to be non-null and empty when joining (member epoch is 0)

• If MemberEpoch  is 0, Topology should be non-null. Otherwise, it should be null.

STREAMS_INVALID_TOPOLOGY is returned when the request contains a new topology and should the topology not obey the following invariants:

• A StateChangelogTopic topics must not have a defined partition number.

• A RepartitionSourceTopic cannot be in SourceTopics or StateChangelogTopics of any subtopology.

• A StateChangelogTopic cannot be in SourceTopics or RepartitionSinkTopic or RepartitionSourceTopics of any subtopology.

• A RepartitionSourceTopic of one subtopology must be a RepartitionSinkTopic of at least one other subtopology.

• All indices in CopartitionGroups must be valid indices in the corresponding topic arrays.

Request Handling

When the group coordinator handles a StreamsGroupHeartbeat request:

1. Performs request validation.

2. If the member joins the group (i.e. member epoch is 0):

   1. Look up or create the group.
      1. If the group is created the topology epoch of the group is set to the topology epoch sent by the member.
   2. GROUP_ID_NOT_FOUND is returned if the group ID is associated with a group type that is not streams or classic (the latter will be allowed for migration).
   3. Compare the topology epoch of the request ER to the topology epoch of the group EG
      1. If ER=EG, check if topology metadata in the request is equal to the topology metadata of the group, otherwise fail with STREAMS_INVALID_TOPOLOGY_EPOCH
      2. If ER<EG, fail with STREAMS_TOPOLOGY_FENCED 
      3. If ER>EG+1, fail with STREAMS_INVALID_TOPOLOGY_EPOCH
      4. If ER=EG+1, update the topology by writing the new topology record to the offset topic
   4. Creates the member.
3. If the member is already part of the group (i.e. member epoch is greater than 0):
   1. Looks up the group.
   2. GROUP_ID_NOT_FOUND is returned if the group ID does not exist anymore.
   3. If the member does not exist, returns UNKNOWN_MEMBER_ID
   4. Checks whether the member epoch matches the member epoch in its current assignment. FENCED_MEMBER_EPOCH is returned otherwise. The member is also removed from the group.
      • There is an edge case here. When the group coordinator transitions a member to its target epoch, the heartbeat response with the new member epoch may be lost. In this case, the member will retry with the member epoch that it knows about and its request will be rejected with a FENCED_MEMBER_EPOCH. This will be handled as in KIP-848.
   5. If the topology epoch in the request is less than the topology epoch of the group, set state STALE_TOPOLOGY  in the response.
4. Updates information of the member if needed. The group epoch is incremented if there is any change.
5. If the topology or topic metadata changed, detect any topology / topic mismatches as described earlier in this document.
6. Reconcile the member assignments as explained earlier in this document.
7. When a member assignment gets changed it will bump the StreamsGroup EndpointEpoch  and bump the corresponding member EndpointInformationEpoch in the response.  When all members EndpointInformationEpoch values converge on the same value as the group-wide EndpointEpoch updated IQ information will be included in the member heartbeat response.  

Response Schema

The group coordinator will only set the ActiveTasks, StandbyTasks and WarmupTasks fields until the member acknowledges that it has converged to the desired assignment. This is done to ensure that the members converge to the target assignment.

{
  "apiKey": TBD,
  "type": "response",
  "name": "StreamsGroupHeartbeatResponse",
  "validVersions": "0",  // "0" in 4.2/4.3 release; bumped to "0-1" in 4.4 release 
  "flexibleVersions": "0+",
  // Supported errors:
  // - GROUP_AUTHORIZATION_FAILED (version 0+)
  // - GROUP_ID_NOT_FOUND (version 0+)
  // - NOT_COORDINATOR (version 0+)
  // - COORDINATOR_NOT_AVAILABLE (version 0+)
  // - COORDINATOR_LOAD_IN_PROGRESS (version 0+)
  // - INVALID_REQUEST (version 0+)
  // - UNKNOWN_MEMBER_ID (version 0+)
  // - FENCED_MEMBER_EPOCH (version 0+)
  // - UNRELEASED_INSTANCE_ID (version 0+)
  // - GROUP_MAX_SIZE_REACHED (version 0+)
  // - TOPIC_AUTHORIZATION_FAILED (version 0+) 
  // - CLUSTER_AUTHORIZATION_FAILED (version 0+)
  // - STREAMS_INVALID_TOPOLOGY (version 0+)
  // - STREAMS_INVALID_TOPOLOGY_EPOCH (version 0+)
  // - STREAMS_TOPOLOGY_FENCED (version 0+)
  "fields": [
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
      "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 top-level error code, or 0 if there was no error" },
    { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "The top-level error message, or null if there was no error." },      
    { "name": "MemberId", "type": "string", "versions": "0+",
      "about": "The member id is always generated by the streams consumer."},     
    { "name": "MemberEpoch", "type": "int32", "versions": "0+",
      "about": "The member epoch." },

    { "name": "HeartbeatIntervalMs", "type": "int32", "versions": "0+",
      "about": "The heartbeat interval in milliseconds." },
    { "name": "AcceptableRecoveryLag", "type": "int32", "versions": "0+",
      "about": "The maximal lag a warm-up task can have to be considered caught-up." },  // incorrectly implemented as `int32` in 4.2 release; updated to `int64` and request/response version bump to "1" in 4.4 release.
    { "name": "TaskOffsetIntervalMs", "type": "int32", "versions": "0+",
      "about": "The interval in which the task changelog offsets on a client are updated on the broker. The offsets are sent with the next heartbeat after this time has passed." },

    { "name": "Status", "type":  "[]Status", "versions":  "0+",  "nullableVersions": "0+", "default": "null",
      "about": "Indicate zero or more status for the group.  Null if unchanged since last heartbeat." },

    // The streams app knows which partitions to fetch from given this information
    { "name": "ActiveTasks", "type": "[]TaskIds", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "Assigned active tasks for this client. Null if unchanged since last heartbeat." },
    { "name": "StandbyTasks", "type": "[]TaskIds", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "Assigned standby tasks for this client. Null if unchanged since last heartbeat." },
    { "name": "WarmupTasks", "type": "[]TaskIds", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "Assigned warm-up tasks for this client. Null if unchanged since last heartbeat." },

    // IQ-related information
    { "name": "EndpointInformationEpoch", "type": "int32", "versions": "0+",
      "about": "The endpoint epoch set in the response"}, 
    { "name": "PartitionsByUserEndpoint", "type": "[]EndpointToPartitions", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "Global assignment information used for IQ. Null if unchanged since last heartbeat." ,
      "fields": [
        { "name": "UserEndpoint", "type": "Endpoint", "versions": "0+",
          "about": "User-defined endpoint to connect to the node" },
        { "name": "ActivePartitions", "type": "[]TopicPartition", "versions": "0+",
          "about": "All active partitions available on the node" },
        { "name": "StandbyPartitions", "type": "[]TopicPartition", "versions": "0+",
          "about": "All standby paritions available on the node"}
      ]
    }
  ],
  "commonStructs": [
    { "name": "Status", "versions": "0+", "fields": [
      // Possible status codes
      //  0 - STALE_TOPOLOGY                 - The topology epoch supplied is lower than the topology epoch for this streams group.
      //  1 - MISSING_SOURCE_TOPICS          - One or more source topics are missing or a source topic regex resolves to zero topics.
      //                                       Missing topics are indicated in the StatusDetail.
      //  2 - INCORRECTLY_PARTITIONED_TOPICS - One or more topics are incorrectly partitioned, that is, they are not copartitioned despite being 
      //                                       part of a copartition group, or the number of partitions in a changelog topic does not correspond
      //                                       to the maximal number of source topic partition for that subtopology.
      //                                       Incorrectly partitioned topics are indicated in the StatusDetail.
      //  3 - MISSING_INTERNAL_TOPICS        - One or more internal topics are missing. 
      //                                       Missing topics are indicated in the StatusDetail. 
      //                                       The group coordinator will attempt to create all missing internal topics, if any errors occur during
      //                                       topic creation, this will be indicated in StatusDetail.
      //  4 - SHUTDOWN_APPLICATION           - A client requested the shutdown of the whole application.
      //  5 - ASSIGNMENT_DELAYED             - No assignment was provided because assignment computation was delayed.
      //  6 - MISSING_CLIENT_TAGS            - (Response v1+ only) The GC is configured to use `client.tag.` for rack-aware standby task assignment, but the client does not provide this tag
      { "name": "StatusCode", "type": "int8

{
  "apiKey": TBD,
  "type": "response",
  "name": "StreamsGroupHeartbeatResponse",
  "validVersions": "0",
  "flexibleVersions": "0+",
  // Supported errors:
  // - GROUP_AUTHORIZATION_FAILED (version 0+)
  // - NOT_COORDINATOR (version 0+)
  // - COORDINATOR_NOT_AVAILABLE (version 0+)
  // - COORDINATOR_LOAD_IN_PROGRESS (version 0+)
  // - INVALID_REQUEST (version 0+)
  // - UNKNOWN_MEMBER_ID (version 0+)
  // - FENCED_MEMBER_EPOCH (version 0+)
  // - UNRELEASED_INSTANCE_ID (version 0+)
  // - GROUP_MAX_SIZE_REACHED (version 0+)
  "fields": [
    // Same as consumer group heart beat
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
      "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 top-level error code, or 0 if there was no error" },
    { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "The top-level error message, or null if there was no error." },
    { "name": "MemberId", "type": "string", "versions": "0+",
 "nullableVersions": "0+", "default": "null",
      "about": "The member id generated by the coordinator. Only provided when the member joins with MemberEpoch == 0.A code to indicate that a particular status is active for the group membership" },
      { "name": "MemberEpochStatusDetail", "type": "int32string", "versions": "0+",
        "about": "The member epoch." A string representation of the status." }
    ]},
    { "name": "HeartbeatIntervalMsTopicPartition", "type": "int32", "versions": "0+",
      "aboutfields": "The[
 heartbeat interval in milliseconds." },
    { "name": "AcceptableRecoveryLagTopic", "type": "int32string", "versions": "0+",
        "about": "The maximal lag a warm-up task can have to be considered caught-up.topic name" },
      { "name": "TaskOffsetIntervalMsPartitions", "type": "[]int32", "versions": "0+",
        "about": "partitions"The interval}
 in which the changelog]},
 task offsets on a client are updated on the broker. The offsets are sent with the next heartbeat after this time has passed." },

    // Topology updating
{ "name": "TaskIds", "versions": "0+", "fields": [
      { "name": "ShouldInitializeTopologySubtopologyId", "type":  "boolstring", "versions":  "0+",
 "default": false,
      "about": "trueThe if this streams application should initialize the topology on the brokersubtopology identifier." },
      { "name": "GroupTopologyIdPartitions", "type": "string[]int32", "versions": "0+", "nullableVersions": "0+", "default": null,

         "about": "The current IDpartitions of the topologyinput fortopics theprocessed group.by Nullthis if unchanged since last heartbeat, empty if group is still initializing." member." }
    ]},

    { "name": "StatusEndpoint", "typeversions":  "[]Status0+", "versionsfields": [
    "0+  { "name": "Host",  "nullableVersionstype": "0+string", "defaultversions": "null0+",
  
      "about": "Indicatehost of zerothe or more status for the group.  Null if unchanged since last heartbeat." },

    // The streams app knows which partitions to fetch from given this information
    { "name": "ActiveTasks", endpoint" },
      { "name": "Port", "type": "uint16", "versions": "0+",
        "about": "port of the endpoint" }
    ]}
  ]
}

StreamsGroupDescribe API

Request Schema

{
  "apiKey": TBD,
  "type": "[]TaskIdsrequest",
  "versionslisteners": ["0+broker"],
  "nullableVersionsname": "0+StreamsGroupDescribeRequest",
  "defaultvalidVersions": "null0",
      "aboutflexibleVersions": "0+"Assigned,
 active tasks for this client. Null if unchanged since last heartbeat." }, "fields": [
    { "name": "StandbyTasksGroupIds", "type": "[]TaskIdsstring", "versions": "0+", "nullableVersionsentityType": "0+", "default": "nullgroupId",
      "about": "AssignedThe standbyids tasksof forthe thisgroups client. Null if unchanged since last heartbeat.to describe" },
    { "name": "WarmupTasksIncludeAuthorizedOperations", "type": "[]TaskIdsbool", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "AssignedWhether warm-upto tasksinclude for this client. Null if unchanged since last heartbeatauthorized operations." },
  ]
}

Required ACL

• DESCRIBE on the group resource

• DESCRIBE on all topics used in the group's topology

Request Validation

INVALID_REQUEST is returned should the request not obey the schema definition.

INVALID_GROUP_ID is returned should the group ID be empty.

Request Handling

When the group coordinator handle a StreamsGroupDescribeRequest request:

• Checks whether the group ids exists. If it does not, GROUP_ID_NOT_FOUND is returned. Similarly, it is returned if the group ID is associated with a group type that is not streams .

• Looks up the groups and returns the response.

Response Schema

{
  "apiKey": TBD,
 
    // IQ-related information
    { "name": "PartitionsByUserEndpoint", "type": "[]EndpointToPartitionsresponse",
  "versionsname": "0+StreamsGroupDescribeResponse",
  "nullableVersionsvalidVersions": "0+",
  "defaultflexibleVersions": "null0+",
  // Supported   "about"errors:
 "Global assignment// information used for IQ. Null if unchanged since last heartbeat." ,
      "fields": [
        { "name": "UserEndpoint", "type": "Endpoint", "versions": "0+",
          "about": "User-defined endpoint to connect to the node" },
    - GROUP_AUTHORIZATION_FAILED (version 0+)
  // - NOT_COORDINATOR (version 0+)
  // - COORDINATOR_NOT_AVAILABLE (version 0+)
  // - COORDINATOR_LOAD_IN_PROGRESS (version 0+)
  // - INVALID_REQUEST (version 0+)
  // - INVALID_GROUP_ID (version 0+)
  // - GROUP_ID_NOT_FOUND (version 0+)
  "fields": [
    { "name": "PartitionsThrottleTimeMs", "type": "[]TopicPartitionint32", "versions": "0+",
          "about": "All partitions available on the node" }
      ]
    }
  ],
  "commonStructs": [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": "StatusGroups", "versionstype": "0+[]DescribedGroup", "fieldsversions": ["0+",
      // Possible status codes"about": "Each described group.",
      //  0 - GROUP_INITIALIZING"fields": [
        { - The group is not initialized.
      //  1 - INCONSISTENT_TOPOLOGY   - The topology ID supplied is inconsistent with the topology for this streams group.
      //  2 - MISSING_SOURCE_TOPICS   - One or more source topics do not exist. Missing topics are indicated in the StatusDetail."name": "ErrorCode", "type": "int16", "versions": "0+",
          "about": "The describe error, or 0 if there was no error." },
        { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
      //   3 - MISSING_INTERNAL_TOPICS - One"about": "The top-level error message, or morenull internalif topicsthere dowas notno existerror. Missing topics are indicated in the StatusDetail.
      //  4 - SHUTDOWN_APPLICATION    - A client requested the shutdown of the whole application.
" },
        { "name": "GroupId", "type": "string", "versions": "0+", "entityType": "groupId",
          "about": "The group ID string." },
        { "name": "StatusCodeGroupState", "type": "int8string", "versions": "0+",
          "about": "AThe codegroup tostate indicatestring, that a particular status is active for or the groupempty membershipstring." },
        { "name": "StatusDetailGroupEpoch", "type": "stringint32", "versions": "0+",
          "about": "AThe string representation of the statusgroup epoch." },
    ]},
    { "name": "TopicPartitionAssignmentEpoch", "type": "int32", "versions": "0+",
          "fieldsabout": [
"The assignment epoch." },

        { "name":  "TopicTopology", "type": "stringTopology", "versions": "0+", "nullableVersions": "0+", "default": "null",
          "about": "The topology metadata currently initialized for the streams application. Can be null in case of a describe error.",
          "aboutfields": "topic name" },
[
            { "name": "PartitionsEpoch", "type": "[]int32", "versions": "0+",
              "about": "partitionsThe epoch of the currently initialized topology for this group." },
        ]},
    { "name": "TaskIdsSubtopologies", "versionstype": "0+[]Subtopology", "fields": [
      { "name"versions": "Subtopology0+", "typenullableVersions": "string0+", "versionsdefault": "0+null",
              "about": "The subtopologysubtopologies identifier." },
      { "name": "Partitions", "type": "[]int32", "versions": "0+",
        "about": "The partitions of the input topics processed by this member." }
    ]}
    { "name": "Endpoint", "versions": "0+", "fields": [
of the streams application. This contains the configured subtopologies, where the number of partitions are set and any regular expressions are resolved to actual topics. Null if the group is uninitialized, source topics are missing or incorrectly partitioned.",
         { "name": "Host", "type": "string", "versionsfields": "0+",[
        "about": "host of the endpoint" },
      { "name": "PortSubtopologyId", "type": "int32string", "versions": "0+",
        "about": "port of the endpoint" }
    ]}
  ]
}

StreamsGroupDescribe API

Request Schema

{
  "apiKey"about": TBD,
"String to "type": "request",
  "listeners": ["broker"]uniquely identify the subtopology." },
  "name": "StreamsGroupDescribeRequest",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "GroupIdsSourceTopics", "type": "[]string", "versions": "0+", "entityType": "groupId",

                  "about": "The idstopics of the groupssubtopology toreads describefrom." },
                { "name": "IncludeAuthorizedOperationsRepartitionSinkTopics", "type": "bool[]string", "versions": "0+",
                  "about": "Whether to include authorized operationsThe repartition topics the subtopology writes to." }
  ]
}

Required ACL

• Read Group

Request Validation

INVALID_REQUEST is returned should the request not obey the schema definition.

INVALID_GROUP_ID is returned should the group ID be empty.

Request Handling

When the group coordinator handle a StreamsGroupDescribeRequest request:

• Checks whether the group ids exists. If it does not, GROUP_ID_NOT_FOUND is returned. Similarly, it is returned if the group ID is associated with a group type that is not streams .

• Looks up the groups and returns the response.

Response Schema

{
  "apiKey": TBD,
  "type": "response",
  "name": "StreamsGroupDescribeResponse",
  "validVersions": "0",
  "flexibleVersions,
                { "name": "StateChangelogTopics", "type": "[]TopicInfo", "versions": "0+",
                  "about": "The set of state changelog topics associated with this subtopology. Created automatically." },
                { "name": "RepartitionSourceTopics", "type": "[]TopicInfo", "versions": "0+",
  // Supported errors:
        // - GROUP_AUTHORIZATION_FAILED (version 0+)
  // - NOT_COORDINATOR (version 0+)
  // - COORDINATOR_NOT_AVAILABLE (version 0+)
  // - COORDINATOR_LOAD_IN_PROGRESS (version 0+)
  // - INVALID_REQUEST (version 0+)
  // - INVALID_GROUP_ID (version 0+)
  // - GROUP_ID_NOT_FOUND (version 0+)
  "fields": [
 "about": "The set of source topics that are internally created repartition topics. Created automatically." }
              ]}
          ]},
        { "name": "ThrottleTimeMsMembers", "type": "int32[]Member", "versions": "0+",
          "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." },
    members.",
          "fields": [
            { "name": "GroupsMemberId", "type": "[]DescribedGroupstring", "versions": "0+",
              "about": "EachThe describedmember groupID." },
      "fields": [
        { "name": "ErrorCodeMemberEpoch", "type": "int16int32", "versions": "0+",
              "about": "The member describeepoch." error},
 or 0
 if there was no error." },
        { "name": "ErrorMessageInstanceId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
              "about": "The top-levelmember errorinstance message,ID orfor null if there was no errorstatic membership." },
            { "name": "GroupIdRackId", "type": "string", "versions": "0+", "entityTypenullableVersions": "0+", "default": "groupIdnull",
              "about": "The grouprack ID string." },

            { "name": "GroupStateClientId", "type": "string", "versions": "0+",
              "about": "The groupclient stateID." string}, or
 the empty string." },
        { "name": "GroupEpochClientHost", "type": "int32string", "versions": "0+",
              "about": "The groupclient epochhost." },

            { "name": "AssignmentEpochTopologyEpoch", "type": "int32", "versions": "0+",
              "about": "The assignment epochepoch of the topology on the client." },

            { "name": "TopologyIdProcessId", "type": "string", "versions": "0+",
              "about": "The IDIdentity of the currentlystreams instance initializedthat topologymay forhave thismultiple groupclients. " }, 
        
            { "name": "TopologyUserEndpoint", "type": "[]SubtopologyEndpoint", "versions": "0+", "nullableVersions": "0+", "default": "null",
              "about": "The subUser-topologiesdefined ofendpoint thefor streamsInteractive applicationQueries.",
          "fields": [
            Null if not defined for this client." },
            { "name": "SubtopologyClientTags", "type": "string[]KeyValue", "versions": "0+",
              "about": "StringUsed tofor uniquelyrack-aware identifyassignment the subtopologyalgorithm." },
            { "name": "SourceTopicsTaskOffsets", "type": "[]stringTaskOffset", "versions": "0+",
              "about": "TheCumulative topicschangelog theoffsets topologyfor reads fromtasks." },
            { "name": "SourceTopicRegexTaskEndOffsets", "type": "string[]TaskOffset", "versions": "0+", "nullableVersions": "0+", "default": "null",
              
              "about": "TheCumulative regularchangelog expressionsend identifyingoffsets topics the topology reads from. null if not provided.for tasks." },

            { "name": "RepartitionSinkTopicsAssignment", "type": "[]stringAssignment", "versions": "0+",
              "about": "The topics the topology writes tocurrent assignment." },
            { "name": "StateChangelogTopicsTargetAssignment", "type": "[]TopicInfoAssignment", "versions": "0+",
              "about": "The set of state changelog topics associated with this subtopology. Created automatically.target assignment." },
            { "name": "RepartitionSourceTopicsIsClassic", "type": "[]TopicInfobool", "versions": "0+",
              "about": "The set of source topics that are internally created repartition topics. Created automatically." }
          ]
        ": "True for classic members that have not been upgraded yet." }
        ]},
        { "name": "MembersAuthorizedOperations", "type": "[]Memberint32", "versions": "0+", "default": "-2147483648",
          "about": "The members.",
          "fields": [
32-bit bitfield to represent authorized operations for this group." }
      ]
    }
  ],
 { "namecommonStructs": "MemberId",[
    { "typename": "stringEndpoint", "versions": "0+",
              "about"fields": "The member ID." },
      [
      { "name": "MemberEpochHost", "type": "int32string", "versions": "0+",
              "about": "Thehost of memberthe epoch.endpoint" },
            { "name": "InstanceIdPort", "type": "stringuint16", "versions": "0+", "nullableVersions": "0+", "default": "null",
              "about": "The member instance ID." },
     port of the endpoint" }
    ]},
     { "name": "RackIdTaskOffset", "typeversions": "string0+", "fields": [
 "versions     { "name": "0+SubtopologyId", "nullableVersionstype": "0+string", "defaultversions": "null0+",
              "about": "The racksubtopology IDidentifier." },

            { "name": "ClientIdPartition", "type": "stringint32", "versions": "0+",
              "about": "The client IDpartition." },
            { "name": "ClientHostOffset", "type": "stringint64", "versions": "0+",
              "about": "The client hostoffset." }
    ]},

          { "name": "TopicPartitions", "versions": "0+", "fields": [
      { "name": "TopologyIdTopicId", "type": "stringuuid", "versions": "0+",
              "about": "The topic ID of the topology on the client." },

            { "name": "ProcessIdTopicName", "type": "string", "versions": "0+",
      "entityType": "topicName",
        "about": "Identity of the streams instance that may have multiple clients.  "The topic name." },
            { "name": "ClientTagsPartitions", "type": "[]KeyValueint32", "versions": "0+",
              "about": "The partitions."Used for}
 rack-aware assignment algorithm." ]},
    { "name": "Assignment", "versions": "0+", "fields": [
      { "name": "TaskOffsetsActiveTasks", "type": "[]TaskOffsetTaskIds", "versions": "0+",
              "about": "CumulativeActive offsetstasks for this tasksclient." },
            { "name": "TaskEndOffsetsStandbyTasks", "type": "[]TaskOffsetTaskIds", "versions": "0+",
              "about": "CumulativeStandby endtasks offsetsfor forthis tasksclient." },

            { "name": "AssignmentWarmupTasks", "type": "Assignment[]TaskIds", "versions": "0+",
              "about": "The current assignment.Warm-up tasks for this client. " },
    ]},
        { "name": "TargetAssignmentTaskIds", "typeversions": "Assignment0+", "versionsfields": "0+",[
      { "name": "SubtopologyId",      "about"type": "string", "versions": "The target assignment." }
"0+",
        "about": "The subtopology identifier." ]},
        { "name": "AuthorizedOperationsPartitions", "type": "[]int32", "versions": "0+", "default": "-2147483648",
          "about": "32-bit bitfield to represent authorized operations forThe partitions of the input topics processed by this groupmember." }
      ]
    }
  ],
  "commonStructs": [
    { "name": "TaskOffsetKeyValue", "versions": "0+", "fields": [
      { "name": "SubtopologyKey", "type": "string", "versions": "0+",
        "about": "Thekey of subtopologythe identifier.config" },
      { "name": "PartitionValue", "type": "int32string", "versions": "0+",
        "about": "The partition." },value of the config" }
    ]},
    { "name": "TopicInfo", "versions": "0+", "fields": [
      { "name": "OffsetName", "type": "int64string", "versions": "0+",
        "about": "The offset name of the topic." },
    ]},
    { "name": "TopicPartitionsPartitions", "versionstype": "0+int32", "fieldsversions": ["0+",
      {  "nameabout": "TopicId", "type": "uuid", "versions": "0+",
        "about": "The topic IDThe number of partitions in the topic. Can be 0 if no specific number of partitions is enforced. Always 0 for changelog topics." },
      { "name": "TopicNameReplicationFactor", "type": "stringint16", "versions": "0+", "entityType": "topicName",
        "about": "The replication factor of the topic name. Can be 0 if the default replication factor should be used." },
         { "name": "PartitionsTopicConfigs", "type": "[]int32KeyValue", "versions": "0+",
        "about": "The partitionsTopic-level configurations as key-value pairs."
  }
    ]},
    { "name": "Assignment", "versions": "0+", "fields": [
      { "name": "ActiveTasks", "type": "[]TaskIds", "versions": "0+",
        "about": "Active tasks for this client." },
      { "name": "StandbyTasks", "type": "[]TaskIds", "versions": "0+",
        "about": "Standby tasks for this client." },
      { "name": "WarmupTasks", "type": "[]TaskIds", "versions": "0+",
        "about": "Warm-up tasks for this client. " }
    ]},
    { "name": "TaskIds", "versions": "0+", "fields": [
      { "name": "Subtopology", "type": "string", "versions": "0+",
        "about": "The subtopology identifier." },
      { "name": "Partitions", "type": "[]int32", "versions": "0+",
        "about": "The partitions of the input topics processed by this member." }
    ]},
    { "name": "KeyValue", "versions": "0+", "fields": [
  ]}
  ]
}

Response Handling

Nothing particular.

Records

This section describes the new record types required for the new protocol. Similar to KIP-848, we define new record types that persist all data related to groups, their members and their assignments in the __consumer_offsets compacted topic. Each record has a dedicated key type that uses the version to differentiate the records from the records introduced in KIP-848, and those used to persist the committed offsets.

Compared to consumer groups, we introduce one extra record type, that stores information about the topology of the group.

The size of the persisted member metadata in the __consumer_offset topic will have to respect the max message size. But those records are persisted separately for each member, so it should be less of a concern. The same problem affects plain consumers in KIP-848 for very large consumer groups. Eventually, we may implement a way to split the records into multiple messages, and there are currently plans to add transaction support to the group coordinator to make sure the records can be updated atomically. For the first version of the protocol, we will limit those records by the max message size.

Group Metadata

The assignment-related metadata for a group with N members is stored with N+1 records. One metadata record per member and one metadata record for the group.

• The member metadata record, keyed by the group ID and the member ID, stores all relevant metadata that is sent by a member to the coordinator in a heartbeat request. A notable exception are task changelog offsets and task changelog end offsets, which are not persisted, as they are constantly changing. 
• The group metadata, keyed by the group ID, contains
  • The group epoch
  • A hash of the topology metadata
  • The epoch of the last validated topology
  • The last configuration used by the assignor

The records evolve in the following way:

• When the first member joins the group, the group metadata and the member metadata records for the member are first created.
• Every time a new member joins the group, another member metadata record is added
• Whenever a member updates its member metadata through the heartbeat, a new version of the member metadata record is appended.
• When a member leaves the group, a tombstone is written for the member metadata key.

When the group lost all members (or never had any)  and offsets.retention.minutes expires, the group metadata tombstone is written. This can only happen in group state EMPTY.

StreamsGroupMemberMetadataKey

{
  "type": "data",
  "name": "StreamsGroupMemberMetadataKey",
  "validVersions": TBD,
  "flexibleVersions": "none",
  "fields": [
    { "name": "KeyGroupId", "type": "string", "versions": "0+"TBD,
        "about": "keyThe of the configgroup id." },
      { "name": "ValueMemberId", "type": "string", "versions": "0+"TBD,
        "about": "valueThe of the configmember id." }
    ]
}

StreamsGroupMemberMetadataValue

{
  "type},
    { "name": "TopicInfodata",
  "versionsname": "0+StreamsGroupMemberMetadataValue", "fields": [
      { "name"validVersions": "Name0",
 "type": "string", "versionsflexibleVersions": "0+",
        "aboutfields": "The name of the topic." },
 [
     { "name": "PartitionsInstanceId", "typeversions": "int320+", "versionsnullableVersions": "0+",
 "type": "string",
      "about": "The number of partitions in the topic. Can be 0 if no specific number of partitions is enforced. Always 0 for changelog topics "The (optional) instance ID for static membership." },
      { "name": "TopicConfigs", "type": "[]KeyValue"RackId", "versions": "0+", "nullableVersions": "0+", "defaulttype": "nullstring",
        "about": "Topic-level configurations as key-value pairs."The (optional) rack id." },
    { "name": "ClientId", "versions": "0+",  }"type": "string",
    ]}
  ]
}

Response Handling

Nothing particular.

Other RPCs

Streams group IDs can be used in DeleteGroups , DescribeGroups , 

Records

This section describes the new record types required for the new protocol. Similar to KIP-848, we define new record types that persist all data related to groups, their members and their assignments in the __consumer_offsets compacted topic. Each record has a dedicated key type that uses the version to differentiate the records from the records introduced in KIP-848, and those used to persist the committed offsets.

Compared to consumer groups, we introduce one extra record type, that stores information about the topology of the group.

The size of the persisted member metadata in the __consumer_offset topic will have to respect the max message size. But those records are persisted separately for each member, so it should be less of a concern. The same problem affects plain consumers in KIP-848 for very large consumer groups. Eventually, we may implement a way to split the records into multiple messages, and there are currently plans to add transaction support to the group coordinator to make sure the records can be updated atomically. For the first version of the protocol, we will limit those records by the max message size.

Group Metadata

The assignment-related metadata for a group with N members is stored with N+1 records. One metadata record per member and one metadata record for the group.

• The member metadata record, keyed by the group ID and the member ID, stores all relevant metadata that is sent by a member to the coordinator in a heartbeat request. A notable exception are task offsets and task lag offsets, which are not persisted, as they are constantly changing. 
• The group metadata, keyed by the group ID, contains the group epoch.

The records evolve in the following way:

• When the first member joins the group, the group metadata and the member metadata records for the member are first created.
• Every time a new member joins the group, another member metadata record is added
• When a previously non-existing group is initialized via a StreamsGroupInitialize  RPC, the group member metadata is created.
• Whenever a member updates its member metadata through the heartbeat, a new version of the member metadata record is appended.
• When a member leaves the group, a tombstone is written for the member metadata key.

When the group lost all members (or never had any)  and offsets.retention.minutes expires, the group metadata tombstone is written. This can happen in both states EMPTY and INITIALIZING.

StreamsGroupMemberMetadataKey

{
  "type": "data",
  "name": "StreamsGroupMemberMetadataKey",
  "validVersions": TBD,
  "flexibleVersions": "none",
   "about": "The client id." },
    { "name": "ClientHost", "versions": "0+", "type": "string",
      "about": "The client host." },
    { "name": "RebalanceTimeoutMs", "type": "int32", "versions": "0+", "default": -1,
      "about": "The rebalance timeout." },

    { "name": "TopologyEpoch", "type": "int32", "versions": "0+",
      "about": "The epoch of the topology. Must be non-zero." },

    { "name": "ProcessId", "type": "string", "versions": "0+",
      "about": "Identity of the streams instance that may have multiple consumers." },
    { "name": "UserEndpoint", "type": "Endpoint", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "User-defined endpoint for running interactive queries on this instance." },
    { "name": "ClientTags", "type": "[]KeyValue", "versions": "0+",
      "about": "Used for rack-aware assignment algorithm." }
  ],
  "commonStructs": [
    { "name": "Endpoint", "versions": "0+", "fields": [
      { "name": "GroupIdHost", "type": "string", "versions": TBD"0+",
        "about": "Thehost of groupthe id.endpoint" },
      { "name": "MemberIdPort", "type": "stringuint16", "versions": TBD"0+",
        "about": "Theport of memberthe id.endpoint" }
    ]
}

StreamsGroupMemberMetadataValue

{
  "type},
    { "name": "KeyValue", "versions": "data0+",
      "fields": [
        { "name": "StreamsGroupMemberMetadataValueKey",
  "validVersionstype": "0string",
  "flexibleVersionsversions": "0+",
  "fields": [
        "about": "key of the config" },
        { "name": "InstanceIdValue", "versionstype": "0+string", "nullableVersionsversions": "0+", "type": "string",
          "about": "Thevalue (optional)of instancethe id.config" },

      ]
    }
  ]
}

StreamsGroupMetadataKey

{
  "nametype": "RackIddata",
  "versionsname": "0+StreamsGroupMetadataKey",
  "nullableVersionsvalidVersions": "0+",TBD,
  "typeflexibleVersions": "stringnone",
      "aboutfields": "The (optional) rack id." },[
    { "name": "ClientIdGroupId", "versionstype": "0+string", "typeversions": "string"TBD,
      "about": "The clientgroup id." },
  ]
}

StreamsGroupMetadataValue

{
  { "nametype": "ClientHostdata",
  "versionsname": "0+StreamsGroupMetadataValue",
  "typevalidVersions": "string0",
      "aboutflexibleVersions": "The0+",
 client host."fields": },[
    { "name": "RebalanceTimeoutMsEpoch", "type": "int32", "versions": "0+", "defaulttype": -1"int32",
      "about": "The rebalancegroup timeoutepoch." },

    { "name": "TopologyIdMetadataHash", "typeversions": "string0+", "versionstype": "0+int64",
      "about": "The IDhash of theall topology.topics Mustin bethe non-emptygroup." },

    { "name": "ProcessIdValidatedTopologyEpoch", "typeversions": "string0+", "versionstaggedVersions": "0+", "tag": 0, "default": -1, "type": "int32",
      "about": "Identity of the streams instance that may have multiple consumersThe topology epoch whose topics are validated to be present in a valid configuration in the metadata." },
    { "name": "UserEndpointLastAssignmentConfigs", "typetaggedVersions": "Endpoint0+", "versionsnullableVersions": "0+",
      "about"tag": "User-defined endpoint for running interactive queries on this instance." },
 1, "default": null,
   { "name": "ClientTags", "type": "[]KeyValueLastAssignmentConfig", "versionsabout": "0+",The last used configuration parameters as key-value pairs." }
    ],
  "aboutcommonStructs": [
 "Used for rack-aware assignment{ algorithm."name" }
  ],
  "commonStructs: "LastAssignmentConfig", "versions": "0+", "fields": [
      { "name": "Key", "type": "Endpointstring", "versions": "0+",
  "fields": [      "about": "Key of the config." },
      { "name": "HostValue", "type": "string", "versions": "0+",
        "about": "host of the endpoint" },
      { "name": "Port", "type": "int32", "versions": "0+",
        "about": "port of the endpoint" }
    ]},
    { "name": "KeyValue", "versions": "0+",
      "fields": [
        {Value of the config." }
    ]}
  ]
}

Group Topology Metadata

The topology for the group, and the metadata for input topic partitions consumed in the topology, are persisted in two records.

• The partition metadata record, keyed by the group ID, contains metadata of all topic partitions consumed by the current topology
• The topology record, keyed by the group ID, stores the topology metadata sent in the heartbeat request when members join.

The records are evolved in the following way

• When the group coordinator receives valid topology metadata in a heartbeat request, both records are created, using the topology sent in the request and the topic metadata currently known to the group coordinator.
• When the group coordinator detects that the partition metadata for one of the input partitions used in the topology changed (by comparing the previous partition metadata record to the currently known topic metadata in the broker) during the handling of a group heartbeat, the stream group partition metadata record is updated, and a new assignment is computed.

When the group lost all members and offsets.retention.minutes expires, tombstones is written to remove the records.

StreamsGroupPartitionMetadataKey

{
  "type": "data",
  "name": "KeyStreamsGroupPartitionMetadataKey",
  "typevalidVersions": "string"TBD,
  "versionsflexibleVersions": "0+none",
          "aboutfields": "key of the config" },[
        { "name": "ValueGroupId", "type": "string", "versions": "0+"TBD,
          "about": "valueThe of the configgroup id." }
      ]
    }
  ]
}

...

StreamsGroupPartitionMetadataValue

{
  "type": "data",
  "name": "StreamsGroupMetadataKeyStreamsGroupPartitionMetadataValue",
  "validVersions": TBD"0",
  "flexibleVersions": "none0+",
  "fields": [
    { "name": "GroupIdTopics", "typeversions": "string0+", "versionstype": TBD"[]TopicMetadata",
      "about": "The list groupof topic idmetadata." }
  ]
}

StreamsGroupMetadataValue

{
  "type": "data",
 , "fields": [
      { "name": "TopicId", "versions": "0+", "type": "uuid",
        "about": "The topic id." },
      { "name": "StreamsGroupMetadataValueTopicName",
  "validVersionsversions": "0+",
  "flexibleVersionstype": "0+string",
        "fieldsabout": [ "The topic name." },
      { "name": "EpochNumPartitions", "versions": "0+", "type": "int32",
      "about": "The group epoch." }
  ]
}

Group Topology Metadata

The topology for the group, and the metadata for input topic partitions consumed in the topology, are persisted in two records.

• The partition metadata record, keyed by the group ID, contains metadata of all topic partitions consumed by the current topology
• The topology record, keyed by the group ID, stores the topology metadata sent as part of the StreamsGroupInitialize  request.

The records are evolved in the following way

• When the group coordinator receives a valid StreamsGroupInitialize  request, both records are created, using the topology sent in the request and the topic metadata currently known to the group coordinator.
• When the group coordinator detects that the partition metadata for one of the input partitions used in the topology changed (by comparing the previous partition metadata record to the currently known topic metadata in the broker) during the handling of a group heartbeat, the stream group partition metadata record is updated, and a new assignment is computed.

When the group lost all members (or never had any) and offsets.retention.minutes expires, tombstones is written to remove the records.

StreamsGroupPartitionMetadataKey

{
  "type": "data",
  "name": "StreamsGroupPartitionMetadataKey",
  "validVersions": TBD,
  "flexibleVersions": "none",
       "about": "The number of partitions of the topic." },
      { "name": "PartitionMetadata", "versions": "0+", "type": "[]PartitionMetadata",
        "about": "Partitions mapped to a set of racks. If the rack information is unavailable for all the partitions, an empty list is stored", "fields": [
          { "name": "GroupIdPartition", "typeversions": "string0+", "versionstype": TBD"int32",
      "about": "The group id." }
  ]
}

StreamsGroupPartitionMetadataValue

{
  "typeabout": "data",
The partition "namenumber.": "StreamsGroupPartitionMetadataValue"},
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "TopicsRacks", "versions": "0+", "type": "[]TopicMetadatastring",
            "about": "The listset of topic metadataracks that the partition is mapped to.", "fields": [ }
      { "name": "TopicId", "versions": "0+", "type": "uuid",
        "about": "The topic id." },
     ]}
    ]}
  ]
}

StreamsGroupTopologyKey

{
  "type": "data",
  { "name": "TopicNameStreamsGroupTopologyKey",
  "versionsvalidVersions": "0+"TBD,
  "typeflexibleVersions": "stringnone",
        "aboutfields": "The topic name." },[
      { "name": "NumPartitionsGroupId", "versionstype": "0+string", "typeversions": "int32"TBD,
        "about": "The number of partitions of the topicgroup id." },
  ]
}

StreamsGroupTopologyValue

{
  "type    { "name": "PartitionMetadatadata",
  "versionsname": "0+StreamsGroupTopologyValue",
  "typevalidVersions": "[]PartitionMetadata0",
        "aboutflexibleVersions": "Partitions mapped to a set of racks. If the rack information is unavailable for all the partitions, an empty list is stored","0+",
  "fields": [
          { "name": "PartitionEpoch", "versionstype": "0+int32", "typeversions": "int320+",
            "about": "The epoch of partitionthe numbertopology. " },
          { "name":  "RacksSubtopologies", "versionstype": "0+[]Subtopology", "typeversions": "[]string0+",
            "about": "The setsub-topologies of racks that the partition is mapped tostreams application.",
 }
      ]}"fields": [
    ]}
  ]
}

StreamsGroupTopologyKey

{
  "type    { "name": "dataSubtopologyId",
  "nametype": "StreamsGroupTopologyKeystring",
  "validVersionsversions": TBD"0+",
          "flexibleVersionsabout": "none"String to uniquely identify the subtopology." },
   "fields": [
    { "name": "GroupIdSourceTopics", "type": "[]string", "versions": TBD"0+",
          "about": "The group id topics the topology reads from." },
  ]
}

StreamsGroupTopologyValue

{
  "type        { "name": "dataSourceTopicRegex",
  "nametype": "StreamsGroupTopologyValue[]string",
  "validVersionsversions": "0+",
          "flexibleVersionsabout": "0+",
  "fields": [
Regular expressions identifying topics the subtopology reads from." },
        { "name": "TopologyIdStateChangelogTopics", "type": "string[]TopicInfo", "versions": "0+",
          "about": "The IDset of state changelog thetopics topology.associated Mustwith bethis non-emptysubtopology. " },
        { "name":  "TopologyRepartitionSinkTopics", "type": "[]Subtopologystring", "versions": "0+",
          "about": "The sub-topologiesrepartition oftopics the streamssubtopology writes applicationto.",
      "fields": [ },
        { "name": "SubtopologyRepartitionSourceTopics", "type": "string[]TopicInfo", "versions": "0+",
          "about": "String to uniquely identify the subtopologyThe set of source topics that are internally created repartition topics." },
        { "name": "SourceTopicsCopartitionGroups", "type": "[]string", "versions": "0+"CopartitionGroup", "versions": "0+",
          "about": "A subset of source topics that must be copartitioned.",
          "aboutfields": "The topics the topology reads from." },
[
            { "name": "SourceTopicRegexSourceTopics", "type": "string[]int16", "versions": "0+",
 "nullableVersions": "0+", "default": "null",
          "about": "The regular expressions identifying topics the topology reads from. null if not provided Index into the array on the subtopology level." },
            { "name": "StateChangelogTopicsSourceTopicRegex", "type": "[]TopicInfoint16", "versions": "0+",
              "about": "The set of state changelog topics associated with this subtopology. Regular expressions identifying topics the subtopology reads from. Index into the array on the subtopology level." },
	            { "name": "RepartitionSinkTopicsRepartitionSourceTopics", "type": "[]stringint16", "versions": "0+",
              "about": "The repartitionset topicsof thesource sub-topologytopics writesthat to." }, 
        { "name": "RepartitionSourceTopics", "type": "[]TopicInfo", "versions": "0+",
     are internally created repartition topics. Index into the array on the subtopology level." }
     "about": "The set of source topics]
 that are internally created repartition topics. " }
      ]
    }
  ],
  "commonStructs": [
    { "name": "TopicConfig", "versions": "0+", "fields": [
      { "name": "key", "type": "string", "versions": "0+",
        "about": "The key of the topic-level configuration." },
      { "name": "value", "type": "string", "versions": "0+",
        "about": "The value of the topic-level configuration," }
    ]
    },
    { "name": "TopicInfo", "versions": "0+", "fields": [
      { "name": "Name", "type": "string", "versions": "0+",
        "about": "The name of the topic." },
      { "name": "Partitions", "type": "int32", "versions": "0+",
        "about": "The number of partitions in the topic. Can be 0 if no specific number of partitions is enforced. Always 0 for changelog topics." },
      { "name": "TopicConfigsReplicationFactor", "type": "[]TopicConfigint16", "versions": "0+",
        "nullableVersionsabout": "0+The replication factor of the topic. Can be 0 if the default replication factor should be used." },
      { "name": "TopicConfigs", "type": "[]TopicConfig", "defaultversions": "null0+",
        "about": "Topic-level configurations as key-value pairs."
      }
    ]}
  ]
}

...

For each member, there is a record to store the current member assignment. The record is first created when a member joins (with an empty assignment). Every time an assignment is computed for the member during reconciliation, it is compared to the previous assignment for the member. If the assignment changed, a new version of the current member assignment record is appended to the consumer offset topic. When the member leaves the group, a tombstone is written to remove the record.the member. If the assignment changed, a new version of the current member assignment record is appended to the consumer offset topic. When the member leaves the group, a tombstone is written to remove the record.

Together with the assignment, we track the epoch at which an active task was assigned. This helps with offset commit validation: We allow offset commits for partitions that belong to tasks in all epochs since the assignment epoch of the task. 

StreamsGroupCurrentMemberAssignmentKey

...

{
  "type": "data",
  "name": "StreamsGroupCurrentMemberAssignmentValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "MemberEpoch", "versions": "0+", "type": "int32",
      "about": "The current member epoch that is expected from the member in the heartbeat request." },
    { "name": "PreviousMemberEpoch", "versions": "0+", "type": "int32",
      "about": "If the last epoch bump is lost before reaching the member, the member will retry with the previous epoch." },
    { "name": "State", "versions": "0+", "type": "int8",
      "about": "The member state. See StreamsGroupMember.MemberState for the possible values." },
    { "name": "ActiveTasks", "versions": "0+", "type": "[]TaskIds",
      "about": "Currently assigned active tasks for this streams client." },
    { "name": "StandbyTasks", "versions": "0+", "type": "[]TaskIds",
      "about": "Currently assigned standby tasks for this streams client." },
    { "name": "WarmupTasks", "versions": "0+", "type": "[]TaskIds",
      "about": "Currently assigned warm-up tasks for this streams client": "Currently assigned warm-up tasks for this streams client." },
    { "name": "ActiveTasksPendingRevocation", "versions": "0+", "type": "[]TaskIds",
      "about": "The active tasks that must be revoked by this member." },
    { "name": "StandbyTasksPendingRevocation", "versions": "0+", "type": "[]TaskIds",
      "about": "The standby tasks that must be revoked by this member." },
    { "name": "ActiveTasksPendingRevocationWarmupTasksPendingRevocation", "versions": "0+", "type": "[]TaskIds",
      "about": "The activewarmup tasks that must be revoked by this member." }
  ],
  "commonStructs": [
    { "name": "TaskIds", "versions": "0+", "fields": [
      { "name": "SubtopologySubtopologyId", "type": "string", "versions": "0+",
        "about": "The subtopology identifier." },
      { "name": "Partitions", "type": "[]int32", "versions": "0+",
        "about": "The partitions of the input topics processed by this member input topics processed by this member." },
      { "name": "AssignmentEpochs", "versions": "0+", "nullableVersions": "0+", "taggedVersions": "0+", "tag": 0, "type": "[]int32", "default": null,
        "about": "The epoch at which any partition was assigned to the member. Used to fence zombie commits requests. Of the same length as partitions. If null, all assignment epochs are considered to be equal to the member epoch." }
      ]}
  ]
}

Streams Group Target Assignment

...

{
  "type": "data",
  "name": "StreamsGroupTargetAssignmentMemberValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "ActiveTasks", "versions": "0+", "type": "[]TaskIds",
      "about": "Currently assigned active tasks for this streams client." },
    { "name": "StandbyTasks", "versions": "0+", "type": "[]TaskIds",
      "about": "Currently assigned standby tasks for this streams client." },
    { "name": "WarmupTasks", "versions": "0+", "type": "[]TaskIds",
      "about": "Currently assigned warm-up tasks for this streams client." }
  ],
  "commonStructs": [
    { "name": "TaskIds", "versions": "0+", "fields": [
      { "name": "Subtopology", "type": "string", "versions": "0+",
        "about": "The subtopology identifier streams client." },
      { "name": "PartitionsWarmupTasks", "typeversions": "[]int320+", "versionstype": "0+[]TaskIds",
        "about": "TheCurrently partitionsassigned ofwarm-up thetasks inputfor topicsthis processed by this memberstreams client." }
    ]},
  ]
}

StreamsGroupTargetAssignmentMetadataKey

{
  "type"commonStructs": "data",[
    { "name": "StreamsGroupTargetAssignmentMetadataKeyTaskIds",
  "validVersionsversions": TBD,
  "flexibleVersions": "none",
 "0+", "fields": [
      { "name": "GroupIdSubtopologyId", "type": "string", "versions": TBD"0+",
        "about": "The groupsubtopology ididentifier." }
  ]
}

StreamsGroupTargetAssignmentMetadataValue

{
  "type": "data",
  "name": "StreamsGroupTargetAssignmentMetadataValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "AssignmentEpochPartitions", "versionstype": "0+[]int32", "typeversions": "int320+",
        "about": "The assignment epoch." }
  ]
}

Broker Metrics

The existing group metrics are extended to differentiate between streams groups and consumer groups and account for streams group states.

• Number of groups based on type of the protocol, where the list of protocols is extended by the protocol=streams variation.

kafka.server:type=group-coordinator-metrics,name=group-count,protocol={consumer|classic|streams}

• Number of streams groups based on state

kafka.server:type=group-coordinator-metrics,name=streams-group-count,state={empty|initializing|assigning|reconciling|stable|dead}

• Streams group rebalances sensor

kafka.server:type=group-coordinator-metrics,name=streams-group-rebalance-rate

kafka.server:type=group-coordinator-metrics,name=streams-group-rebalance-count

Broker Configurations

The new group protocol would be guarded by enabling a new version of the group.version feature flag on the broker side.

Updated properties

New properties

Group Configurations

We will add new configurations for the resource type GROUP in DescribeConfigs and IncrementalAlterConfigs to override the default broker configurations dynamically for specific groups.

": "The partitions of the input topics processed by this member." }
    ]}
  ]
}

StreamsGroupTargetAssignmentMetadataKey

{
  "type": "data",
  "name": "StreamsGroupTargetAssignmentMetadataKey",
  "validVersions": TBD,
  "flexibleVersions": "none",
  "fields": [
    { "name": "GroupId", "type": "string", "versions": TBD,
      "about": "The group id." }
  ]
}

StreamsGroupTargetAssignmentMetadataValue

{
  "type": "data",
  "name": "StreamsGroupTargetAssignmentMetadataValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "AssignmentEpoch", "versions": "0+", "type": "int32",
      "about": "The assignment epoch." }
  ]
}

Broker Metrics

The existing group metrics are extended to differentiate between streams groups and consumer groups and account for streams group states.

• Number of groups based on type of the protocol, where the list of protocols is extended by the protocol=streams variation.

kafka.server:type=group-coordinator-metrics,name=group-count,protocol={consumer|classic|streams}

• Number of streams groups based on state

kafka.server:type=group-coordinator-metrics,name=streams-group-count,state={empty|not_ready|assigning|reconciling|stable|dead}

• Streams group rebalances sensor

kafka.server:type=group-coordinator-metrics,name=streams-group-rebalance-rate

kafka.server:type=group-coordinator-metrics,name=streams-group-rebalance-count

Broker Configurations

The new group protocol would be guarded by enabling a new version of the group.version feature flag on the broker side.

Updated properties

New properties

Streams API

...

Streams Configurations

New configurations

Deprecations

In a future major version, when the classic group protocol will be deprecated, the following configuration options will be deprecated:

acceptable.recovery.lag

max.warmup.replicas

num.standby.replicas

probing.rebalance.interval.ms

rack.aware.assignment.strategy

rack.aware.assignment.traffic_cost

rack.aware.assignment.non_overlap_cost

task.assignor.class

Note that both the rack-aware task assignor and customizable client-side task assignment may be introduced into the new protocol in follow-up KIPs (see Future Work section). In this case, the latter 5 configuration options would not be deprecated in the future major version.

Admin API

Add the following methods to the org.apache.kafka.client.admin.AdminClient  interface, mostly backed by the same implementations as the consumer group API.

...

Method signature

...

Description

...

RPC used

...

Group Configurations

We will add new configurations for the resource type GROUP in DescribeConfigs and IncrementalAlterConfigs to override the default broker configurations dynamically for specific groups.

Streams API

In a future major version, when the classic group protocol will be deprecated, the interfaces introduced in KIP-924 would be deprecated, unless an integration with this KIP is proposed in a follow-up KIP.

Streams Configurations

New configurations

Deprecations

In a future major version, when the classic group protocol will be deprecated, the following configuration options will be deprecated:

acceptable.recovery.lag

max.warmup.replicas

num.standby.replicas

probing.rebalance.interval.ms

rack.aware.assignment.tags

rack.aware.assignment.strategy

rack.aware.assignment.traffic_cost

rack.aware.assignment.non_overlap_cost

task.assignor.class

Note that both the rack-aware task assignor and customizable client-side task assignment may be introduced into the new protocol in follow-up KIPs (see Future Work section). In this case, the latter 5 configuration options would not be deprecated in the future major version.

Admin API

Add the following methods to the org.apache.kafka.client.admin.AdminClient  interface, mostly backed by the same implementations as the consumer group API.

These correspond to existing APIs for consumer groups and some of the code will be shared. The main reason for duplicating them is the naming, as the current API uses "consumerGroup" in the name, despite the RPCs just using "group". There are some differences:

• The describeStreamsGroups uses the DescribeStreamsGroup RPC and contains other information than consumer groups.
• A streams group has an extra state -  NOT_READY, and no legacy states from the classic protocol.
• removeMembersFromConsumerGroup will not have a corresponding API in this first version, as it is using the LeaveGroup RPC for classic consumer groups, which is not available for KIP-848-style groups.

Admin

These correspond to existing APIs for consumer groups and some of the code will be shared. The main reason for duplicating them is the naming, as the current API uses "consumerGroup" in the name, despite the RPCs just using "group". There are some differences:

• The describeStreamsGroups uses the DescribeStreamsGroup RPC and contains other information than consumer groups.
• A streams group has an extra state -  INITIALIZING, and no legacy states from the classic protocol.
• removeMembersFromConsumerGroup will not have a corresponding API in this first version, as it is using the LeaveGroup RPC for classic consumer groups, which is not available for KIP-848-style groups.

Admin

   /**
    * Alters offsets for the specified group. In order to succeed, the group must be empty.
    *
    * <p>This is a convenience method for {@link #alterStreamsGroupOffsets(String, Map, AlterStreamsGroupOffsetsOptions)} with default options.
    * See the overload for more details.
    *
    * @param groupId The group for which to alter offsets.
    * @param offsets A map of offsets by partition.
    * @return The AlterStreamsGroupOffsetsResult.
    */
   default AlterStreamsGroupOffsetsResult alterStreamsGroupOffsets(String groupId, Map<TopicPartition, Long> offsets) {
       return alterStreamsGroupOffsets(groupId, offsets, new AlterStreamsGroupOffsetsOptions());
   }
 
   /**
    * Alters offsets for the specified group. In order to succeed, the group must be empty.
    *
    * <p>This operation is not transactional so it may succeed for some partitions while fail for others.
    *
    * @param groupId The group for which to alter offsets.
    * @param offsets A map of offsets by partition. Partitions not specified in the map are ignored.
    * @param options The options to use when altering the offsets.
    * @return The AlterStreamsGroupOffsetsResult.
    */
   AlterStreamsGroupOffsetsResult alterStreamsGroupOffsets(String groupId, Map<TopicPartition, Long> offsets, AlterStreamsGroupOffsetsOptions options);
 
   /**
    * DeleteAlters offsets for a set of topics in a Streamsthe specified group. In order to succeed, the group withmust thebe default optionsempty.
    *
    * <p>This is a convenience method for {@link #deleteStreamsGroupOffsets#alterStreamsGroupOffsets(String, SetMap, DeleteStreamsGroupOffsetsOptionsAlterStreamsGroupOffsetsOptions)} with default options.
    * See the overload for more details.
    *
    * @param groupId The group for which to deletealter offsets.
    * @param topics The topicsoffsets A map of offsets by partition.
    * @return The DeleteStreamsGroupOffsetsResultAlterStreamsGroupOffsetsResult.
    */
   default DeleteStreamsGroupOffsetsResultAlterStreamsGroupOffsetsResult deleteStreamsGroupOffsetsalterStreamsGroupOffsets(String groupId, Map<TopicPartition, Set<String>Long> topicsoffsets) {
       return deleteStreamsGroupOffsetsalterStreamsGroupOffsets(groupId, topicsoffsets, new DeleteStreamsGroupOffsetsOptionsAlterStreamsGroupOffsetsOptions());
   }
 
   /**
    * Delete offsets for a set of topics in a Streams group
    * Alters offsets for the specified group. In order to succeed, the group must be empty.
    *
    * <p>This operation is not transactional so it may succeed for some partitions while fail for others.
    *
    * @param groupId The group for which to deletealter offsets.
    * @param topics The topics offsets A map of offsets by partition. Partitions not specified in the map are ignored.
    * @param options The options to use when deletingaltering the offsets in a Streams group.
    * @return The DeleteStreamsGroupOffsetsResultAlterStreamsGroupOffsetsResult.
    */
   DeleteStreamsGroupOffsetsResultAlterStreamsGroupOffsetsResult deleteStreamsGroupOffsetsalterStreamsGroupOffsets(String groupId,
 Map<TopicPartition,      Set<String> topics,
       DeleteStreamsGroupOffsetsOptions Long> offsets, AlterStreamsGroupOffsetsOptions options);
 
   /**
    * Delete Streams groups from the clusteroffsets for a set of topics in a Streams group with the default options.
    *
    * <p>This is a convenience method for {@link #deleteStreamsGroups(Collection<String>#deleteStreamsGroupOffsets(String, Set, DeleteStreamsGroupsOptionsDeleteStreamsGroupOffsetsOptions)} with default options.
    * See the overload for more details.
    *
    * @param groupIdsgroupId The IDsgroup offor thewhich groups to delete offsets.
    * @param topics The topics.
    * @return The DeleteStreamsGroupsResultDeleteStreamsGroupOffsetsResult.
    */
   default DeleteStreamsGroupsResultDeleteStreamsGroupOffsetsResult deleteStreamsGroups(Collection<String> groupIdsdeleteStreamsGroupOffsets(String groupId, Set<String> topics) {
       return deleteStreamsGroups(groupIdsdeleteStreamsGroupOffsets(groupId, topics, new DeleteStreamsGroupsOptionsDeleteStreamsGroupOffsetsOptions());
   }
 
   /**
    * Delete Streams groups from the cluster offsets for a set of topics in a Streams group.
    *
    * @param groupIdsgroupId The IDsgroup offor the groupswhich to delete offsets.
    * @param topics The topics.
    * @param options The options to use when deleting offsets in a Streams group.
    * @return The DeleteStreamsGroupsResultDeleteStreamsGroupOffsetsResult.
    */
   DeleteStreamsGroupsResultDeleteStreamsGroupOffsetsResult deleteStreamsGroups(Collection<String> groupIds, DeleteStreamsGroupsOptionsdeleteStreamsGroupOffsets(String groupId,
       Set<String> topics,
       DeleteStreamsGroupOffsetsOptions options);
 
   /**
    * Describe someDelete Streams groups infrom the cluster, with the default options.
    *
    * <p>This is a convenience method for {@link #describeStreamsGroups#deleteStreamsGroups(CollectionCollection<String>, DescribeStreamsGroupsOptionsDeleteStreamsGroupsOptions)}
 with default options.
 * with default options.* See the overload for more details.
    *
    * @param groupIds The IDs of the groups to describedelete.
    * @return The DescribeStreamsGroupsResultDeleteStreamsGroupsResult.
    */
   default DescribeStreamsGroupsResultDeleteStreamsGroupsResult describeStreamsGroupsdeleteStreamsGroups(Collection<String> groupIds) {
       return describeStreamsGroupsdeleteStreamsGroups(groupIds, new DescribeStreamsGroupsOptionsDeleteStreamsGroupsOptions());
   }
 
   /**
    * Describe someDelete Streams groups infrom the cluster.
    *
    * @param groupIds The IDs of the groups to describedelete.
    * @param options  The options to use when deleting describinga theStreams groupsgroup.
    * @return The DescribeStreamsGroupsResultDeleteStreamsGroupsResult.
    */
   DescribeStreamsGroupsResultDeleteStreamsGroupsResult describeStreamsGroupsdeleteStreamsGroups(Collection<String> groupIds,
                                                     DescribeStreamsGroupsOptions DeleteStreamsGroupsOptions options);
 
   /**
    * List the Streams group offsets available in the cluster for the specifiedDescribe some Streams groups in the cluster, with the default options.
    *
    * <p>This is a convenience method for {@link #listStreamsGroupOffsets#describeStreamsGroups(MapCollection, ListStreamsGroupOffsetsOptionsDescribeStreamsGroupsOptions)}
    * towith listdefault offsetsoptions. ofSee allthe partitionsoverload for the specified Streams groups with default optionsmore details.
    *
    * @param groupSpecsgroupIds Map of Streams group ids to a spec that specifies the topic partitions The IDs of the groupgroups to list offsets fordescribe.
    * @return The ListStreamsGroupOffsetsResultDescribeStreamsGroupsResult.
    */
   default ListStreamsGroupOffsetsResultDescribeStreamsGroupsResult listStreamsGroupOffsetsdescribeStreamsGroups(Map<String, ListStreamsGroupOffsetsSpec> groupSpecsCollection<String> groupIds) {
       return listStreamsGroupOffsetsdescribeStreamsGroups(groupSpecsgroupIds, new ListStreamsGroupOffsetsOptionsDescribeStreamsGroupsOptions());
   }
 
   /**
    * ListDescribe thesome Streams groupgroups offsets available in the cluster for the specified Streams.
    *
    * @param groupIds The IDs of the groups to describe.
    * @param options  The options to use when describing the groups.
    * @return The DescribeStreamsGroupsResult.
    */
 @param groupSpecs Map of Streams group ids to a spec that specifies the topic partitions of the group to list offsets for.
  DescribeStreamsGroupsResult describeStreamsGroups(Collection<String> groupIds,
                       * @param options The options to use when listing the Streams group offsets.
    * @return The ListStreamsGroupOffsetsResult
    */
   ListStreamsGroupOffsetsResult listStreamsGroupOffsets(Map<String, ListStreamsGroupOffsetsSpec> groupSpecs, ListStreamsGroupOffsetsOptionsDescribeStreamsGroupsOptions options);
 
   /**
    * List the Streams group groupsoffsets available in the cluster for the specified Streams groups with the default options.
    *
    * <p>This is a convenience method for {@link #listStreamsGroups(ListStreamsGroupsOptions)} #listStreamsGroupOffsets(Map, ListStreamsGroupOffsetsOptions)}
    * to list offsets of all partitions for the specified Streams groups with default options.
    *
    * See the overload for more details.
    * @param groupSpecs Map of Streams group ids to a spec that specifies the topic partitions of the group to list offsets for.
    * @return The ListStreamsGroupsResult.ListStreamsGroupOffsetsResult
    */
   default ListStreamsGroupsResultListStreamsGroupOffsetsResult listStreamsGroups(listStreamsGroupOffsets(Map<String, ListStreamsGroupOffsetsSpec> groupSpecs) {
       return listStreamsGroupslistStreamsGroupOffsets(groupSpecs, new ListStreamsGroupsOptionsListStreamsGroupOffsetsOptions());
   }
 
   /**
   }
 
   /**
    * List the Streams group offsets available in the cluster for the specified Streams groups.
    *
    * @param ListgroupSpecs Map theof Streams groups available in group ids to a spec that specifies the topic partitions of the cluster.
group to list offsets *for.
    * @param options The options to use when listing the Streams group groupsoffsets.
    * @return The ListStreamsGroupsResult.ListStreamsGroupOffsetsResult
    */
   ListStreamsGroupsResultListStreamsGroupOffsetsResult listStreamsGroups(ListStreamsGroupsOptionslistStreamsGroupOffsets(Map<String, ListStreamsGroupOffsetsSpec> groupSpecs, ListStreamsGroupOffsetsOptions options);

AlterStreamsGroupOffsetResult

...

/**
 * A detailed description of a single streams group in the cluster.
 */
public class StreamsGroupDescription {

    public StreamsGroupDescription(
        final String groupId,
        final int groupEpoch,
        final String topologyId int targetAssignmentEpoch,
        final int topologyEpoch,
        final Collection<StreamsGroupSubtopologyDescription> subtopologies,
        final Collection<StreamsGroupMemberDescription> members,
        final GroupState groupState,
        final Node coordinator,
        final Set<AclOperation> authorizedOperations
    );

    /**
     * The id of the streams group.
     */
    finalpublic Collection<StreamsGroupMemberDescription> members,String groupId();
    
    final/**
 StreamsGroupState state,
   * The epoch of the final Node coordinator,consumer group.
     */
    finalpublic Set<AclOperation>int authorizedOperationsgroupEpoch();
 
    /**
     * The idepoch of the streamstarget groupassignment.
     */
    public Stringint groupIdtargetAssignmentEpoch();

      /**
     * The idepoch of the currently used topology, or null if uninitialized.
     */
    public Stringint topologyIdtopologyEpoch();
    
     /**
     * A list of the members of the streams group.
     */
    public Collection<StreamsGroupMemberDescription> members();

    /**
     * A list of the subtopologies in the streams group.
     */
    public Collection<StreamsGroupSubtopologyDescription> subtopologies();

    /**
     * The state of the streams group state, or UNKNOWN if the state is too new for us to parse.
     */
    public StreamsGroupStateGroupState stategroupState();

    /**
     * The streams group coordinator, or null if the coordinator is not known.
     */
    public Node coordinator();

    /**
     * authorizedOperations for this group, or null if that information is not known.
     */
    public Set<AclOperation> authorizedOperations();
}

...

/**
 * A detailed description of a single member in the group.
 */
public class StreamsGroupMemberDescription {

    public StreamsGroupMemberDescription(
        final String memberId, 
        final Optional<String>int instanceIdmemberEpoch,
        final Stringint clientIdcurrentAssignmentEpoch,
        final StringOptional<String> clientHostinstanceId, 
        final StringOptional<String> topologyIdrackId, 
        final String processId,
        final Map<String, String> clientTags,
        final List<TaskOffset> taskOffsets, 
        final List<TaskOffset> taskEndOffsets,
        final StreamsGroupMemberAssignment assignment,
        final Optional<StreamsGroupMemberAssignment> targetAssignment);
clientId,
    /**
    final *String TheclientHost,
 id of the group member.
   final int */topologyEpoch,
    public String memberId();

  final String /**processId,
     * The id offinal theOptional<Endpoint> instanceuserEndpoint, if available.
     */
   final publicMap<String, Optional<String> instanceId();
String> clientTags,
    /**
    final *List<TaskOffset> ThetaskOffsets,
 client id of the group member.
  final List<TaskOffset>  */taskEndOffsets,
    public String clientId();

  final StreamsGroupMemberAssignment /**assignment,
     *  The hostfinal ofStreamsGroupMemberAssignment thetargetAssignment,
 group member.
     */
 final boolean isClassic
 public String clientHost( );

    /**
     * The id of the topology present on the clientgroup member.
     */
    public String topologyIdmemberId();

    /**
     * IdentityThe epoch of the streams instance that may have multiple clientsgroup member.
     */
    public Stringint processIdmemberEpoch();

    /**
     * Used The id of the instance, used for rack-aware assignment algorithm static membership, if available.
     */
    public Map<String, String>Optional<String> clientTagsinstanceId(); 
    
    /**
     * Cumulative offsets for tasks The rack ID of the group member.
     */
    public List<TaskOffset>Optional<String> taskOffsetsrackId();
    
    /**
     * Cumulative end offsets for tasksThe client id of the group member.
     */
    public List<TaskOffset>String taskEndOffsetsclientId();

    /**
     * The current assignment host of the group member.
     */
    public StreamsGroupMemberAssignmentString assignmentclientHost();

    /**
     * The target assignment.epoch of the group member. 
     */
    public Optional<StreamsGroupMemberAssignment>int targetAssignmentmemberEpoch();
    
    /**
     * The cumulativeepoch offsetof forthe onecurrent taskassignment.
     */
    public static class TaskOffset {

 int currentAssignmentEpoch();

    /**
     * The epoch public TaskOffset(final String subtopology, final int partition, final long offset);
of the topology present on the client.
     */
    public int topologyEpoch();
 
    /**
     * Identity of the *streams Theinstance subtopologythat identifier.
may have multiple  clients.
     */
        public String subtopologyprocessId(); 

        /**
         * The partition of the task.
    User-defined endpoint for Interactive Queries.
     */
        public intOptional<Endpoint> partitionuserEndpoint();

         /**
         * TheUsed cumulativefor offset (sum of offsets in all input partitions).
  rack-aware assignment algorithm.
       */
    public Map<String,   public long offsetString> clientTags();

    }
}

StreamsGroupMemberAssignment

/**
 *  A description of* theCumulative assignmentsoffsets offor atasks.
 specific streams group member.
 */
public class StreamsGroupMemberAssignment {

 public List<TaskOffset>  public StreamsGroupMemberAssignment(taskOffsets();

    /**
    final List<TaskIds>* activeTasks,
Cumulative task changelog end offsets for tasks.
   final List<TaskIds> standbyTasks,*/
    public    final List<TaskIds> warmupTasksList<TaskOffset> taskEndOffsets();

    /**
     * ActiveThe tasks for this clientcurrent assignment.
     */
    public List<TaskIds>StreamsGroupMemberAssignment activeTasksassignment();

    /**
     * StandbyThe tasks for this clienttarget assignment.
     */
    public List<TaskIds>StreamsGroupMemberAssignment standbyTaskstargetAssignment();     
    
    /**
     * The flag *indicating Warmupwhether tasksa formember thisis clientclassic.
     */
    public List<TaskIds>boolean warmupTasksisClassic();

     /**
     * The Allcumulative tasksoffset for one subtopology of a membertask.
     */
    public static class TaskIdsTaskOffset {

        public TaskIdsTaskOffset(final String subtopologysubtopologyId, final int partition, final List<Integer>long partitionsoffset);

        /**
         * The sub-topologysubtopology identifier.
         */
        public String subtopologysubtopologyId();

        /**
         * The partitionspartition of the input topics processed by this member task.
         */
        public int partition();

        /**
         * The cumulative offset (sum of offsets in all input partitions).
         */
        public List<Integer>long partitionsoffset();
    }
}

...

StreamsGroupMemberAssignment

/**
 * A description of detailedthe descriptionassignments of a specific single subtopologystreams group member.
 */
public class StreamsGroupSubtopologyDescriptionStreamsGroupMemberAssignment {

    public StreamsGroupSubtopologyDescription(
        final String subtopology,
        final List<String> sourceTopics,
  public StreamsGroupMemberAssignment(
     final String sourceTopicRegex, final List<String>List<TaskIds> repartitionSinkTopicsactiveTasks,
        final Map<String, TopicInfo> stateChangelogTopicsList<TaskIds> standbyTasks,
        final Map<String, TopicInfo> repartitionSourceTopics List<TaskIds> warmupTasks
    );

    /**
     * StringActive totasks uniquelyfor identifythis the subtopologyclient.
     */
    public StringList<TaskIds> subtopologyactiveTasks();

    /**
     * TheStandby topicstasks thefor topologythis reads fromclient.
     */
    public List<String>List<TaskIds> sourceTopicsstandbyTasks();
    
    /**
     * TheWarmup regulartasks expressionsfor identifying topics the topology reads from. null if not provided.this client.
     */
    public StringList<TaskIds> sourceTopicRegexwarmupTasks();

    /**
     * All Thetasks repartitionfor topicsone thesubtopology topologyof writesa tomember.
     */
    public List<String> repartitionSinkTopics( static class TaskIds {

        public TaskIds(final String subtopologyId, final List<Integer> partitions);

        /**
     *  The set of* stateThe changelogsubtopology topicsidentifier.
 associated with this sub-topology.
     */
        public Map<String, TopicInfo> stateChangelogTopicsString subtopologyId();

        /**
         * The setpartitions of source topics that are internally created repartition topics.
 the input topics processed by this member.
         */
        public Map<String, TopicInfo> repartitionSourceTopicsList<Integer> partitions();

    }
}

StreamsGroupSubtopologyDescription

/**
 * A detailed description *of Informationa about a topic.single subtopology
     */
    public static class TopicInfoStreamsGroupSubtopologyDescription {

    public StreamsGroupSubtopologyDescription(
    public TopicInfo(final int partitions, final Map<StringString subtopologyId,
 String> topicConfigs);

      final List<String> /**sourceTopics,
        final *List<String> TherepartitionSinkTopics,
 number of partitions in the topic.
  final Map<String, TopicInfo> stateChangelogTopics,
    */
    final Map<String,   public int partitions(TopicInfo> repartitionSourceTopics);

        /**
     * String to uniquely * Configurationsidentify of the topicsubtopology.
         */
        public Map<String, String> topicConfigs();
    }

}

DescribeStreamsGroupsOptions

package org.apache.kafka.client.admin;
  
 String subtopologyId();

    /**
     * The Optionstopics forthe {@link Admin#describeStreamsGroups(Collection<String>, DescribeStreamsGroupsOptions)}topology reads from.
 *
 * The API of*/
 this class is evolving,public see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class DescribeStreamsGroupsOptions extends AbstractOptions<DescribeStreamsGroupsOptions> {
    public DescribeStreamsGroupsOptions includeAuthorizedOperations(boolean includeAuthorizedOperations);
 List<String> sourceTopics();

    /**
     * The repartition topics the topology writes to.
     */
    public booleanList<String> includeAuthorizedOperationsrepartitionSinkTopics();
}

ListStreamsGroupOffsetsResult

package org.apache.kafka.clients.admin;
  

    /**
     * The resultset of the {@link Admin#listStreamsGroupOffsets(Map<String, ListStreamsGroupOffsetsSpec>, ListStreamsGroupOffsetsOptions)} call.
 * <p>
 state changelog topics associated with this subtopology.
     */
    public Map<String, TopicInfo> stateChangelogTopics();

    /**
     * The APIset of thissource classtopics isthat evolving,are seeinternally {@linkcreated Admin} for detailsrepartition topics.
     */
@InterfaceStability.Evolving
    public classMap<String, ListStreamsGroupOffsetsResult {TopicInfo> repartitionSourceTopics();

    /**
     * ReturnInformation about a futuretopic.
 which yields all Map<String, Map<TopicPartition, Long> objects, if requests for all the groups succeed.
 */
    public static class TopicInfo {

      */
  public TopicInfo(final publicint KafkaFuture<Map<Stringpartitions, final Map<TopicPartitionMap<String, Long>>>String> all() {
 topicConfigs);

   }
 
    /**
     * Return a future which* yieldsThe anumber map of topicpartitions partitionsin tothe offsetstopic.
 for the specified group.
     */
        public KafkaFuture<Map<TopicPartition, Long>> partitionsToOffset(String groupId) {
int partitions();

     }
}

ListStreamsGroupOffsetsOptions

package org.apache.kafka.client.admin;
  
/**
 * Options for {@link Admin#listStreamsGroupOffsets(Map<String, ListStreamsGroupOffsetsSpec>, ListStreamsGroupOffsetsOptions)}.
 * <p>
 * TheConfigurations APIof ofthe thistopic.
 class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class ListStreamsGroupOffsetsOptions extends AbstractOptions<ListStreamsGroupOffsetsOptions> {
}

...

        public Map<String, String> topicConfigs();
    }

}

DescribeStreamsGroupsOptions

package org.apache.kafka.client.admin;
  
/**
 * SpecificationOptions of Streams group offsets to list using for {@link Admin#listStreamsGroupOffsetsAdmin#describeStreamsGroups(Map<StringCollection<String>, ListStreamsGroupOffsetsSpec>, ListStreamsGroupOffsetsOptionsDescribeStreamsGroupsOptions)}.
 * <p>
 * The API of this class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class ListStreamsGroupOffsetsSpec {
  public ListStreamsGroupOffsetsSpec();
 
  /**
   * Set the topic partitions whose offsets are to be listed for a Streams group.
   */
  ListStreamsGroupOffsetsSpec topicPartitions(Collection<TopicPartition> topicPartitions);
 
  /**
   * Returns the topic partitions whose offsets are to be listed for a Streams group.
   */
  Collection<TopicPartition> topicPartitions, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class DescribeStreamsGroupsOptions extends AbstractOptions<DescribeStreamsGroupsOptions> {
    public DescribeStreamsGroupsOptions includeAuthorizedOperations(boolean includeAuthorizedOperations);
 
    public boolean includeAuthorizedOperations();
}

...

ListStreamsGroupOffsetsResult

package org.apache.kafka.clients.admin;
  
/**
 * The result of the {@link Admin#listStreamsGroups(ListStreamsGroupsOptionsAdmin#listStreamsGroupOffsets(Map<String, ListStreamsGroupOffsetsSpec>, ListStreamsGroupOffsetsOptions)} call.
 * <p>
 * The API of this class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class ListStreamsGroupsResultListStreamsGroupOffsetsResult {
    /**
     * ReturnsReturn a future thatwhich yields eitherall anMap<String, exceptionMap<TopicPartition, or the full set of Streams group listingsLong> objects, if requests for all the groups succeed.
     */
    public KafkaFuture<Collection<StreamsGroupListing>>KafkaFuture<Map<String, Map<TopicPartition, Long>>> all() {
    }
 
    /**
     * ReturnsReturn a future which yields justa themap validof listings.
topic partitions to offsets  */
    public KafkaFuture<Collection<StreamsGroupListing>> valid() {
    }
  
    /**
     * Returns a future which yields just the errors which occurredfor the specified group.
     */
    public KafkaFuture<Collection<Throwable>> errors( KafkaFuture<Map<TopicPartition, Long>> partitionsToOffset(String groupId) {
    }
}

...

ListStreamsGroupOffsetsOptions

package org.apache.kafka.client.admin;
 
import org.apache.kafka.common.StreamsGroupState;
 
/**
 * AOptions listingfor of a Streams group in the cluster{@link Admin#listStreamsGroupOffsets(Map<String, ListStreamsGroupOffsetsSpec>, ListStreamsGroupOffsetsOptions)}.
 * <p>
 * The API of this class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class StreamsGroupListingListStreamsGroupOffsetsOptions {
extends  public StreamsGroupListing(String groupId);
  public StreamsGroupListing(String groupId, Optional<StreamsGroupState> state);
 
  /**
   * The id of the Streams group.
   */
  public String groupId();
 
  /**
   * The Streams group state.
   */
  public Optional<StreamsGroupState> state();
}

...

AbstractOptions<ListStreamsGroupOffsetsOptions> {
}

ListStreamsGroupOffsetsSpec

package org.apache.kafka.client.admin;
 
import org.apache.kafka.common.StreamsGroupState;
 
/**
 * Options for {@link Admin#listStreamsGroups(ListStreamsGroupsOptionsSpecification of Streams group offsets to list using {@link Admin#listStreamsGroupOffsets(Map<String, ListStreamsGroupOffsetsSpec>, ListStreamsGroupOffsetsOptions)}.
 * <p>
 * The API of this class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class ListStreamsGroupOffsetsSpec ListStreamsGroupsOptions{
 extends AbstractOptions<ListStreamsGroupsOptions> {public ListStreamsGroupOffsetsSpec();
  
  /**
     * IfSet statesthe istopic set,partitions onlywhose groupsoffsets in these states willare to be returned.listed Otherwise,for alla groupsStreams are returnedgroup.
     */
    public ListStreamsGroupsOptions inStates(Set<StreamsGroupState> statesListStreamsGroupOffsetsSpec topicPartitions(Collection<TopicPartition> topicPartitions);
 
    /**
     * ReturnReturns the listtopic ofpartitions Stateswhose thatoffsets are requestedto orbe emptylisted iffor noa states have been specifiedStreams group.
     */
    public Set<StreamsGroupState> statesCollection<TopicPartition> topicPartitions();
}

GroupType

Another case is added to the org.apache.kafka.common.GroupType  enum:

StreamsGroupState

GroupState

Another case is added to theA new enum org.apache.kafka.common.StreamsGroupStateGroupState  is addedenum:

NOT_READY

Exceptions

The following new exceptions are added to the org.apache.kafka.common.errors package, corresponding to the new error codes in the Kafka protocol.

• StreamsInvalidTopologyStreamsInvalidTopologyException  - The supplied topology is invalid. Returned if the client sends a topology that does not fulfill the expected invariants.
• StreamsMissingSourceTopics  - There are source topics missing for a topology that is supposed to be initialized. Also returned if the source topic regular expression matched no topics.
• StreamsInconsistentInternalTopics  - There are internal topics present on the broker that are not consistent with the internal topic requirements of the provided topology.

StreamsInvalidTopology is fatal, StreamsMissingSourceTopics and StreamsInconsistentInternalTopics are all subclasses of RetriableException.

Command-line tools

• StreamsInvalidTopologyEpochException  - The client provided a invalid topology epoch with respect to the stream group state, for example, the topology was changed by the epoch not bumped.
• StreamsTopologyFencedException  - When a client attempts to join with an outdated topology epoch.

Command-line tools

kafka-groups.sh

We add an option --streams  to the above commad-line tool, which filter the groups to show share groups. Furthermore streams  becomes one of the possible values for --group-type . 

kafka-streams-groups.sh

A new tool called kafka-streams-groups.sh is added for working with streams groups. It has the following options: