...

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 with which the topology was initialized provided by the first heartbeat with the new topology.

• 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 that 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 - 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 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 standby 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, 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.

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

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 he 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.

...

• A (classic)
• B (classic)
• C (streams)

...

• 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]

...

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. Source topics inconsistent
   Condition: The source topics are inconsistent, if two source 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.
   Behavior: The group will enter/remain in state NOT_READY.  Heartbeat responses will indicate status SOURCE_TOPICS_INCONSISTENT . In the status detail, we specify at least one inconsistency.
3. Internal topics are inconsistent
   Condition: One or more internal topics are inconsistent, for example, they are not copartition 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
   Behavior: The group will enter/remain in state NOT_READY.  Heartbeat responses will indicate status INTERNAL_TOPICS_INCONSISTENT . In the status detail, we specify at least one inconsistency.
4. 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.
5. 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 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 that 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 - 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 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 standby 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, 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.

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

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 he 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

...

• 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]

...

• • , 0_5], standby=[0_

...

• • 0

...

• • ,

...

• • 0_

...

• • 2, 0_4]

...

    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 IDC 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]

A follow    C's warm-up rebalance is triggered so that A can report the revoked task 0_2 is caught up. A is requested to revoke active task 0_2. Since , thus A does not need to revoke tasks anymore the increase its 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_2, 0_3], standby=[0_1, 0_5], warm-up=[]
    • JoinGroupResponse: generation ID=2524
    • 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_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.

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.

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".

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",
  "flexibleVersions": "0+",
  "fields": [

{
  "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 for static membership otherwise." },
    { "name": "RackIdGroupId", "type": "string", "versions": "0+",  "nullableVersionsentityType": "0+groupId", "default": "null",
      "about": "null if not provided or if it didn't change since the last heartbeat; the rack ID of consumer otherwiseThe 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": "RebalanceTimeoutMsMemberEpoch", "type": "int32", "versions": "0+", "default": -1,
      "about": "-1 if it didn't change since the last heartbeat; the maximum time in millisecondsThe current member epoch; 0 to join the group; -1 to leave the group; -2 to indicate that the coordinator will wait on the static member to revoke its partitions otherwisewill rejoin." },

	    { "name":  "TopologyInstanceId", "type": "Topologystring", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "Thenull topologyif datanot ofprovided theor streamsif application. 
				Used to initializeit didn't change since the topologylast ofheartbeat; 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."instance ID for static membership otherwise." },
     { "fieldsname": [
        { "name"RackId", "type": "TopologyIdstring", "typeversions": "string0+",  "versionsnullableVersions": "0+",
 "default": "null",
    	  "about": "Thenull if IDnot ofprovided theor topology.if Usedit todidn't checkchange ifsince the topologylast correspondsheartbeat; to the topologyrack initializedID onof theconsumer brokersotherwise." },
		    { "name": "SubtopologiesRebalanceTimeoutMs", "type": "[]Subtopologyint32", "versions": "0+",
 "default": -1,
    	  "about": "The sub-topologies1 ofif theit streams application.",
	      "fields": [
	        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":  "SubtopologyIdTopology", "type": "stringTopology", "versions": "0+",
	 "nullableVersions": "0+", "default": "null",
      "about": "StringThe totopology uniquelydata identifyof the streams sub-topologyapplication. Deterministically generated from
				Used to initialize the topology" },
	        { "name": "SourceTopics", "type": "[]string", "versions": "0+",
	          "about": "The topics the topology reads from." },
	 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": "SourceTopicRegexTopologyId", "type": "[]string", "versions": "0+",
  
	        	   "about": "The regularID expressionsof identifying topicsthe topology. Used to check if the topology corresponds to the sub-topology initialized on readsthe frombrokers." },
	        	{ "name": "StateChangelogTopicsSubtopologies", "type": "[]TopicInfoSubtopology", "versions": "0+",
	        	  "about": "The setsub-topologies of statethe changelog topics associated with this sub-topology. Created automatically." },streams application.",
	      "fields": [
	        { "name": "RepartitionSinkTopicsSubtopologyId", "type": "[]string", "versions": "0+",
	          "about": "TheString to repartitionuniquely topicsidentify the sub-topology writes to.. Deterministically generated from the topology" }, 
	        { "name": "RepartitionSourceTopicsSourceTopics", "type": "[]TopicInfostring", "versions": "0+",
	          "about": "The set of source topics thatthe aretopology internally created repartition topics. Created automaticallyreads from." },
	        { "name": "CopartitionGroupsSourceTopicRegex", "type": "[]CopartitionGroupstring", "versions": "0+", 
	               "about": "AThe subsetregular ofexpressions sourceidentifying topics thatthe mustsub-topology bereads copartitionedfrom.",
	          "fields": [ },
	            { "name": "SourceTopicsStateChangelogTopics", "type": "[]int16TopicInfo", "versions": "0+",
	              "about": "The topicsset theof topologystate readschangelog from.topics Indexassociated intowith the array on the subtopology levelthis sub-topology. Created automatically." },
    	        { "name": "SourceTopicRegexRepartitionSinkTopics", "type": "[]int16string", "versions": "0+",
	        	      "about": "RegularThe expressions identifyingrepartition topics the subtopologysub-topology readswrites from. Index into the array on the subtopology level.to." }, 
    	     	    { "name": "RepartitionSourceTopics", "type": "[]int32TopicInfo", "versions": "0+",
	            	  "about": "The set of source topics that are internally created repartition topics. Index into the array on the subtopology levelCreated automatically." }
          	]}
        ]},
	  ]	
    }	
 
    { "name": "ActiveTasksCopartitionGroups", "type": "[]TaskIdsCopartitionGroup", "versions": "0+",
	 "nullableVersions": "0+", "default": "null",
      "about": "CurrentlyA ownedsubset activeof taskssource fortopics thisthat client.must Null if unchanged since last heartbeat." },
be copartitioned.",
	         { "namefields": "StandbyTasks[
	            { "name": "SourceTopics", "type": "[]TaskIdsint16", "versions": "0+", "nullableVersions": "0+", "default": "null",

	              "about": "CurrentlyThe topics ownedthe standbytopology tasksreads forfrom. thisIndex client.into Nullthe ifarray unchangedon sincethe lastsubtopology heartbeatlevel." },
    	        { "name": "WarmupTasksSourceTopicRegex", "type": "[]TaskIdsint16", "versions": "0+", "nullableVersions": "0+", "default": "null",

        	      "about": "CurrentlyRegular expressions ownedidentifying warm-uptopics tasksthe forsubtopology thisreads clientfrom. Null if unchanged since last heartbeatIndex into the array on the subtopology level." },

         	    { "name": "ProcessIdRepartitionSourceTopics", "type": "string[]int32", "versions": "0+", "nullableVersions": "0+", "default": "null",

            	  "about": "IdentityThe set of thesource streams instancetopics that mayare internally havecreated multiplerepartition consumerstopics. Index Nullinto the ifarray unchangedon sincethe lastsubtopology heartbeatlevel." }
 },         	]}
        ]}
	  ]	
    }	
 
    { "name": "UserEndpointActiveTasks", "type": "Endpoint[]TaskIds", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "User-defined endpointCurrently owned active tasks for Interactivethis Queriesclient. Null if unchanged since last heartbeat." },
    { "name": "ClientTagsStandbyTasks", "type": "[]KeyValueTaskIds", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "UsedCurrently owned standby tasks for rack-aware assignment algorithmthis client. Null if unchanged since last heartbeat." },

    { "name": "TaskOffsetsWarmupTasks", "type": "[]TaskOffsetTaskIds", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "Cumulative changelog offsets for tasks. Only updated when a Currently owned warm-up tasktasks hasfor caught up, and according to the task offset intervalthis client. Null if unchanged since last heartbeat." },

    { "name": "TaskEndOffsetsProcessId", "type": "[]TaskOffsetstring", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "CumulativeIdentity changelogof end-offsetsthe forstreams tasks.instance Onlythat updatedmay whenhave a warm-up task has caught up, and according to the task offset interval. Null if unchanged since last heartbeat.multiple consumers. Null if unchanged since last heartbeat." },
    { "name": "ShutdownApplicationUserEndpoint", "type": "boolEndpoint", "versions": "0+", "nullableVersions": "0+", "default": false"null",
      "about": "Whether all Streams clients in the group should shut downUser-defined endpoint for Interactive Queries. Null if unchanged since last heartbeat." }
  ],

  "commonStructs": [
    { "name": "KeyValueClientTags", "versionstype": "0+[]KeyValue", "fields": [
      { "name"versions": "Key0+", "typenullableVersions": "string0+", "versionsdefault": "0+null",
        "about": "key of the configUsed for rack-aware assignment algorithm. Null if unchanged since last heartbeat." },

      { "name": "ValueTaskOffsets", "type": "string[]TaskOffset", "versions": "0+",
   "nullableVersions": "0+", "default": "null",
      "about": "valueCumulative ofchangelog theoffsets config" }
    ]},
    { "name": "TopicInfo", "versions": "0+", "fields": [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": "NameTaskEndOffsets", "type": "string[]TaskOffset", "versions": "0+",
        "aboutnullableVersions": "The name of the topic." },
      { "name0+", "default": "Partitionsnull", "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 topicsCumulative 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": "TopicConfigsShutdownApplication", "type": "[]KeyValuebool", "versions": "0+", "nullableVersionsdefault": "0+"false,
 "default": "null",
        "about": "Topic-levelWhether configurationsall asStreams key-value pairs."
     clients in the group should shut down." }
  ],

   ]},
"commonStructs": [
     { "name": "EndpointKeyValue", "versions": "0+", "fields": [
      { "name": "HostKey", "type": "string", "versions": "0+",
        "about": "hostkey of the endpointconfig" },
      { "name": "PortValue", "type": "int32string", "versions": "0+",
        "about": "portvalue of the endpointconfig" }
    ]},
     { "name": "TaskOffsetTopicInfo", "versions": "0+", "fields": [
      { "name": "SubtopologyIdName", "type": "string", "versions": "0+",
        "about": "The sub-topology identifier name of the topic." },
      { "name": "PartitionPartitions", "type": "int32", "versions": "0+",
        "about": "The partition number of partitions in the topic. Can be 0 if no specific number of partitions is enforced. Always 0 for changelog topics." },
      { "name": "OffsetTopicConfigs", "type": "int64[]KeyValue", "versions": "0+", "nullableVersions": "0+", "default": "null",
        "about": "The offset." Topic-level configurations as key-value pairs."
      }
    ]},
     { "name": "TaskIdsEndpoint", "versions": "0+", "fields": [
      { "name": "SubtopologyIdHost", "type": "string", "versions": "0+",
        "about": "The sub-topology identifier.host of the endpoint" },
      { "name": "PartitionsPort", "type": "[]int32", "versions": "0+",
        "about": "The partitionsport of the input topics processed by this member endpoint" }
    ]},
    { "name": "TaskOffset", "versions": "0+", "fields": [
      { "name": "SubtopologyId", "type": "string", "versions": "0+",
        "about": "The sub-topology identifier." },
      { "name": "Partition", "type":  ]}
  ]
}

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_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.

• 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 group ID.

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

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.

STREAMS_MISSING_SOURCE_TOPICS is returned if there are source topics missing during the initialization of the topology. 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.

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. If the group is initialized:
      1. Looks up the group
   2. If the group is not initialized:
      1. Creates group and initializes the topology by creating all required internal topics.
      2. Writes the topology, keyed with the GroupId to the consumer offset topic. Existing records will be overwritten.
   3. 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).
   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.
4. Updates information of the member if needed. 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.

"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 sub-topology 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_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.

• 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 group ID.

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

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. If the group is initialized:
      1. Looks up the group
   2. If the group is not initialized:
      1. Creates group and initializes the topology by creating all required internal topics.
      2. Writes the topology, keyed with the GroupId to the consumer offset topic. Existing records will be overwritten.
   3. 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).
   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.
4. Compares the topology ID to the group topology (if initialized) and initializes the group topology, if sent with the heartbeat
   1. Checks if all topics exist in the right configuration on the broker
   2. Write the topology record to the offset topic
   3. If any topics are inconsistent, this is indicated in the status in the heartbeat response, the the group will enter state NOT_READY .
5. Updates information of the member if needed. The group epoch is incremented if there is any change.
6. Reconcile the member assignments as explained earlier in this document.
7. If we find that internal topics are missing for group, we will send a corresponing create topic request to the controller.

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.

{
  "apiKey": TBD,
  "type": "response",
  "name": "StreamsGroupHeartbeatResponse",
  "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+)
  // - 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+)
  "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." },
    { "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." },

{
  "apiKey": TBD,
  "type": "response",
  "name": "StreamsGroupHeartbeatResponse",
  "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+)
  // - 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_MISSING_SOURCE_TOPICS (version 0+)
  // - STREAMS_INCONSISTENT_INTERNAL_TOPICS (version 0+) 
  "fields": [
    // Same as consumer group heart beat
    { "name": "ThrottleTimeMsTaskOffsetIntervalMs", "type": "int32", "versions": "0+",
      "about": "The durationinterval in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },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." },

    // Topology updating
    { "name": "ErrorCodeGroupTopologyId", "type": "int16string", "versions": "0+", "nullableVersions": "0+", "default": "0+"null,
      "about": "The top-level error code, or 0current ID of the topology for the group. Null if thereunchanged wassince nolast errorheartbeat." },

    { "name": "ErrorMessageStatus", "type":  "string[]Status", "versions":  "0+",  "nullableVersions": "0+", "default": "null", 
      "about": "The top-level error message, or nullIndicate zero or more status for the group.  Null if thereunchanged wassince nolast errorheartbeat." },

    // The streams app knows which partitions to fetch from given this information
    { "name": "MemberIdActiveTasks", "type": "string[]TaskIds", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "TheAssigned memberactive idtasks generatedfor bythis the coordinatorclient. OnlyNull providedif whenunchanged thesince member joins with MemberEpoch == 0last heartbeat." },
    { "name": "MemberEpochStandbyTasks", "type": "int32[]TaskIds", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "The member epochAssigned standby tasks for this client. Null if unchanged since last heartbeat." },
    { "name": "HeartbeatIntervalMsWarmupTasks", "type": "int32[]TaskIds", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "The heartbeat interval in millisecondsAssigned warm-up tasks for this client. Null if unchanged since last heartbeat." },

    // IQ-related information
    { "name": "AcceptableRecoveryLagPartitionsByUserEndpoint", "type": "int32[]EndpointToPartitions", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "TheGlobal maximalassignment laginformation aused warm-up task can have to be considered caught-upfor IQ. Null if unchanged since last heartbeat." },
      "fields": [
        { "name": "TaskOffsetIntervalMsUserEndpoint", "type": "int32Endpoint", "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.User-defined endpoint to connect to the node" },

    // Topology updating
    { "name": "GroupTopologyIdPartitions", "type": "string[]TopicPartition", "versions": "0+",
 "nullableVersions": "0+", "default": null,
      "about": "TheAll currentpartitions IDavailable ofon the topologynode" }
 for the group. Null if unchanged]
 since last heartbeat." }
  ],
  "commonStructs": [
    { "name": "Status", "type":  "[]Status", "versions":  "0+",  "nullableVersionsfields": "0+", "default": "null", 
[
      // Possible  "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": "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": "Partitions", "type": "[]TopicPartition", "versions": "0+",
          "about": "All partitions available on the node" }
      ]
    }
  ],
  "commonStructs": [
    { "name": "Status", "versions": "0+", "fields": [
      // Possible status codes
      //  0 - INCONSISTENT_TOPOLOGY      - The topology ID supplied is inconsistent with the topology for this streams group.
      //  1 - MISSING_SOURCE_TOPICS      - One or more source topics are missing or a source topic regex resolves to zero topics.status codes
      //  0 - INCONSISTENT_TOPOLOGY        - The topology ID supplied is inconsistent with the topology 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 - INCONSISTENT_SOURCE_TOPICS   - One or more source topics are inconsistent, for example, they are not copartition despite being 
      //                                     part of a copartition group.
      //                                     Inconsistent topics are indicated in the StatusDetail.
      //  3 - INCONSISTENT_INTERNAL_TOPICS - One or more internal topics are inconsistent, for example, they are not copartition 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. 
      //                                      In the status detail, we specify all missing source topics and all regular expressions matching zero topics.Inconsistent topics are indicated in the StatusDetail. 
      //  24 - INCONSISTENTMISSING_SOURCEINTERNAL_TOPICS      - One or more internal topics are inconsistent, for example, they are not copartition despite being missing. 
      //                                     Missing topics are             part of a copartition group, or the number of partitions in a changelog topic does not correspondindicated in the StatusDetail. 
      //                                       The group coordinator will attempt to create all missing internal topics, if any toerrors theoccur maximalduring
 number of source topic partition for// that  subtopology.
            //    2  - MISSING_INTERNAL_TOPICS - One or more internal topics do not exist. Missing topics are                topic creation, this will be indicated in the StatusDetail.
        //  35 - SHUTDOWN_APPLICATION         - A client requested the shutdown of the whole application.
      { "name": "StatusCode", "type": "int8", "versions": "0+",
        "about": "A code to indicate that a particular status is active for the group membership" },
      { "name": "StatusDetail", "type": "string", "versions": "0+",
        "about": "A string representation of the status." }
    ]},
    { "name": "TopicPartition", "versions": "0+", "fields": [
      { "name": "Topic", "type": "string", "versions": "0+",
        "about": "topic name" },
      { "name": "Partitions", "type": "[]int32", "versions": "0+",
        "about": "partitions" }
    ]},
    { "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." }
    ]}
    { "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" }
    ]}
  ]
}

...

{
  "apiKey": TBD,
  "type": "response",
  "name": "StreamsGroupDescribeResponse",
  "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+)
  // - INVALID_GROUP_ID (version 0+)
  // - GROUP_ID_NOT_FOUND (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": "Groups", "type": "[]DescribedGroup", "versions": "0+",
      "about": "Each described group.",
      "fields": [
        { "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",
          "about": "The top-level error message, or null if there was no error." },
        { "name": "GroupId", "type": "string", "versions": "0+", "entityType": "groupId",
          "about": "The group ID string." },
        { "name": "GroupState", "type": "string", "versions": "0+",
          "about": "The group state string, or the empty string." },
        { "name": "GroupEpoch", "type": "int32", "versions": "0+",
          "about": "The group epoch." },
        { "name": "AssignmentEpoch", "type": "int32", "versions": "0+",
          "about": "The assignment epoch." },
        { "name": "TopologyId", "type": "string", "versions": "0+",
          "about": "The ID of the currently initialized topology for this group." }, 
        { "name": "Topology", "type": "[]Subtopology", "versions": "0+",
          "about": "The resolved sub-topologies of the streams application.",
 This contains the configured sub-topologies, where the number of "fields": [
            { "name": "SubtopologyId", "type": "string", "versions": "0+",
    partitions are set and any regular expressions are resolved to actual topics. Null if the group is uninitialized, source topics are missing or inconsistent.",
          "aboutfields": "String to uniquely identify the subtopology." },[
            { "name": "SourceTopicsSubtopologyId", "type": "[]string", "versions": "0+",
              "about": "The topicsString to uniquely identify the topology reads fromsubtopology." },
            { "name": "SourceTopicRegexSourceTopics", "type": "[]string", "versions": "0+",
                           "about": "The regular expressions identifying topics the topology reads from." },
            { "name": "RepartitionSinkTopics", "type": "[]string", "versions": "0+",
              "about": "The topics the topology writes to." },
            { "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+",
              "about": "The set of source topics that are internally created repartition topics. Created automatically." }
          ]
        },
        { "name": "Members", "type": "[]Member", "versions": "0+",
          "about": "The members.",
          "fields": [
            { "name": "MemberId", "type": "string", "versions": "0+",
              "about": "The member ID." },
            { "name": "MemberEpoch", "type": "int32", "versions": "0+",
              "about": "The member epoch." },
            { "name": "InstanceId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
              "about": "The member instance ID for static membership." },
            { "name": "RackId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
              "about": "The rack ID." },

            { "name": "ClientId", "type": "string", "versions": "0+",
              "about": "The client ID." },
            { "name": "ClientHost", "type": "string", "versions": "0+",
              "about": "The client host." },

            { "name": "TopologyId", "type": "string", "versions": "0+",
              "about": "The ID of the topology on the client." },

            { "name": "ProcessId", "type": "string", "versions": "0+",
              "about": "Identity of the streams instance that may have multiple clients. " },
            { "name": "ClientTags", "type": "[]KeyValue", "versions": "0+",
              "about": "Used for rack-aware assignment algorithm." },
            { "name": "TaskOffsets", "type": "[]TaskOffset", "versions": "0+",
              "about": "Cumulative changelog offsets for tasks." },
            { "name": "TaskEndOffsets", "type": "[]TaskOffset", "versions": "0+",
              "about": "Cumulative changelog end offsets for tasks." },

            { "name": "Assignment", "type": "Assignment", "versions": "0+",
              "about": "The current assignment." },
            { "name": "TargetAssignment", "type": "Assignment", "versions": "0+",
              "about": "The target assignment." }
          ]},
        { "name": "AuthorizedOperations", "type": "int32", "versions": "0+", "default": "-2147483648",
          "about": "32-bit bitfield to represent authorized operations for this group." }
      ]
    }
  ],
  "commonStructs": [
    { "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": "TopicPartitions", "versions": "0+", "fields": [
      { "name": "TopicId", "type": "uuid", "versions": "0+",
        "about": "The topic ID." },
      { "name": "TopicName", "type": "string", "versions": "0+", "entityType": "topicName",
        "about": "The topic name." },
      { "name": "Partitions", "type": "[]int32", "versions": "0+",
        "about": "The partitions." }
    ]},
    { "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": "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." }
    ]},
    { "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": "TopicConfigs", "type": "[]KeyValue", "versions": "0+", "nullableVersions": "0+", "default": "null",
        "about": "Topic-level configurations as key-value pairs."
      }
    ]}
  ]
}

...

• 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

...

• The describeStreamsGroups uses the DescribeStreamsGroup RPC and contains other information than consumer groups.
• A streams group has an extra state -  INITIALIZINGNOT_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.

...

/**
 * A detailed description of a single subtopology
 */
public class StreamsGroupSubtopologyDescription {

    public StreamsGroupSubtopologyDescription(
        final String subtopology,
        final List<String> sourceTopics,
        final String sourceTopicRegex, 
        final List<String> repartitionSinkTopics,
        final Map<String, TopicInfo> stateChangelogTopics,
        final Map<String, TopicInfo> repartitionSourceTopics);

    /**
     * String to uniquely identify the subtopology.
     */
    public String subtopology();

    /**
     * The topics the topology reads from.
     */
    public List<String> sourceTopics();

    /**
     * The regular expressions identifying topics the topology reads from. null if not provided.
     */
    public String sourceTopicRegex();

    /**
     * The repartition topics the topology writes to.
     */
    public List<String> repartitionSinkTopics();

    /**
     * The set of state changelog topics associated with this sub-topology.
     */
    public Map<String, TopicInfo> stateChangelogTopics();

    /**
     * The set of source topics that are internally created repartition topics.
     */
    public Map<String, TopicInfo> repartitionSourceTopics();

    /**
     * Information about a topic.
     */
    public static class TopicInfo {

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

        /**
         * The number of partitions in the topic.
         */
        public int partitions();

        /**
         * Configurations of the topic.
         */
        public Map<String, String> topicConfigs();
    }

}

...

A new enum org.apache.kafka.common.StreamsGroupState  is added:

NOT_READY

Exceptions

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

• StreamsInvalidTopology  - 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

kafka-streams-groups.sh

...

...