Status

Current state: Under DiscussionAccepted

Discussion thread: here

JIRA: here

...

• 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 extensibility, even though subtopologies are currently numbered) and a partition number.

• The process ID to identify multiple 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 the hash 0 by default. The epoch can be bumped by the client to deploy a new version of the topology representation by default.

  Topology

  metadata

  needed for creating internal topics and computing the assignment

  .

• For each task, the current sum of task 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.

...

The metadata sent to the broker via the heartbeat 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 synchronous 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 the all source topics and internal topics are createdexist, the group coordinator starts assigning tasks to the members of the group.

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 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 member sends a topology with that heartbeattopology epoch is not bumped, the group coordinator re-initializes will respond with an error, indicating that the member is incorrectly updating the broker-side topology with the topology metadata contained in the first heartbeat of the Streams member. 

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 epoch . The first heartbeat will only send a topology when a member is joining for the first time, not when rejoining, e.g. after being fenced. The member can also opt out of updating the topology, and there is a client-side configuration to disable topology updates. If the topology ID of an existing member does not correspond to the current topology ID of the grouptopology epoch of the group 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 explicitly 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.

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.

Handling topic topology mismatches

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

...

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 partitionedSource topics inconsistent
   Condition: The source Some topics are inconsistentincorrectly partitioned, 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, 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 SOURCEINCORRECTLY_PARTITIONED_TOPICS_INCONSISTENT . In the status detail, we specify at least one inconsistencyincorrectly partitioned topic.
3. Internal topics are inconsistentmissing
   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.Internal topics are missing
   Condition: One or more internal topics are missing.
   Behavior: If discovered during a heartbeat, the group coordinator will 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.

...

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

...

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

...

The AK implementation will provide the following assignors

• highly_available - Like Similar to the current HighAvailabilityTaskAssignor

• sticky - Like similar to the current StickyTaskAssignor

...

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

...

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

...

The possible states of the streams group are EMPTY, ASSIGNING, RECONCILING , STABLE, NOT_READY, DEAD. The states are similar as for 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 and , 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 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 it had to revoke any partitions.

...

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

...

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

{
  "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", "type": "string", "versions": "0+", "entityType": "groupId",
      "about": "The group identifier." },     
    { "name": "MemberId", "type": "string", "versions": "0+",
      "about": "The member ID generated by the streams coordinatorconsumer. The member ID must be kept during the entire lifetime of the memberstreams consumer process." },
      { "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": "InstanceIdEndpointInformationEpoch", "type": "stringint32", "versions": "0+",
      "nullableVersionsabout": "0+The current endpoint epoch of this client, represents the latest endpoint epoch this client received"},
    { "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": "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 consumerthe member 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 partitionstasks otherwise." },

	    { "name":  "Topology", "type": "Topology", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "The topology datametadata 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": "TopologyIdEpoch", "type": "stringint32", "versions": "0+",
      	    "about": "The IDepoch 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 sub-topologysubtopology. 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 sub-topologysubtopology reads from." },
	            { "name": "StateChangelogTopics", "type": "[]TopicInfo", "versions": "0+",
	              "about": "The set of state changelog topics associated with this sub-topologysubtopology. Created automatically." },
	            { "name": "RepartitionSinkTopics", "type": "[]string", "versions": "0+",
	              "about": "The repartition topics the subtopology sub-topology 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": "[]int32int16", "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": "TopicConfigsReplicationFactor", "type": "[]KeyValueint16", "versions": "0+",
        "nullableVersionsabout": "0+The replication factor of the topic. Can be 0 if the default replication factor should be used." },
      { "name": "TopicConfigs", "type": "[]KeyValue", "defaultversions": "null0+",
        "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": "int32uint16", "versions": "0+",
        "about": "port of the endpoint" }
    ]},
    { "name": "TaskOffset", "versions": "0+", "fields": [
      { "name": "SubtopologyId", "type": "string", "versions": "0+",
        "about": "The sub-topologysubtopology 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 sub-topologysubtopology identifier." },
      { "name": "Partitions", "type": "[]int32", "versions": "0+",
        "about": "The partitions of the input topics processed by this member." }
    ]}
  ]
}

...

• 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  on all topics included in the message

...

• GroupId must be non-empty.

• Either MemberId is must be 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 setsEach 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)

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

...

1. Performs request validation.

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

   1. If the group is initialized:Looks up Look up or create the group.
      1. 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.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. Creates the member.
   4. 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
   5. 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_
   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.
   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 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. If we find that internal topics are missing for group, we will send a corresponing create topic request to the controller.

...

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

{
  "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- 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+", "nullableVersions": "0+", "default": "null",
      "about": "The member id is always generated by the coordinator. Only provided when the member joins with MemberEpoch == 0." },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." },

    // Topology updating
    { "name": "GroupTopologyIdStatus", "type":  "string[]Status", "versions":  "0+",  "nullableVersions": "0+", "default": "null",
      "about": "TheIndicate currentzero IDor ofmore the topologystatus for the group.  Null if unchanged since last heartbeat." },

    { "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 // 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": "PartitionsActivePartitions", "type": "[]TopicPartition", "versions": "0+",
          "about": "All active partitions available on the node" },
      ]
  {  }
  ],
  "commonStructs"name": "StandbyPartitions", "type": "[
    ]TopicPartition", "versions": "0+",
          "about": "All standby paritions available on the node"}
      ]
    }
  ],
  "commonStructs": [
    { "name": "Status", "versions": "0+", "fields": [
      // Possible status codes
      //  0 - INCONSISTENTSTALE_TOPOLOGY                 - The topology IDepoch supplied is inconsistentlower withthan 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 - INCONSISTENTINCORRECTLY_SOURCEPARTITIONED_TOPICS   - One or more source topics are inconsistentincorrectly partitioned, forthat exampleis, they are not copartitioncopartitioned despite being 
      //                                       part of a copartition group.
, or the number of partitions //in  a  changelog  topic  does  not  correspond
            //             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
  to the maximal number //of source topic partition for that subtopology.
         //                      part of a copartition group, or the number of partitions in a changelog topic does not correspond
  Incorrectly partitioned topics are //indicated in the StatusDetail.
      //  3 - MISSING_INTERNAL_TOPICS        - One or more internal topics are missing. 
      //  to the maximal number of source topic partition for that subtopology. 
      //                                     InconsistentMissing topics are indicated in the StatusDetail. 
      //  4     - MISSING_INTERNAL_TOPICS      - One or more internal topics are missing. 
      //                      The  group  coordinator  will  attempt to create all  missing  Missinginternal topics, areif indicatedany inerrors theoccur StatusDetail. during
      //                                     The group coordinator will attempt to create all missing internal topics, if any errors occur during
      //                                     topic creation, this will be indicated in StatusDetail.
        //  54 - SHUTDOWN_APPLICATION             - A client requested the shutdown of the whole application.
      { "name": "StatusCode", "type": "int8", "versions": "0+",
//  5 - ASSIGNMENT_DELAYED          "about": "A code to- indicateNo thatassignment awas particularprovided statusbecause isassignment activecomputation forwas thedelayed.
 group membership" },
   //  6 { "name": "StatusDetail", "type": "string", "versions": "0+",
- MISSING_CLIENT_TAGS             "about": "A string representation of the - (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", "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": "int32uint16", "versions": "0+",
        "about": "port of the endpoint" }
    ]}
   ]
}

StreamsGroupDescribe API

...

{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker"],
  "name": "StreamsGroupDescribeRequest",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "GroupIds", "type": "[]string", "versions": "0+", "entityType": "groupId",
      "about": "The ids of the groups to describe" },
    { "name": "IncludeAuthorizedOperations", "type": "bool", "versions": "0+",
      "about": "Whether to include authorized operations." }
  ]
}

Required ACL

• Read Group

• DESCRIBE on the group resource

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

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

...

• 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": "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":  "TopologyIdTopology", "type": "stringTopology", "versions": "0+",
          "about"nullableVersions": "The ID of the currently initialized topology for this group." }, 
        { "name0+", "default": "Topologynull", "type": "[]Subtopology", "versions": "0+",
          "about": "The topology metadata resolvedcurrently sub-topologiesinitialized offor the streams application. ThisCan containsbe thenull configured sub-topologies, 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 inconsistentin case of a describe error.",
          "fields": [
            { "name": "SubtopologyIdEpoch", "type": "stringint32", "versions": "0+",
              "about": "String to uniquely identify the subtopologyThe epoch of the currently initialized topology for this group." },
            { "name": "SourceTopicsSubtopologies", "type": "[]stringSubtopology", "versions": "0+",
    "nullableVersions": "0+", "default": "null",
              "about": "The subtopologies 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 topology reads from." }, group is uninitialized, source topics are missing or incorrectly partitioned.",
              "fields": [
                { "name": "RepartitionSinkTopicsSubtopologyId", "type": "[]string", "versions": "0+",
                  "about": "TheString topicsto theuniquely topologyidentify writesthe tosubtopology." },
                { "name": "StateChangelogTopicsSourceTopics", "type": "[]TopicInfostring", "versions": "0+",
                  "about": "The set of state changelog topics associated withthe this subtopology. Createdreads automaticallyfrom." },
                { "name": "RepartitionSourceTopicsRepartitionSinkTopics", "type": "[]TopicInfostring", "versions": "0+",
                  "about": "The setrepartition of source topics thatthe aresubtopology internally created repartition topics. Created automatically." }
  writes to." },
        ]
        },
        { "name": "MembersStateChangelogTopics", "type": "[]MemberTopicInfo", "versions": "0+",
                  "about": "The members.",
set of state changelog topics associated with this subtopology. Created automatically."fields": [},
                { "name": "MemberIdRepartitionSourceTopics", "type": "string[]TopicInfo", "versions": "0+",
                  "about": "The member ID." }, set of source topics that are internally created repartition topics. Created automatically." }
              ]}
          ]},
        { "name": "MemberEpochMembers", "type": "int32[]Member", "versions": "0+",
              "about": "The member epoch." }, members.",
          "fields": [
            { "name": "InstanceIdMemberId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
              "about": "The member instance ID for static membership." },
            { "name": "RackIdMemberEpoch", "type": "stringint32", "versions": "0+",
 "nullableVersions": "0+", "default": "null",
              "about": "The rackmember IDepoch." },
  
            { "name": "ClientIdInstanceId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
              "about": "The member clientinstance ID for static membership." },
            { "name": "ClientHostRackId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
              "about": "The clientrack hostID." },

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

            { "name": "ProcessIdClientHost", "type": "string", "versions": "0+",
              "about": "IdentityThe of the streams instance that may have multiple clients. client host." },

            { "name": "ClientTagsTopologyEpoch", "type": "[]KeyValueint32", "versions": "0+",
              "about": "Used for rack-aware assignment algorithmThe epoch of the topology on the client." },

            { "name": "TaskOffsetsProcessId", "type": "[]TaskOffsetstring", "versions": "0+",
              "about": "Cumulative changelog offsets for tasks.Identity of the streams instance that may have multiple clients. " },
            { "name": "TaskEndOffsetsUserEndpoint", "type": "[]TaskOffsetEndpoint", "versions": "0+", "nullableVersions": "0+", "default": "null",
              "about": "Cumulative changelog end offsets for tasksUser-defined endpoint for Interactive Queries. Null if not defined for this client." },

                 { "name": "AssignmentClientTags", "type": "Assignment[]KeyValue", "versions": "0+",
              "about": "The currentUsed for rack-aware assignment algorithm." },
            { "name": "TargetAssignmentTaskOffsets", "type": "Assignment[]TaskOffset", "versions": "0+",
              "about": "The target assignmentCumulative changelog offsets for tasks." },
          ]},
        { "name": "AuthorizedOperationsTaskEndOffsets", "type": "int32[]TaskOffset", "versions": "0+", "default": "-2147483648",

              "about": "32-bitCumulative bitfieldchangelog toend representoffsets authorized operations for this grouptasks." },

      ]
    }
  ],
{  "commonStructsname": [
    { "name"Assignment", "type": "TaskOffsetAssignment", "versions": "0+",
              "fieldsabout": [ "The current assignment." },
            { "name": "SubtopologyIdTargetAssignment", "type": "stringAssignment", "versions": "0+",
              "about": "The subtopologytarget identifierassignment." },
            { "name": "PartitionIsClassic", "type": "int32bool", "versions": "0+",
        "      "about": "The partitionTrue for classic members that have not been upgraded yet." }
        ]},
        { "name": "OffsetAuthorizedOperations", "type": "int64int32", "versions": "0+", "default": "-2147483648",
          "about": "The offset32-bit bitfield to represent authorized operations for this group." }
      ]},
    { "name": "TopicPartitions", "versions": "0+", "fields}
  ],
  "commonStructs": [
      { "name": "TopicId", "type": "uuidEndpoint", "versions": "0+",
        "about"fields": "The topic ID." },[
      { "name": "TopicNameHost", "type": "string", "versions": "0+", "entityType": "topicName",
        "about": "Thehost of topicthe name.endpoint" },
      { "name": "PartitionsPort", "type": "[]int32uint16", "versions": "0+",
        "about": "The partitions.port of the endpoint" }
    ]},
      { "name": "AssignmentTaskOffset", "versions": "0+", "fields": [
      { "name": "ActiveTasksSubtopologyId", "type": "[]TaskIdsstring", "versions": "0+",
        "about": "ActiveThe tasks for this clientsubtopology identifier." },
      { "name": "StandbyTasksPartition", "type": "[]TaskIdsint32", "versions": "0+",
        "about": "Standby tasks for this clientThe partition." },
      { "name": "WarmupTasksOffset", "type": "[]TaskIdsint64", "versions": "0+",
        "about": "Warm-up tasks for this client. The offset." }
    ]},
    { "name": "TaskIdsTopicPartitions", "versions": "0+", "fields": [
      { "name": "SubtopologyIdTopicId", "type": "stringuuid", "versions": "0+",
        "about": "The subtopologytopic identifierID." },
      { "name": "PartitionsTopicName", "type": "[]int32string", "versions": "0+", "entityType": "topicName",
        "about": "The partitions of the input topics processed by this member topic name." },
      { "name": "Partitions", "type": "[]int32", "versions": "0+",
        "about": "The partitions." }
    ]},
    { "name": "KeyValueAssignment", "versions": "0+", "fields": [
      { "name": "KeyActiveTasks", "type": "string[]TaskIds", "versions": "0+",
        "about": "keyActive tasks offor thethis configclient." },
      { "name": "ValueStandbyTasks", "type": "string[]TaskIds", "versions": "0+",
        "about": "valueStandby tasks offor thethis configclient." },
    ]},
    { "name": "TopicInfoWarmupTasks", "versionstype": "0+[]TaskIds", "fieldsversions": ["0+",
      {  "nameabout": "Name", "type": "string", "Warm-up tasks for this client. " }
    ]},
    { "name": "TaskIds", "versions": "0+", "fields": [
      { "name": "SubtopologyId", "type": "string", "versions": "0+",
        "about": "The name of the topicsubtopology identifier." },
      { "name": "Partitions", "type": "[]int32", "versions": "0+",
        "about": "The numberpartitions of partitions in the topic.input Cantopics beprocessed 0by if no specific number of partitions is enforced. Always 0 for changelog topics." this member." }
    ]},
      { "name": "TopicConfigsKeyValue", "typeversions": "[]KeyValue", "versions0+", "fields": [
      { "name": "0+Key", "nullableVersionstype": "0+string", "defaultversions": "null0+",
        "about": "Topic-levelkey configurationsof as key-value pairs."the config" },
      }
    ]}
  ]
}

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.

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

{ "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."
      }
    ]}
  ]
}

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": "GroupId", "type": "string", "versions": TBD,
      "about": "The group id." },
    { "name": "MemberId", "type": "string", "versions": TBD,
      "about": "The member id." }
  ]
}

StreamsGroupMemberMetadataValue

{
  "type": "data",
  "name": "StreamsGroupMemberMetadataValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "InstanceId", "versions": "0+", "nullableVersions": "0+", "type": "string",
      "about": "The (optional) instance ID for static membership." },
    { "name": "RackId", "versions": "0+", "nullableVersions": "0+", "type": "string",
      "about": "The (optional) rack id." },
    { "name": "ClientId", "versions": "0+", "type": "string",
      "about": "The client id." },
    { "name": "ClientHost", "versions": "0+", "type": "string",
      "about": "The client host." },

{
  "type": "data",
  "name": "StreamsGroupMemberMetadataKey",
  "validVersions": TBD,
  "flexibleVersions": "none",
  "fields": [
    { "name": "GroupIdRebalanceTimeoutMs", "type": "stringint32", "versions": TBD"0+", "default": -1,
      "about": "The grouprebalance idtimeout." },

    { "name": "MemberIdTopologyEpoch", "type": "stringint32", "versions": TBD"0+",
      "about": "The member id epoch of the topology. Must be non-zero." },

    ]
}

StreamsGroupMemberMetadataValue

{
  "typename": "dataProcessId",
  "nametype": "StreamsGroupMemberMetadataValuestring",
  "validVersionsversions": "0+",
      "flexibleVersionsabout": "0+",
  "fields": [Identity of the streams instance that may have multiple consumers." },
    { "name": "UserEndpoint", "type": "InstanceIdEndpoint", "versions": "0+", "nullableVersions": "0+", "typedefault": "stringnull",
        "about": "The (optional) instance ID for static membershipUser-defined endpoint for running interactive queries on this instance." },
    { "name": "RackIdClientTags", "versionstype": "0+[]KeyValue", "nullableVersionsversions": "0+",
      "typeabout": "string"Used for rack-aware assignment algorithm." }
  ],
  "commonStructs": [
   "about": "The (optional) rack id." },
 { "name": "Endpoint", "versions": "0+", "fields": [
      { "name": "ClientIdHost", "versionstype": "0+string", "typeversions": "string0+",
        "about": "Thehost of clientthe id.endpoint" },
      { "name": "ClientHostPort", "versionstype": "0+uint16", "typeversions": "string0+",
        "about": "Theport of clientthe host.endpoint" }
    ]},
    { "name": "RebalanceTimeoutMsKeyValue", "type": "int32", "versions": "0+", "default": -1,
      "aboutfields": [
 "The rebalance timeout." },

    { "name": "TopologyIdKey", "type": "string", "versions": "0+",
          "about": "The IDkey of the topology. Must be non-empty.config" },

        { "name": "ProcessIdValue", "type": "string", "versions": "0+",
          "about": "Identityvalue of the streamsconfig" instance}
 that may have multiple consumers." },]
    }
  ]
}

StreamsGroupMetadataKey

{
  "nametype": "UserEndpointdata",
  "typename": "EndpointStreamsGroupMetadataKey",
  "versionsvalidVersions": "0+"TBD,
      "aboutflexibleVersions": "User-defined endpoint for running interactive queries on this instance." },"none",
  "fields": [
    { "name": "ClientTagsGroupId", "type": "[]KeyValuestring", "versions": "0+"TBD,
      "about": "UsedThe for rack-aware assignment algorithmgroup id." }
  ]
}

StreamsGroupMetadataValue

{,
  "commonStructstype": ["data",
  "name": "StreamsGroupMetadataValue",
 { "namevalidVersions": "Endpoint0",
  "versionsflexibleVersions": "0+",
  "fields": [
      { "name": "HostEpoch", "typeversions": "string0+", "versionstype": "0+int32",
        "about": "hostThe of the endpointgroup epoch." },
      { "name": "PortMetadataHash", "typeversions": "int320+", "versionstype": "0+int64",
        "about": "portThe hash of theall endpoint"topics }
in the group."  ]},
    { "name": "KeyValueValidatedTopologyEpoch", "versions": "0+",
 "taggedVersions": "0+",   "tag": 0, "fieldsdefault": [
        { "name-1, "type": "Keyint32", "type": "string", "versions
      "about": "0+",
          "about": "key of the configThe topology epoch whose topics are validated to be present in a valid configuration in the metadata." },
        { "name": "ValueLastAssignmentConfigs", "typetaggedVersions": "string0+", "versionsnullableVersions": "0+",
          "about"tag": "value of the config" }1, "default": null,
      ]
    }
  ]
}

StreamsGroupMetadataKey

{
  "type": "data[]LastAssignmentConfig",
  "nameabout": "StreamsGroupMetadataKey",
  "validVersions": TBD,
  "flexibleVersions": "none"The last used configuration parameters as key-value pairs." }
  ],
  "fieldscommonStructs": [
    { "name": "GroupIdLastAssignmentConfig", "typeversions": "string0+", "versionsfields": TBD,[
      "about": "The group id." }
  ]
}

StreamsGroupMetadataValue

{
  "typename": "dataKey",
  "nametype": "StreamsGroupMetadataValuestring",
  "validVersionsversions": "0+",
        "flexibleVersionsabout": "0+"Key of the config." },
  "fields": [
    { "name": "EpochValue", "versionstype": "0+string", "typeversions": "int320+",
        "about": "TheValue of groupthe epochconfig." }
    ]}
  ]
}

Group Topology Metadata

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

...

{
  "type": "data",
  "name": "StreamsGroupTopologyValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "TopologyIdEpoch", "type": "stringint32", "versions": "0+",
      "about": "The IDepoch of the topology. Must be non-empty. " },
    { "name":  "TopologySubtopologies", "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." },
        { "name": "SourceTopics", "type": "[]string", "versions": "0+",
          "about": "The topics the topology reads from." },
        { "name": "SourceTopicRegex", "type": "[]string", "versions": "0+",
          "about": "TheRegular regular expressions identifying topics the topologysubtopology reads from." },
        { "name": "StateChangelogTopics", "type": "[]TopicInfo", "versions": "0+",
          "about": "The set of state changelog topics associated with this subtopology. " },
	        { "name": "RepartitionSinkTopics", "type": "[]string", "versions": "0+",
          "about": "The repartition topics the sub-topologysubtopology writes to." },
 
          { "name": "RepartitionSourceTopics", "type": "[]TopicInfo", "versions": "0+",
          "about": "The set of source topics that are internally created repartition topics. " },
        { "name": "CopartitionGroups", "type": "[]CopartitionGroup", "versions": "0+",
          "about": "A subset of source topics that must be copartitioned.",
          "fields": [
            { "name": "SourceTopics", "type": "[]int32int16", "versions": "0+",
              "about": "The topics the topology reads from. Index into the array on the subtopology level." },
            { "name": "SourceTopicRegex", "type": "[]int32int16", "versions": "0+",
              "about": "Regular expressions identifying topics the subtopology reads from. Index into the array on the subtopology level." },
            { "name": "RepartitionSourceTopics", "type": "[]int32int16", "versions": "0+",
              "about": "The set of source topics that are internally created repartition topics. Index into the array on the subtopology level." }
          ]
        }
         ]
    }
  ],
  "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+0+",
        "about": "The replication factor of the topic. Can be 0 if the default replication factor should be used." },
      { "name": "TopicConfigs", "nullableVersionstype": "0+[]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.

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 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": "ActiveTasksWarmupTasks", "versions": "0+", "type": "[]TaskIds",
      "about": "Currently assigned activewarm-up tasks for this streams client." },
    { "name": "StandbyTasksActiveTasksPendingRevocation", "versions": "0+", "type": "[]TaskIds",
      "about": "Currently assigned standby tasks forThe active tasks that must be revoked by this streams clientmember." },
    { "name": "WarmupTasksStandbyTasksPendingRevocation", "versions": "0+", "type": "[]TaskIds",
      "about": "Currently assigned warm-up tasks forThe standby tasks that must be revoked by this streams clientmember." },
    { "name": "ActiveTasksPendingRevocationWarmupTasksPendingRevocation", "versions": "0+", "type": "[]TaskIds",
      "about": "The activewarmup tasks that must be revoked by this member." }
  ],
  "commonStructs": [
    { "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": "SubtopologyIdAssignmentEpochs", "typeversions": "string0+", "versionsnullableVersions": "0+",
        "abouttaggedVersions": "The subtopology identifier." },
      { "name0+", "tag": "Partitions"0, "type": "[]int32", "versionsdefault": "0+"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 of the input topics processed by this member. If null, all assignment epochs are considered to be equal to the member epoch." }
      ]}
  ]
}

Streams Group Target Assignment

...

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.

...

Deprecations

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

...

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:

...

   /**
    * 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);
 
   /**
    * Delete offsets for a set of topics in a Streams group with the default options.
    *
    * <p>This is a convenience method for {@link #deleteStreamsGroupOffsets(String, Set, DeleteStreamsGroupOffsetsOptions)} with default options.
    * See the overload for more details.
    *
    * @param groupId The group for which to delete offsets.
    * @param topics The topics.
    * @return The DeleteStreamsGroupOffsetsResult.
    */
   default DeleteStreamsGroupOffsetsResult deleteStreamsGroupOffsets(String groupId, Set<String> topics) {
       return deleteStreamsGroupOffsets(groupId, topics, new DeleteStreamsGroupOffsetsOptions());
   }
 
   /**
    * Delete offsets for a set of topics in a Streams group.
    *
    * @param groupId The group for which to delete offsets.
    * @param topics The topics.
    * @param options The options to use when deleting offsets in a Streams group.
    * @return The DeleteStreamsGroupOffsetsResult.
    */
   DeleteStreamsGroupOffsetsResult deleteStreamsGroupOffsets(String groupId,
       Set<String> topics,
       DeleteStreamsGroupOffsetsOptions options);
 
   /**
    * Delete Streams groups from the cluster with the default options.
    *
    * <p>This is a convenience method for {@link #deleteStreamsGroups(Collection<String>, DeleteStreamsGroupsOptions)} with default options.
    * See the overload for more details.
    *
    * @param groupIds The IDs of the groups to delete.
    * @return The DeleteStreamsGroupsResult.
    */
   default DeleteStreamsGroupsResult deleteStreamsGroups(Collection<String> groupIds) {
       return deleteStreamsGroups(groupIds, new DeleteStreamsGroupsOptions());
   }
 
   /**
    * Delete Streams groups from the cluster.
    *
    * @param groupIds The IDs of the groups to delete.
    * @param options The options to use when deleting a Streams group.
    * @return The DeleteStreamsGroupsResult.
    */
   DeleteStreamsGroupsResult deleteStreamsGroups(Collection<String> groupIds, DeleteStreamsGroupsOptions options);
 
   /**
    * Describe some Streams groups in the cluster, with the default options.
    *
    * <p>This is a convenience method for {@link #describeStreamsGroups(Collection, DescribeStreamsGroupsOptions)}
    * with default options. See the overload for more details.
    *
    * @param groupIds The IDs of the groups to describe.
    * @return The DescribeStreamsGroupsResult.
    */
   default DescribeStreamsGroupsResult describeStreamsGroups(Collection<String> groupIds) {
       return describeStreamsGroups(groupIds, new DescribeStreamsGroupsOptions());
   }
 
   /**
    * Describe some Streams groups in the cluster.
    *
    * @param groupIds The IDs of the groups to describe.
    * @param options  The options to use when describing the groups.
    * @return The DescribeStreamsGroupsResult.
    */
   DescribeStreamsGroupsResult describeStreamsGroups(Collection<String> groupIds,
                                                     DescribeStreamsGroupsOptions options);
 
   /**
    * List the Streams group offsets available in the cluster for the specified Streams groups with the default options.
    *
    * <p>This is a convenience method for {@link #listStreamsGroupOffsets(Map, ListStreamsGroupOffsetsOptions)}
    * to list offsets of all partitions for the specified Streams groups with default options.
    *
    * @param groupSpecs Map of Streams group ids to a spec that specifies the topic partitions of the group to list offsets for.
    * @return The ListStreamsGroupOffsetsResult
    */
   default ListStreamsGroupOffsetsResult listStreamsGroupOffsets(Map<String, ListStreamsGroupOffsetsSpec> groupSpecs) {
       return listStreamsGroupOffsets(groupSpecs, new ListStreamsGroupOffsetsOptions());
   }
 
   /**
    * List the Streams group offsets available in the cluster for the specified Streams groups.
    *
    * @param groupSpecs Map of Streams group ids to a spec that specifies the topic partitions of the group to list offsets for.
    * @param options The options to use when listing the Streams group offsets.
    * @return The ListStreamsGroupOffsetsResult
    */
   ListStreamsGroupOffsetsResult listStreamsGroupOffsets(Map<String, ListStreamsGroupOffsetsSpec> groupSpecs, ListStreamsGroupOffsetsOptions options);

AlterStreamsGroupOffsetResult

package org.apache.kafka.clients.admin;
 
/**
 * The result of the {@link Admin#alterStreamsGroupOffsets(String groupId, Map<TopicPartition, * @return The ListStreamsGroupOffsetsResult
    */
   ListStreamsGroupOffsetsResult listStreamsGroupOffsets(Map<String, ListStreamsGroupOffsetsSpec> groupSpecs, ListStreamsGroupOffsetsOptions options);
 
Long>), AlterStreamsGroupOffsetsOptions)} call.
 * <p>
 * The API of this class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class AlterStreamsGroupOffsetsResult {
    /**
     * Return Lista thefuture Streamswhich groupssucceeds availableif inall the clusteralter with the default optionsoffsets succeed.
     */
    public *KafkaFuture<Void> <p>This is a convenience method for {@link #listStreamsGroups(ListStreamsGroupsOptions)} with default options.
all() {
    }
 
    /**
     * SeeReturn thea overloadfuture forwhich morecan details.
be used to check *
the result for a * @return The ListStreamsGroupsResultgiven partition.
     */
   default public ListStreamsGroupsResultKafkaFuture<Void> listStreamsGroupspartitionResult(final TopicPartition partition) {
       return listStreamsGroups(new ListStreamsGroupsOptions())}
}

AlterStreamsGroupOffsetsOptions

package org.apache.kafka.client.admin;
   }
 
   /**
    * List the Streams groups available in* Options for the cluster.
    *
    * @param options The options to use when listing the Streams groups.
    * @return The ListStreamsGroupsResult.
    */
   ListStreamsGroupsResult listStreamsGroups(ListStreamsGroupsOptions options);

...

{@link Admin#alterStreamsGroupOffsets(String groupId, Map<TopicPartition, Long>), AlterStreamsGroupOffsetsOptions)} call.
 * <p>
 * The API of this class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class AlterStreamsGroupOffsetsOptions extends AbstractOptions<AlterStreamsGroupOffsetsOptions> {
}

DeleteStreamsGroupOffsetsResult

package org.apache.kafka.clients.admin;
  
/**
 * The result of the {@link Admin#alterStreamsGroupOffsetsAdmin#deleteStreamsGroupOffsets(String groupId, Map<TopicPartitionSet<String>, Long>), AlterStreamsGroupOffsetsOptionsDeleteStreamsGroupOffsetsOptions)} call.
 * <p>
 * The API of this class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class AlterStreamsGroupOffsetsResultDeleteStreamsGroupOffsetsResult {
    /**
     * Return a future which succeeds only if all the alterdeletions offsets succeed.
     */
    public KafkaFuture<Void> all() {
    }
 
    /**
     * Return a future which can be used to check the result for a given partitiontopic.
     */
    public KafkaFuture<Void> partitionResult(final TopicPartitionString partitiontopic) {
    }
}

...

DeleteStreamsGroupOffsetsOptions

package org.apache.kafka.client.admin;
  
/**
 * Options for the {@link Admin#alterStreamsGroupOffsetsAdmin#deleteStreamsGroupOffsets(String groupId, Map<TopicPartitionSet<String>, Long>DeleteStreamsGroupOffsetsOptions), AlterStreamsGroupOffsetsOptions)} call.
 * <p>
 * The API of this class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class AlterStreamsGroupOffsetsOptionsDeleteStreamsGroupOffsetsOptions extends AbstractOptions<AlterStreamsGroupOffsetsOptions>AbstractOptions<DeleteStreamsGroupOffsetsOptions> {
}

...

DeleteStreamsGroupsResult

package org.apache.kafka.clients.admin;
  
/**
 * The result of the {@link Admin#deleteStreamsGroupOffsetsAdmin#deleteStreamsGroups(StringCollection<String>, Set<String>, DeleteStreamsGroupOffsetsOptionsDeleteStreamsGroupsOptions)} call.
 * <p>
 * The API of this class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class DeleteStreamsGroupOffsetsResultDeleteStreamsGroupsResult {
    /**
     * Return a future which succeeds only if all the deletions succeed.
     */
    public KafkaFuture<Void> all() {
    }
 
    /**
     * Return a future map from group id to futures which can be used to check the resultstatus forof aindividual given topicdeletions.
     */
    public KafkaFuture<Void> partitionResult(final String topicMap<String, KafkaFuture<Void>> deletedGroups() {
    }
}

...

DeleteStreamsGroupsOptions

package org.apache.kafka.client.admin;
  
/**
 * Options for the {@link Admin#deleteStreamsGroupOffsetsAdmin#deleteStreamsGroups(StringCollection<String>, Set<String>, DeleteStreamsGroupOffsetsOptionsDeleteStreamsGroupsOptions)} call.
 * <p>
 * The API of this class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class DeleteStreamsGroupOffsetsOptionsDeleteStreamsGroupsOptions extends AbstractOptions<DeleteStreamsGroupOffsetsOptions>AbstractOptions<DeleteStreamsGroupsOptions> {
}

...

DescribeStreamsGroupsResult

package org.apache.kafka.clients.admin;
  
/**
 * The result of the {@link Admin#deleteStreamsGroupsAdmin#describeStreamsGroups(Collection<String>, DeleteStreamsGroupsOptionsDescribeStreamsGroupsOptions)} call.
 * <p>
 * The API of this class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class DeleteStreamsGroupsResultDescribeStreamsGroupsResult {
    /**
     * Return a future which yields succeeds onlyall StreamsGroupDescription objects, if all the deletionsdescribes succeed.
     */
    public KafkaFuture<Map<String, KafkaFuture<Void>StreamsGroupDescription>> all() {
    }
 
    /**
     * Return a map from group id to futures which canyield be used to check the status of individual deletionsgroup descriptions.
     */
    public Map<String, KafkaFuture<Void>>KafkaFuture<StreamsGroupDescription>> deletedGroupsdescribedGroups() {
    }
}

...

StreamsGroupDescription

package org.apache.kafka.client.admin;
  
/**
 * Options for the {@link Admin#deleteStreamsGroups(Collection<String>, DeleteStreamsGroupsOptions)} call.
 * <p>
 * The API of this class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class DeleteStreamsGroupsOptions extends AbstractOptions<DeleteStreamsGroupsOptions> {
}

DescribeStreamsGroupsResult

package org.apache.kafka.clients.admin;
  
/**
 * The result of the {@link Admin#describeStreamsGroups(Collection<String>, DescribeStreamsGroupsOptions)} call.
 * <p>
 * The API of this class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class DescribeStreamsGroupsResult { A detailed description of a single streams group in the cluster.
 */
public class StreamsGroupDescription {

    public StreamsGroupDescription(
        final String groupId,
        final int groupEpoch,
        final 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.
     */
    public String groupId();
    
    /**
     * Return a future which yields all StreamsGroupDescription objects, if all the describes succeed.
  The epoch of the consumer group.
     */
    public int groupEpoch();
 
    /*/*
    public KafkaFuture<Map<String, StreamsGroupDescription>> all() {
 * The epoch of the target assignment.
     }*/
 
   public int targetAssignmentEpoch();

    /**
     * ReturnThe aepoch mapof fromthe groupcurrently id to futures which yield group descriptionsused topology.
     */
    public Map<String, KafkaFuture<StreamsGroupDescription>> describedGroupsint topologyEpoch();
  {  
    }
}

StreamsGroupDescription

 /**
     * A detailed descriptionlist of the members of a singlethe streams group.
 in  the cluster.
 */
public class StreamsGroupDescription {

 public Collection<StreamsGroupMemberDescription>  public StreamsGroupDescription(
members();

    /**
    final String* groupId,
A list of the subtopologies in the streams finalgroup.
 String topologyId,
   */
     finalpublic Collection<StreamsGroupSubtopologyDescription> subtopologies,();

    /**
    final Collection<StreamsGroupMemberDescription>* members,
The state of the streams group, or UNKNOWN finalif StreamsGroupStatethe state,
 is too new for us to parse.
 final Node coordinator,
  */
    public  final Set<AclOperation> authorizedOperationsGroupState groupState();

    /**
     * The id ofgroup coordinator, or null if the coordinator streamsis not groupknown.
     */
    public StringNode groupIdcoordinator();

    /**
     * TheauthorizedOperations idfor of the currently used topologythis group, or null if uninitialized that information is not known.
     */
    public StringSet<AclOperation> topologyIdauthorizedOperations();

    }

StreamsGroupMemberDescription

/**
     * A listdetailed description of a thesingle membersmember ofin the streams group.
 */
public class StreamsGroupMemberDescription  */{

    public Collection<StreamsGroupMemberDescription> membersStreamsGroupMemberDescription();

        final String memberId,
    /**    final int memberEpoch,
     * A list of the subtopologies in the streams group.
   final int currentAssignmentEpoch,
        final Optional<String> instanceId,
         */final Optional<String> rackId,
    public Collection<StreamsGroupSubtopologyDescription> subtopologies();

  final String /**clientId,
     *  The streamsfinal groupString stateclientHost,
    or UNKNOWN if the statefinal isint tootopologyEpoch,
 new for us to parse.
   final String */processId,
    public StreamsGroupState state();

  final Optional<Endpoint> /**userEndpoint,
     * The streams groupfinal coordinatorMap<String, orString> nullclientTags,
 if the coordinator is not known.
  final List<TaskOffset>  */taskOffsets,
    public Node coordinator();

  final List<TaskOffset> /**taskEndOffsets,
     *  authorizedOperations forfinal thisStreamsGroupMemberAssignment groupassignment,
  or  null if that information isfinal notStreamsGroupMemberAssignment known.targetAssignment,
     */
   final boolean isClassic
 public Set<AclOperation> authorizedOperations( );
}

StreamsGroupMemberDescription

    /**
 * A detailed description of* aThe singleid memberof in the group member.
 */
public  class StreamsGroupMemberDescription {*/

    public StreamsGroupMemberDescriptionString memberId();

    /**
    final String* memberId,The 
epoch of the group  member.
   final Optional<String> instanceId,*/
    public int memberEpoch();

  final String clientId,/**
     * The  final String clientHost, id of the instance, used for static membership, if available.
     */
   final public String topologyId,Optional<String> instanceId(); 
    
    final/**
 String processId,
   * The rack ID of finalthe Map<String, String> clientTags,group member.
     */
   final List<TaskOffset>public taskOffsets, Optional<String> rackId();
    
    /**
 final List<TaskOffset> taskEndOffsets,
  * The client id of the finalgroup StreamsGroupMemberAssignment assignment,
member.
     */
    finalpublic Optional<StreamsGroupMemberAssignment>String targetAssignmentclientId();

    /**
     * The idhost of the group member.
     */
    public String memberIdclientHost();

    /**
     * The idepoch of the instance, used for static membership, if available.group member. 
     */
    public Optional<String>int instanceIdmemberEpoch();

    /**
     * The client idepoch of the groupcurrent memberassignment.
     */
    public Stringint clientIdcurrentAssignmentEpoch();

     /**
     * The hostepoch of the group membertopology present on the client.
     */
    public Stringint clientHosttopologyEpoch();
 
      /**
     * TheIdentity id of the topology present on the clientstreams instance that may have multiple clients.
     */
    public String topologyIdprocessId(); 

    /**
     * IdentityUser-defined ofendpoint thefor streams instance that may have multiple clientsInteractive Queries.
     */
    public StringOptional<Endpoint> processIduserEndpoint();

     /**
     * Used for rack-aware assignment algorithm.
     */
    public Map<String, String> clientTags();

    /**
     * Cumulative offsets for tasks.
     */
    public List<TaskOffset> taskOffsets();

    /**
     * Cumulative task changelog end offsets for tasks.
     */
    public List<TaskOffset> taskEndOffsets();

    /**
     * The current assignment.
     */
    public StreamsGroupMemberAssignment assignment();

    /**
     * The target assignment.
     */
    public Optional<StreamsGroupMemberAssignment>StreamsGroupMemberAssignment targetAssignment();
    
    /**
     * The cumulative offset for one task.
     */
    public static class TaskOffset {

        public TaskOffset(final String subtopology, final int partition, final long offset);

        /**
      
    
    /**
       * The subtopologyflag identifier.
indicating whether a member      */
        public String subtopology();

        /**
         * The partition of the task.
   is classic.
      */
        public intboolean partitionisClassic();

         /**
         * The cumulative offset for (sum of offsets in all input partitions).one task.
     */
    public static class TaskOffset  */{

        public TaskOffset(final String subtopologyId, final int partition, final long offset();

        }
}

StreamsGroupMemberAssignment

/**
 * A description of the assignments of a specific* streamsThe groupsubtopology memberidentifier.
 */
public class StreamsGroupMemberAssignment {

    public StreamsGroupMemberAssignment(*/
        finalpublic List<TaskIds> activeTasks,String subtopologyId();

        final List<TaskIds> standbyTasks,/**
        final List<TaskIds> warmupTasks);

    /**
 * The partition of the task.
    * Active tasks for this client.*/
     */
    public List<TaskIds>int activeTaskspartition();

    /**
     /**
 Standby tasks for this client.
     */
 The cumulative offset public List<TaskIds> standbyTasks();
    
    /**
 (sum of offsets in all input partitions).
    * Warmup tasks for this client.*/
     */
    public List<TaskIds>long warmupTasksoffset();

    }
}

StreamsGroupMemberAssignment

/**
 * A description of *the Allassignments tasksof fora onespecific subtopologystreams ofgroup a member.
     */
    public static class TaskIdsStreamsGroupMemberAssignment {

        public TaskIdsStreamsGroupMemberAssignment(final
 String subtopology, final List<Integer> partitions);

   final     /**List<TaskIds> activeTasks,
        final *List<TaskIds> The sub-topology identifier.
standbyTasks,
        final List<TaskIds>  */warmupTasks
        public String subtopology();

        /**
         * TheActive partitions of the input topics processed by this member.
    tasks for this client.
     */
        public List<Integer>List<TaskIds> partitionsactiveTasks();

    }
}

StreamsGroupSubtopologyDescription

/**
 * A detailed description of a single subtopology
 */
public class StreamsGroupSubtopologyDescription {
/**
     * Standby tasks for this client.
     */
    public List<TaskIds> standbyTasks();
    
    /**
    public StreamsGroupSubtopologyDescription(
* Warmup tasks for this client.
   final String subtopology,*/
    public List<TaskIds> warmupTasks();

  final List<String> sourceTopics,/**
     * All  final List<String> repartitionSinkTopics,
tasks for one subtopology of a member.
     */
    public finalstatic Map<String,class TopicInfo>TaskIds stateChangelogTopics,{

        public TaskIds(final String Map<StringsubtopologyId, final TopicInfo>List<Integer> repartitionSourceTopicspartitions);

        /**
     * String to uniquely identify* theThe subtopology identifier.
         */
        public String subtopologysubtopologyId();

        /**
         * The partitions of the input topics theprocessed topologyby readsthis frommember.
         */
        public List<String>List<Integer> sourceTopicspartitions();
    }
}

StreamsGroupSubtopologyDescription

    /**
     * TheA repartitiondetailed topicsdescription theof topologya writessingle to.subtopology
     */
public class   public List<String> repartitionSinkTopics();StreamsGroupSubtopologyDescription {

    /**public StreamsGroupSubtopologyDescription(
     * The set offinal stateString changelogsubtopologyId,
 topics associated with this sub-topology.
   final List<String> */sourceTopics,
    public Map<String, TopicInfo> stateChangelogTopics();

 final   /**List<String> repartitionSinkTopics,
     * The set offinal sourceMap<String, topics that are internally created repartition topics.
 TopicInfo> stateChangelogTopics,
    */
    publicfinal Map<String, TopicInfo> repartitionSourceTopics();

    /**
     * Information about a topicString to uniquely identify the subtopology.
     */
    public static class TopicInfo {

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

        /**
         * The numbertopics ofthe partitionstopology inreads the topicfrom.
         */
        public intList<String> partitionssourceTopics();

    
    /**
     * The repartition topics *the Configurationstopology ofwrites the topicto.
         */
        public Map<String, String> topicConfigsList<String> repartitionSinkTopics();
    }

}

DescribeStreamsGroupsOptions

package org.apache.kafka.client.admin;
  

    /**
 * Options for {@link Admin#describeStreamsGroups(Collection<String>, DescribeStreamsGroupsOptions)}.
 *
 * The APIset of thisstate classchangelog istopics evolving,associated seewith {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class DescribeStreamsGroupsOptions extends AbstractOptions<DescribeStreamsGroupsOptions> {this subtopology.
     */
    public DescribeStreamsGroupsOptions includeAuthorizedOperations(boolean includeAuthorizedOperations);
 
    public boolean includeAuthorizedOperations();
}

ListStreamsGroupOffsetsResult

package org.apache.kafka.clients.admin;
  
/**
 * The result of the {@link Admin#listStreamsGroupOffsets(Map<String, ListStreamsGroupOffsetsSpec>, ListStreamsGroupOffsetsOptions)} call.
 * <p>
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();

    /**
     * Information Returnabout 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 a mapnumber of topic partitions to offsets forin the specified grouptopic.
     */
    public*/
 KafkaFuture<Map<TopicPartition, Long>> partitionsToOffset(String groupId) {
    }
}

ListStreamsGroupOffsetsOptions

package org.apache.kafka.client.admin;
  
public int partitions();

        /**
 * 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;
  
/**
 * Specification of Streams group offsets to list using Options 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 this class is evolving, 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, Long> orobjects, theif fullrequests setfor ofall Streamsthe groupgroups listingssucceed.
     */
    public KafkaFuture<Collection<StreamsGroupListing>>KafkaFuture<Map<String, Map<TopicPartition, Long>>> all() {
    }
 
    /**
     * ReturnsReturn a future which yields justa themap validof listings.
topic partitions to   */
    public KafkaFuture<Collection<StreamsGroupListing>> valid() {
    }
  
    /**
     * Returns a future which yields just the errors which occurredoffsets for the specified group.
     */
    public publicKafkaFuture<Map<TopicPartition, KafkaFuture<Collection<Throwable>>Long>> errorspartitionsToOffset(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 Specification of Streams group offsets to list using {@link Admin#listStreamsGroups(ListStreamsGroupsOptions Admin#listStreamsGroupOffsets(Map<String, ListStreamsGroupOffsetsSpec>, ListStreamsGroupOffsetsOptions)}.
 * <p>
 * The API of this class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class ListStreamsGroupsOptionsListStreamsGroupOffsetsSpec {
 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 exception is 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.

...

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

...

• KIP-848 originally envisioned that Kafka Streams continues to using a byte-array metadata protocol on top of regular consumer group RPCs, and use a client-side assignor, and provided a rough proposal how to do so. The described approach was never implemented. In this KIP, we propose to use broker-side assignment by default for several reasons (mostly already mentioned in the motivation section):

  • Client-side assignment makes it hard to debug and tune the assignment logic, where a simple parameter change requires redeploying of all streams clients.

  • Using opaque byte-arrays to exchange the required metadata makes it hard to understand the state of a streams group on the protocol level. One has to discover the information from logs spread across all client machines. Central mechanisms of the Kafka protocol, such as versioning of RPCs, needs to be reimplemented on the level of RPCs.

  • One could argue that the Kafka protocol is simpler without Kafka Streams specific RPCs, but we think the opposite is true. The KIP-848 consumer group protocol was extended with client-side assignment, byte-array metadata and metadata versioning just for Kafka Streams. Having the concerns of Kafka Streams dealt with in separate RPCs mainly introduces a clearer separation and gives the Kafka community the ability to simplify the upcoming consumer protocol.

• Instead of using a separate initialize RPCthe heartbeat RPC during joining to initialize the topology, we considered just using the heartbeat with memberEpoch=0 to initialize the Streams group and initiate the creation of all topics. This would have a number of downsides, such as that every member needs to send the initialization data every time they start (and the initialization data may be large already in the current schema and may grow larger in the future), and the heartbeat requiring the ACL for creating topics. Furthermore, since the initialization may take a certain amount of time, we’d likely still have to respond to the initial heartbeat without having fully initialized the group, leading to temporary “initializing” state for the group that we also get in this proposala dedicated group intialization RPC. Using the heartbeat RPC for initialization turned out to be simpler and we realized that the drawbacks that we wanted to avoid with the dedicated initialization RPC where not that significant. For example, we thought that we could save on data sent for intitialization with a dedicated intialization call. However, sending the topology metadata in the heartbeat when joining does not seem to be too much and we can ensure all clients use the same topology metadata. Also with two different calls we had the overhead to coordinate the initialization and the heartbeat requests.

• An earlier draft of this KIP proposed introducing optional client-side task assignment, that would allow customization of the task assignment similar to KIP-924. We decided to remove it from this KIP, to limit the scope of the KIP and not introduce features in the protocol whose use we cannot fully envision yet. The RPCs defined in this KIP are, however, defined in a way that they could easily be extended with client-side assignment in a follow-up KIP, possibly together with a public interface for adding custom task assignors on the client side.

...

• Topology updates

  • Validate topology compatibility broker-sideIntroduce manually bumped topology versioning in addition to automatically managed topology IDs

  • Automatically drain repartition topics on a topology update

• In addition to the existing shutdown-flag, a global pause/resume functionality could be built on top of the protocol

• There is an opportunity to improve EOSv2 in the new protocol. One may add the current transaction.id to the heartbeat request and a list of closing transactions.id that the client should fence in the heartbeat response to improve EOSv2 and avoid the need to rely on a low transaction.timeout.ms config.

...