Page History

...

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

...

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 all source topics and the 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

To avoid unintential 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 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 with a topology 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 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.

Info

title	Example

For example of a rolling bounce to upgrade the topology, let’s assume a sticky assignor with stateless tasks and no standby tasks and one thread per process for illustration.

Assume we have 3 clients A, B, and C with a uniform assignment (#1 #0 is the topology IDepoch):

A(#1#0): T1, T2, T3
B(#1#0): T4, T5, T6
C(#1#0): T7, T8, T9

We shut down C to update its topology. Once we shut down C, the assignment is the following:

A(#1#0): T1, T2, T3, T7
B(#1#0): T4, T5, T6, T8, T9

Then, C rejoins the group with topology ID #2epoch #1. This topology ID does not correspond to epoch bumps the group's topology IDepoch, so C will not get any tasks assigned. A and B get to keep their tasks.

A(#1): T1, T2, T3, T7
B(#1): T4, T5, T6, T8, T9
C(#2): -

C includes the information to and C also includes the information to initialize the new topology in its first heartbeat. Once the topology is initialized, from From now on, A and B will not get new tasks assigned, but C will get new tasks assigned because now #2 #1 is the group's topology IDepoch. The rest of the assignment is unchanged - just that A and B cannot get any new tasks, but they can retain some of the tasks they have. So since the sticky assignor balances out the assignment, we should end up with something like this:

A(#1#0): T2, T3, T7
B(#1#0): T5, T6, T8
C(#2#1): T1, T4, T9

B is shut down for the topology update. Since, again, A cannot get new tasks but C can, the new assignment is:

A(#1#0): T2, T3, T7
C(#2#1): T1, T4, T9, T5, T6, T8

Client B comes back with the new topology IDepoch, and gets new tasks assigned:

A(#1#0): T2, T3, T7
B(#2#1): T1, T4, T9
C(#2#1): T5, T6, T8

And so on.

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

...

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.
Source topics inconsistent
Condition: The source topics are inconsistent, if two source topics are supposed to be copartitioned according to the topology, but in the current topic metadata on the broker, the number of partitions for the two topics is different.
Behavior: The group will enter/remain in state NOT_READY. Heartbeat responses will indicate status SOURCE_TOPICS_INCONSISTENT . In the status detail, we specify at least one inconsistency.
Internal topics are inconsistent
Condition: One or more internal topics are inconsistent, for example, they are not copartition despite being part of a copartition group, or the number of partitions in a changelog topic does not correspond to the maximal number of source topic partition for that subtopology
Behavior: The group will enter/remain in state NOT_READY. Heartbeat responses will indicate status INTERNAL_TOPICS_INCONSISTENT . In the status detail, we specify at least one inconsistency.
Internal topics are missing
Condition: One or more internal topics are missing.
Behavior: If discovered during a heartbeat, the group coordinator will attempt to create the internal topics by sending a corresponding topic create request will be sent to the Kraft coordinator. There can only be one such request in-flight at a time, an appropriate back-off mechanism will be used to prevent too many attempts to create the topics. If the appropriate ACL for topic creation are not assigned to the principle executing the heartbeat, no such attempt will be made. The group will enter/remain in state NOT_READY . Heartbeat responses will indicate status MISSING_INTERNAL_TOPICS. In the status detail, we specify whether an attempt to create the topics was made, whether and why a previous attempt failed, whether sufficient ACLs to create the topics are available.
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 AK implementation will provide the following assignors

highly_available - Like Similar to the current HighAvailabilityTaskAssignor
sticky - Like similar to the current StickyTaskAssignor

...

The possible states of the streams group are EMPTY, ASSIGNING, RECONCILING , STABLE, NOT_READY, DEAD as for consumer groups.

...

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.

...

Code Block

language	yml
linenumbers	true

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

	{ "name":  "Topology", "type": "Topology", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "The topology data 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-topology. Deterministically generated from the topology" },
	        { "name": "SourceTopics", "type": "[]string", "versions": "0+",
	          "about": "The topics the topology reads from." },
	        { "name": "SourceTopicRegex", "type": "[]string", "versions": "0+", 
	           "about": "The regular expressions identifying topics the sub-topology reads from." },
	        { "name": "StateChangelogTopics", "type": "[]TopicInfo", "versions": "0+",
	          "about": "The set of state changelog topics associated with this sub-topology. Created automatically." },
	        { "name": "RepartitionSinkTopics", "type": "[]string", "versions": "0+",
	          "about": "The repartition topics the sub-topology writes to." }, 
	        { "name": "RepartitionSourceTopics", "type": "[]TopicInfo", "versions": "0+",
	          "about": "The set of source topics that are internally created repartition topics. Created automatically." },
	        { "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": "[]int32", "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." },
    { "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": "TopicConfigs", "type": "[]KeyValue", "versions": "0+", "nullableVersions": "0+", "default": "null",
        "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": "int32", "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." }
    ]}
  ]
}

...

GroupId must be non-empty.
Either MemberId is non-empty or MemberEpoch is 0.
MemberEpoch must be >= -2.
InstanceId, if not null, must be non-empty.
RebalanceTimeoutMs must be larger than zero in the first heartbeat request.
ActiveTasks, StandbyTasks and WarmupTasks have to be disjoint sets
Each element of ActiveTasks, StandbyTasks and WarmupTasks has to be a valid task ID in the topology initialized for the group ID.
ActiveTasks, StandbyTasks and WarmupTasks have to be non-null and empty when joining (member epoch is 0)
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:

...

Performs request validation.
If the member joins the group (i.e. member epoch is 0):
1. Look up or create the group.
2. If the group is initialized:
  1. Looks up the group
3. 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.
4. 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).
5. 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 write the new topology record to the offset topic
6. Creates the member.
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.
Updates information of the member if needed. The group epoch is incremented if there is any changeif there is any change.
If the topology or topic metadata changed, detect any topology / topic mismatches as described earlier in this document.
Reconcile the member assignments as explained earlier in this document.
If we find that internal topics are missing for group, we will send a corresponing create topic request to the controller.

Reponse Schema

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

...

Code Block

language	yml
linenumbers	true

{
  "apiKey": TBD,
  "type": "response",
  "name": "StreamsGroupHeartbeatResponse",
  "validVersions": "0",
  "flexibleVersions": "0+",
  // Supported errors:
  // - GROUP_AUTHORIZATION_FAILED (version 0+)
  // - GROUP_ID_NOT_FOUND (version 0+)
  // - NOT_COORDINATOR (version 0+)
  // - COORDINATOR_NOT_AVAILABLE (version 0+)
  // - COORDINATOR_LOAD_IN_PROGRESS (version 0+)
  // - INVALID_REQUEST (version 0+)
  // - UNKNOWN_MEMBER_ID (version 0+)
  // - FENCED_MEMBER_EPOCH (version 0+)
  // - UNRELEASED_INSTANCE_ID (version 0+)
  // - GROUP_MAX_SIZE_REACHED (version 0+)
  // - TOPIC_AUTHORIZATION_FAILED (version 0+) 
  // - CLUSTER_AUTHORIZATION_FAILED (version 0+)
  // - STREAMS_INVALID_TOPOLOGY (version 0+)
  // - STREAMS_INVALID_TOPOLOGY_EPOCH (version 0+)
  // - STREAMS_TOPOLOGY_FENCED (version 0+)
  "fields": [
    // Same as consumer group heart beat
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
      "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
    { "name": "ErrorCode", "type": "int16", "versions": "0+",
      "about": "The top-level error code, or 0 if there was no error" },
    { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "The top-level error message, or null if there was no error." },
    { "name": "MemberId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "The member id generated by the coordinator. Only provided when the member joins with MemberEpoch == 0." },
    { "name": "MemberEpoch", "type": "int32", "versions": "0+",
      "about": "The member epoch." },
    { "name": "HeartbeatIntervalMs", "type": "int32", "versions": "0+",
      "about": "The heartbeat interval in milliseconds." },
    { "name": "AcceptableRecoveryLag", "type": "int32", "versions": "0+",
      "about": "The maximal lag a warm-up task can have to be considered caught-up." },
    { "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": "GroupTopologyIdTaskOffsetIntervalMs", "type": "stringint32", "versions": "0+", "nullableVersions": "0+", "default": null,
      "about": "The currentinterval IDin ofwhich the topology for task changelog offsets on a client are updated on the groupbroker. Null if unchanged since last heartbeatThe offsets are sent with the next heartbeat after this time has passed." },
 
      { "name": "Status", "type":  "[]Status", "versions":  "0+",  "nullableVersions": "0+", "default": "null", 
      "about": "Indicate zero or more status for the group.  Null if unchanged since last heartbeat." },

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

    // IQ-related information
    { "name": "PartitionsByUserEndpoint", "type": "[]EndpointToPartitions", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "Global assignment information used for IQ. Null if unchanged since last heartbeat." ,
      "fields": [
        { "name": "UserEndpoint", "type": "Endpoint", "versions": "0+",
          "about": "User-defined endpoint to connect to the node" },
        { "name": "Partitions", "type": "[]TopicPartition", "versions": "0+",
          "about": "All partitions available on the node" }
      ]
    }
  ],
  "commonStructs": [
    { "name": "Status", "versions": "0+", "fields": [
      // Possible status codes
      //  0 - 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 - INCONSISTENT_SOURCE_TOPICS   - One or more source topics are inconsistent, for example, they are not copartition despite being 
      //                                     part of a copartition group.
      //                                     Inconsistent topics are indicated in the StatusDetail.
      //  3 - INCONSISTENT_INTERNAL_TOPICS - One or more internal topics are inconsistent, for example, they are not copartition despite being
      //                                     part of a copartition group, or the number of partitions in a changelog topic does not correspond
      //                                     to the maximal number of source topic partition for that subtopology. 
      //                                     Inconsistent topics are indicated in the StatusDetail. 
      //  4 - MISSING_INTERNAL_TOPICS      - One or more internal topics are missing. 
      //                                     Missing topics are indicated in the StatusDetail. 
      //                                     The group coordinator will attempt to create all missing internal topics, if any errors occur during
      //                                     topic creation, this will be indicated in StatusDetail.
      //  5 - SHUTDOWN_APPLICATION         - A client requested the shutdown of the whole application.
      { "name": "StatusCode", "type": "int8", "versions": "0+",
        "about": "A code to indicate that a particular status is active for the group membership" },
      { "name": "StatusDetail", "type": "string", "versions": "0+",
        "about": "A string representation of the status." }
    ]},
    { "name": "TopicPartition", "versions": "0+", "fields": [
      { "name": "Topic", "type": "string", "versions": "0+",
        "about": "topic name" },
      { "name": "Partitions", "type": "[]int32", "versions": "0+",
        "about": "partitions" }
    ]},
    { "name": "TaskIds", "versions": "0+", "fields": [
      { "name": "SubtopologyId", "type": "string", "versions": "0+",
        "about": "The subtopology identifier." },
      { "name": "Partitions", "type": "[]int32", "versions": "0+",
        "about": "The partitions of the input topics processed by this member." }
    ]}
    { "name": "Endpoint", "versions": "0+", "fields": [
      { "name": "Host", "type": "string", "versions": "0+",
        "about": "host of the endpoint" },
      { "name": "Port", "type": "int32", "versions": "0+",
        "about": "port of the endpoint" }
    ]}
  ]
}

...

Code Block

language	yml
linenumbers	true

{
  "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": "TopologyIdTopologyEpoch", "type": "string", "versions": "0+",
          "about": "The IDepoch of the currently initialized topology for this group." }, 
         { "name": "Topology", "type": "[]Subtopology", "versions": "0+",
          "about": "The resolved sub-topologies of the streams application. This contains the configured sub-topologies, where the number of partitions are set and any regular expressions are resolved to actual topics. Null if the group is uninitialized, source topics are missing or inconsistent.",
          "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": "RepartitionSinkTopics", "type": "[]string", "versions": "0+",
              "about": "The topics the topology writes to." },
            { "name": "StateChangelogTopics", "type": "[]TopicInfo", "versions": "0+",
              "about": "The set of state changelog topics associated with this subtopology. Created automatically." },
            { "name": "RepartitionSourceTopics", "type": "[]TopicInfo", "versions": "0+",
              "about": "The set of source topics that are internally created repartition topics. Created automatically." }
          ]
        },
        { "name": "Members", "type": "[]Member", "versions": "0+",
          "about": "The members.",
          "fields": [
            { "name": "MemberId", "type": "string", "versions": "0+",
              "about": "The member ID." },
            { "name": "MemberEpoch", "type": "int32", "versions": "0+",
              "about": "The member epoch." },
            { "name": "InstanceId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
              "about": "The member instance ID for static membership." },
            { "name": "RackId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
              "about": "The rack ID." },

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

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

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

            { "name": "Assignment", "type": "Assignment", "versions": "0+",
              "about": "The current assignment." },
            { "name": "TargetAssignment", "type": "Assignment", "versions": "0+",
              "about": "The target assignment." }
          ]},
        { "name": "AuthorizedOperations", "type": "int32", "versions": "0+", "default": "-2147483648",
          "about": "32-bit bitfield to represent authorized operations for this group." }
      ]
    }
  ],
  "commonStructs": [
    { "name": "TaskOffset", "versions": "0+", "fields": [
      { "name": "SubtopologyId", "type": "string", "versions": "0+",
        "about": "The subtopology identifier." },
      { "name": "Partition", "type": "int32", "versions": "0+",
        "about": "The partition." },
      { "name": "Offset", "type": "int64", "versions": "0+",
        "about": "The offset." }
    ]},
    { "name": "TopicPartitions", "versions": "0+", "fields": [
      { "name": "TopicId", "type": "uuid", "versions": "0+",
        "about": "The topic ID." },
      { "name": "TopicName", "type": "string", "versions": "0+", "entityType": "topicName",
        "about": "The topic name." },
      { "name": "Partitions", "type": "[]int32", "versions": "0+",
        "about": "The partitions." }
    ]},
    { "name": "Assignment", "versions": "0+", "fields": [
      { "name": "ActiveTasks", "type": "[]TaskIds", "versions": "0+",
        "about": "Active tasks for this client." },
      { "name": "StandbyTasks", "type": "[]TaskIds", "versions": "0+",
        "about": "Standby tasks for this client." },
      { "name": "WarmupTasks", "type": "[]TaskIds", "versions": "0+",
        "about": "Warm-up tasks for this client. " }
    ]},
    { "name": "TaskIds", "versions": "0+", "fields": [
      { "name": "SubtopologyId", "type": "string", "versions": "0+",
        "about": "The subtopology identifier." },
      { "name": "Partitions", "type": "[]int32", "versions": "0+",
        "about": "The partitions of the input topics processed by this member." }
    ]},
    { "name": "KeyValue", "versions": "0+", "fields": [
      { "name": "Key", "type": "string", "versions": "0+",
        "about": "key of the config" },
      { "name": "Value", "type": "string", "versions": "0+",
        "about": "value of the config" }
    ]},
    { "name": "TopicInfo", "versions": "0+", "fields": [
      { "name": "Name", "type": "string", "versions": "0+",
        "about": "The name of the topic." },
      { "name": "Partitions", "type": "int32", "versions": "0+",
        "about": "The number of partitions in the topic. Can be 0 if no specific number of partitions is enforced. Always 0 for changelog topics." },
      { "name": "TopicConfigs", "type": "[]KeyValue", "versions": "0+", "nullableVersions": "0+", "default": "null",
        "about": "Topic-level configurations as key-value pairs."
      }
    ]}
  ]
}

...

Code Block

language	yml
linenumbers	true

{
  "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." },
    { "name": "RebalanceTimeoutMs", "type": "int32", "versions": "0+", "default": -1,
      "about": "The rebalance timeout." },

    { "name": "TopologyIdTopologyEpoch", "type": "stringint", "versions": "0+",
      "about": "The IDepoch of the topology. Must be non-empty." },

    { "name": "ProcessId", "type": "string", "versions": "0+",
      "about": "Identity of the streams instance that may have multiple consumers." },
    { "name": "UserEndpoint", "type": "Endpoint", "versions": "0+",
      "about": "User-defined endpoint for running interactive queries on this instance." },
    { "name": "ClientTags", "type": "[]KeyValue", "versions": "0+",
      "about": "Used for rack-aware assignment algorithm." }
  ],
  "commonStructs": [
    { "name": "Endpoint", "versions": "0+", "fields": [
      { "name": "Host", "type": "string", "versions": "0+",
        "about": "host of the endpoint" },
      { "name": "Port", "type": "int32", "versions": "0+",
        "about": "port of the endpoint" }
    ]},
    { "name": "KeyValue", "versions": "0+",
      "fields": [
        { "name": "Key", "type": "string", "versions": "0+",
          "about": "key of the config" },
        { "name": "Value", "type": "string", "versions": "0+",
          "about": "value of the config" }
      ]
    }
  ]
}

...

Code Block

language	yml
linenumbers	true

{
  "type": "data",
  "name": "StreamsGroupTopologyValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "TopologyIdTopologyEpoch", "type": "stringint", "versions": "0+",
      "about": "The IDepoch of the topology. Must be non-emptyzero. " },
    { "name":  "Topology", "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": "The regular expressions identifying topics the topology 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-topology 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": "[]int32", "versions": "0+",
              "about": "The topics the topology reads from. Index into the array on the subtopology level." },
            { "name": "SourceTopicRegex", "type": "[]int32", "versions": "0+",
              "about": "Regular expressions identifying topics the subtopology reads from. Index into the array on the subtopology level." },
            { "name": "RepartitionSourceTopics", "type": "[]int32", "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": "TopicConfigs", "type": "[]TopicConfig", "versions": "0+", "nullableVersions": "0+", "default": "null",
        "about": "Topic-level configurations as key-value pairs."
      }
    ]}
  ]
}

...

Name	Type	Default	Doc
group.protocol	enum	classic	A flag which indicates if the new protocol should be used or not. It could be: classic or streams
topology.epoch	int	0	The epoch of the topology for the streams group. Ignored if `group.protocol=classic`

Deprecations

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

...

Code Block

language	java
linenumbers	true
collapse	true

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

    public StreamsGroupDescription(
        final String groupId,
        final Stringint topologyIdtopologyEpoch,
        final Collection<StreamsGroupSubtopologyDescription> subtopologies,
        final Collection<StreamsGroupMemberDescription> members,
        final StreamsGroupState state,
        final Node coordinator,
        final Set<AclOperation> authorizedOperations);

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

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

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

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

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

...

Code Block

language	java
linenumbers	true
collapse	true

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

    public StreamsGroupMemberDescription(
        final String memberId, 
        final Optional<String> instanceId,
        final String clientId,
        final String clientHost, 
        final Stringint topologyIdtopologyEpoch, 
        final String processId,
        final Map<String, String> clientTags,
        final List<TaskOffset> taskOffsets, 
        final List<TaskOffset> taskEndOffsets,
        final StreamsGroupMemberAssignment assignment,
        final Optional<StreamsGroupMemberAssignment> targetAssignment);

    /**
     * The id of the group member.
     */
    public String memberId();

    /**
     * The id of the instance, used for static membership, if available.
     */
    public Optional<String> instanceId();

    /**
     * The client id of the group member.
     */
    public String clientId();

    /**
     * The host of the group member.
     */
    public String clientHost();

    /**
     * The idepoch of the topology present on the client.
     */
    public Stringint topologyIdtopologyEpoch();
 
      /**
     * Identity of the streams instance that may have multiple clients.
     */
    public String processId();

    /**
     * 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> targetAssignment();
    
    /**
     * The cumulative offset for one task.
     */
    public static class TaskOffset {

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

        /**
         * The subtopology identifier.
         */
        public String subtopology();

        /**
         * The partition of the task.
         */
        public int partition();

        /**
         * The cumulative offset (sum of offsets in all input partitions).
         */
        public long offset();
    }
}

...

Code Block

language	java
linenumbers	true
collapse	true

/**
 * A description of the assignments of a specific streams group member.
 */
public class StreamsGroupMemberAssignment {

    public StreamsGroupMemberAssignment(
        final List<TaskIds> activeTasks,
        final List<TaskIds> standbyTasks,
        final List<TaskIds> warmupTasks);

    /**
     * Active tasks for this client.
     */
    public List<TaskIds> activeTasks();

    /**
     * Standby tasks for this client.
     */
    public List<TaskIds> standbyTasks();
    
    /**
     * Warmup tasks for this client.
     */
    public List<TaskIds> warmupTasks();

    /**
     * All tasks for one subtopology of a member.
     */
    public static class TaskIds {

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

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

        /**
         * The partitions of the input topics processed by this member.
         */
        public List<Integer> partitions();
    }
}

...

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.

...

Space shortcuts

Child pages

Versions Compared

Old Version 57

New Version 58

Key

Topology updates

Topology updates

Handling topic topology mismatches

Handling topic topology mismatches

StreamsGroupHeartbeat

Reponse Schema

Deprecations