Page History

...

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

Streams groups

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

Joining, leaving and staying in a Streams group

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

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

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

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

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

• A unique ID of the processing topology. The ID is defined by the client, and will be the hash 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).

...

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

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

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

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

Initializing and updating topology metadata on the broker

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

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

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

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

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

The data metadata sent to the broker via the StreamsGroupInitialize RPC heartbeat is persisted as a new record type in the __consumer_offsets topic and can be retrieved in subsequent requests. If one or more internal topics do not exist, the group coordinator invokes an asynchronous synchronous initialization. Only after the topology metadata is initialized and the internal topics are created, the group coordinator starts assigning tasks to the members of the group. The StreamsGroupInitializeResponse only indicates whether an error occurred or not.

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

...

Topology updates can be initiated by any member upon joining. Whenever a streams Streams member joins the group (whether a new member, or a static member after a restart) with a topology ID that is different from different from the current topology of the streams group, the group coordinator will ask one of the new members to reinitialize re-initializes the broker-side topology with the topology metadata contained in the first heartbeat of the Streams member. If the topology ID of an existing member does not correspond to the current topology ID of the group, this will be indicated in the heartbeat response sent to that member (for informational purposes, e.g., logging a warning).

...

It is up to the client implementation how topology IDs are defined. The protocol and the broker only requires that two clients with the same topology ID should have the same topology. In the initial implementation, the topology ID will be derived from the topology by deterministic hashing. However, it would also be possible to derive an implementation where topology IDs are explicitely 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.

We also allow the StreamsGroupInitialize to be invoked by a client that is not a member of the group and was not asked to initialize the group. This can also be used to create an empty initialized group. However, default mode is that a group is initialized by one of the clients after joining. Adding a strict 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.

...

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

• The topology metadata that with which the topology was initialized provided through the StreamsGroupInitialize APIby the first heartbeat with the new topology.

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

• The current target assignment generated by the assignor.

...

• 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 the topology is updated through the StreamsGroupInitialize API.
  • When a member joins or leaves the group.
  • When a member is fenced or removed from the group by the group coordinator.
  • When the partition metadata is updated. For instance when a new partition is added or a new topic matching the subscribed topics is created.
  • When a member with an assigned warm-up task reports a task changelog offset and task changelog end offset whose difference is less that acceptable.recovery.lag.
  • When a member updates its topology IDmetadata, 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.

...

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

...

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

...

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

Cumulative task changelog end offsets

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

...

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

...

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

Global & Dynamic Group Configuration

...

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

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

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

...

StreamsGroupHeartbeat

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

...

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

Request Schema

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

{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker"],
  "name": "StreamsGroupInitializeRequestStreamsGroupHeartbeatRequest",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "GroupId", "type": "string", "versions": "0+", "entityType": "groupId",
      "about": "The group identifier." },
    { "name": "TopologyIdMemberId", "type": "string", "versions": "0+",
      "about": "The member ID generated by the coordinator. The member ID must be kept during the entire lifetime of the topologymember." },
      { "name":  "TopologyMemberEpoch", "type": "[]Subtopologyint32", "versions": "0+",
      "about": "The current sub-topologies of the streams application.",
      "fields": [
        { "name": "SubtopologyIdmember 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+",
          "about"nullableVersions": "0+", "default": "String to uniquely identify the sub-topology. Deterministically generated from the topology" },
   null",
     { "nameabout": "SourceTopics", "type": "[]string", "versions": "0+",
          "about": "The topics the topology reads fromnull if not provided or if it didn't change since the last heartbeat; the instance ID for static membership otherwise." },
        { "name": "SourceTopicRegexRackId", "type": "string", "versions": "0+",  "nullableVersions": "0+", "default": "null",
           "about": "The regular expressions identifying topics the sub-topology reads from. null if not providednull if not provided or if it didn't change since the last heartbeat; the rack ID of consumer otherwise." },
        { "name": "StateChangelogTopicsRebalanceTimeoutMs", "type": "[]TopicInfoint32", "versions": "0+",
   "default": -1,
       "about": "The set of state changelog topics associated with this sub-topology. Created automatically." },
        -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":  "RepartitionSinkTopicsTopology", "type": "[]stringTopology", "versions": "0+",
 "nullableVersions": "0+", "default": null,
      "about": "The topology repartitiondata topicsof the sub-topology writes to." }, 
        { "name": "RepartitionSourceTopics", "type": "[]TopicInfo", "versions": "0+",
          "about": "The set of source topics that are internally created repartition topics. Created automatically." }
      ]
    }
  ],
  "commonStructs": [
    { "name": "TopicConfig", "versions": "0+",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": "keyTopologyId", "type": "string", "versions": "0+",
        	  "about": "The keyID of the topic-level configuration." },
         topology. Used to check if the topology corresponds to the topology initialized on the brokers." },
		{ "name": "valueSubtopologies", "type": "string[]Subtopology", "versions": "0+",
        	  "about": "The valuesub-topologies of the topic-level configuration," }
      ]
    },
 streams application.",
	   { "name": "TopicInfo", "versions": "0+", "fields": [
	        { "name": "NameSubtopologyId", "type": "string", "versions": "0+",
	          "about": "TheString nameto ofuniquely identify the topic.sub-topology. Deterministically generated from the topology" },
	        { "name": "PartitionsSourceTopics", "type": "int32[]string", "versions": "0+",
	          "about": "The number of partitions intopics the topic.topology Can be 0 if no specific number of partitions is enforced. Always 0 for changelog topicsreads from." },
	        { "name": "TopicConfigsSourceTopicRegex", "type": "[]TopicConfigstring", "versions": "0+", "nullableVersions": "0+", "default": "null",
	             "about": "Topic-level configurations as key-value pairs."
The regular expressions identifying topics the sub-topology reads from. null if not provided." },
	       }
    ]}
  ]
}

Required ACL

• READ on the group

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

• DESCRIBE_CONFIGS on all topics included in the message

Request Validation

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

• GroupId must be non-empty.

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

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

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

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

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

Request Handling

When the group coordinator handles a StreamsGroupInitialize request:

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

• Creates all required internal topics.

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

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

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

Response Schema

{
  "apiKey": TBD,
  "type": "response",
  "name": "StreamsGroupInitializeResponse",
  "validVersions": "0",
  "flexibleVersions": "0+",
  // Supported errors:
  // - GROUP_AUTHORIZATION_FAILED (version 0+)
  // - GROUP_ID_NOT_FOUND (version 0+)
  // - NOT_COORDINATOR (version 0+)
  // - COORDINATOR_NOT_AVAILABLE (version 0+)
  // - COORDINATOR_LOAD_IN_PROGRESS (version 0+)
  // - INVALID_REQUEST (version 0+)
  // - TOPIC_AUTHORIZATION_FAILED (version 0+) 
  // - CLUSTER_AUTHORIZATION_FAILED (version 0+)
  // - STREAMS_INVALID_TOPOLOGY (version 0+)
  // - STREAMS_MISSING_SOURCE_TOPICS (version 0+)
  // - STREAMS_INCONSISTENT_INTERNAL_TOPICS (version 0+)
  "fields": [ { "name": "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": "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": "ThrottleTimeMsProcessId", "type": "int32string", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "TheIdentity duration in milliseconds for which of the requeststreams wasinstance throttledthat duemay tohave amultiple quota violation, or zeroconsumers. Null if theunchanged requestsince did not violate any quotalast heartbeat." },
    { "name": "ErrorCodeUserEndpoint", "type": "int16Endpoint", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "The top-level error code, or 0User-defined endpoint for Interactive Queries. Null if thereunchanged wassince nolast errorheartbeat." },
    { "name": "ErrorMessageClientTags", "type": "string[]KeyValue", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "TheUsed for toprack-levelaware errorassignment message, or nullalgorithm. Null if thereunchanged wassince nolast errorheartbeat." },

    { "name": "TaskOffsets", "type": "[]
}

StreamsGroupHeartbeat

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

Request Schema

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

{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker"],
  "name": "StreamsGroupHeartbeatRequest",
  "validVersions": "0",
  "flexibleVersionsTaskOffset", "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+",
  "fieldsdefault": ["null",
    {  "nameabout": "GroupId", "type": "string", "versions": "0+", "entityType": "groupId",
      "about": "The group identifierCumulative 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": "MemberIdShutdownApplication", "type": "stringbool", "versions": "0+", "default": false,
      "about": "TheWhether memberall idStreams generatedclients byin the coordinator. The member id must be kept during the entire lifetime of the member." },
     group should shut down." }
  ],

  "commonStructs": [
    { "name": "KeyValue", "versions": "0+", "fields": [
      { "name": "MemberEpochKey", "type": "int32string", "versions": "0+",
        "about": "Thekey currentof memberthe epoch; 0 to join the group; -1 to leave the group; -2 to indicate that the static member will rejoin." },
    { "config" },
      { "name": "InstanceIdValue", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",

        "about": "null if not provided or if it didn't change sincevalue of the last heartbeat; the instance ID for static membership otherwise." config" }
    ]},
     { "name": "RackIdTopicInfo", "typeversions": "string0+", "fields": [
      { "versionsname": "0+Name",  "nullableVersionstype": "0+string", "defaultversions": "null0+",
        "about": "nullThe if not provided or if it didn't change since the last heartbeat; the rack ID of consumer otherwisename of the topic." },
      { "name": "RebalanceTimeoutMsPartitions", "type": "int32", "versions": "0+", "default": -1,

        "about": "-1The ifnumber itof didn't change sincepartitions in the lasttopic. heartbeat;Can thebe maximum0 timeif inno millisecondsspecific thatnumber theof coordinatorpartitions willis waitenforced. onAlways the0 memberfor to revoke its partitions otherwisechangelog topics." },

      { "name": "TopologyIdTopicConfigs", "type": "string[]KeyValue", "versions": "0+", "nullableVersions": "0+", "default": "null",
        "about": "TheTopic-level IDconfigurations of the topology. Only sent when memberEpoch = 0, must be non-empty. Null otherwise. Used to check if topology corresponds to server-topology. " },

   as key-value pairs."
      }
    ]},
    { "name": "ActiveTasksEndpoint", "typeversions": "[]TaskIds", "versions0+", "fields": [
      { "name": "0+Host", "nullableVersionstype": "0+string", "defaultversions": "null0+",
        "about": "Currentlyhost ownedof activethe tasks for this client. Null if unchanged since last heartbeat." },
endpoint" },
      { "name": "StandbyTasksPort", "type": "[]TaskIdsint32", "versions": "0+", "nullableVersions": "0+", "default": "null",

        "about": "Currentlyport ownedof standbythe tasks for this client. Null if unchanged since last heartbeat." endpoint" }
    ]},
    { "name": "WarmupTasksTaskOffset", "typeversions": "[]TaskIds0+", "versions"fields": [
      { "name": "0+Subtopology", "nullableVersionstype": "0+string", "defaultversions": "null0+",
        "about": "CurrentlyThe owned warmsub-up tasks for this client. Null if unchanged since last heartbeat.topology identifier." },

      { "name": "ProcessIdPartition", "type": "stringint32", "versions": "0+",
        "nullableVersionsabout": "0+", "default": "null"The partition." },
      { "aboutname": "Identity of the streams instance that may have multiple consumers. Null if unchanged since last heartbeat." Offset", "type": "int64", "versions": "0+",
        "about": "The offset." }
    ]},
    { "name": "UserEndpointTaskIds", "typeversions": "Endpoint0+", "fields": [
      { "versionsname": "0+Subtopology", "nullableVersionstype": "0+string", "defaultversions": "null0+",
        "about": "UserThe sub-defined endpoint for Interactive Queries. Null if unchanged since last heartbeattopology identifier." },
      { "name": "ClientTagsPartitions", "type": "[]KeyValueint32", "versions": "0+", "nullableVersions": "0+", "default": "null",

        "about": "UsedThe forpartitions rack-awareof assignmentthe algorithm.input Nulltopics ifprocessed unchangedby sincethis last heartbeatmember." },

    { "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": "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" }
    ]},
    { "name": "TaskOffset", "versions": "0+", "fields": [
      { "name": "Subtopology", "type": "string", "versions": "0+",
        "about": "The sub-topology 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": "Subtopology", "type": "string", "versions": "0+",
        "about": "The sub-topology identifier." },
      { "name": "Partitions", "type": "[]int32", "versions": "0+",
        "about": "The partitions of the input topics processed by this member." }
    ]}
  ]
}

Required ACL

• Read Group

Request Validation

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

• GroupId must be non-empty.

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

• MemberEpoch must be >= -2.

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

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

• ActiveTasks, StandbyTasks and WarmupTasks have to be disjoint sets

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

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

Request Handling

When the group coordinator handles a StreamsGroupHeartbeat request:

]}
  ]
}

Required ACL

• READ on group

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

• DESCRIBE_CONFIGS on all topics included in the message

Request Validation

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

• GroupId must be non-empty.

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

• MemberEpoch must be >= -2.

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

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

• ActiveTasks, StandbyTasks and WarmupTasks have to be disjoint sets

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

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

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

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

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

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

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

STREAMS_MISSING_SOURCE_TOPICS is returned if there are source topics missing during the initialization of the topology. Also returned if the source topic regular expression matched no topics.

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

Request Handling

When the group coordinator handles a StreamsGroupHeartbeat request:

1. Performs request validation.

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

   1. If the group is initialized:
      1. Looks up the group
   2. If the group is not initialized:
      1. Creates group and initializes the topology by creating all required internal topics.
      2. Writes the topology, keyed with the GroupId to the consumer offset topic. Existing records will be overwritten.
   3. GROUP_ID_NOT_FOUND is returned if the group ID is associated with a group type that is not streams or classic (the latter will be allowed for migration).
   4. Creates the member.
3. If the member is already part of the group (i.e. member epoch is greater than 0):Looks up the group.
   1. Looks up the group.
   1. GROUP_ID_NOT_FOUND is returned if the group ID does not exist
   , or the group ID is associated with a group type that is not streams or classic (the latter will be allowed for migration).Creates the member should the member epoch be zero, otherwise checks whether it exists. If it
   1. anymore.
   2. If the member does not exist, returns UNKNOWN_MEMBER_ID
   is returned.
   2. Checks whether the member epoch matches the member epoch in its current assignment. FENCED_MEMBER_EPOCH is returned otherwise. The member is also removed from the group.
      • There is an edge case here. When the group coordinator transitions a member to its target epoch, the heartbeat response with the new member epoch may be lost. In this case, the member will retry with the member epoch that
   he
   1. • it knows about and its request will be rejected with a FENCED_MEMBER_EPOCH. This will be handled as in KIP-848.
4. Updates information of the members informations member if anyneeded. The group epoch is incremented if there is any change.
5. Reconcile the member assignments as explained earlier in this document. 

Reponse Schema

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

...

{
  "apiKey": TBD,
  "type": "response",
  "name": "StreamsGroupHeartbeatResponse",
  "validVersions": "0",
  "flexibleVersions": "0+",
  // Supported errors: 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+)
  // - GROUPUNKNOWN_AUTHORIZATIONMEMBER_FAILEDID (version 0+)
  // - NOTFENCED_MEMBER_COORDINATOREPOCH (version 0+)
  // - COORDINATORUNRELEASED_NOTINSTANCE_AVAILABLEID (version 0+)
  // - COORDINATORGROUP_LOADMAX_INSIZE_PROGRESSREACHED (version 0+)
  // - INVALIDTOPIC_AUTHORIZATION_REQUESTFAILED (version 0+) 
  // - UNKNOWNCLUSTER_MEMBERAUTHORIZATION_IDFAILED (version 0+)
  // - FENCEDSTREAMS_MEMBERINVALID_EPOCHTOPOLOGY (version 0+)
  // - UNRELEASEDSTREAMS_MISSING_INSTANCESOURCE_IDTOPICS (version 0+)
  // - GROUPSTREAMS_MAXINCONSISTENT_SIZEINTERNAL_REACHEDTOPICS (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": "ShouldInitializeTopologyTaskOffsetIntervalMs", "type":  "boolint32", "versions":  "0+", "default": false,
      "about": "true if this streams application should initialize the topologyThe 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": "GroupTopologyId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": null,
      "about": "The current ID of the topology for the group. Null if unchanged since last heartbeat, empty if group is still initializing." },

    { "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 - GROUP_INITIALIZING      - The group is not initialized.
      //  1 - INCONSISTENT_TOPOLOGY   - The topology ID supplied is inconsistent with the topology for this streams group.
      //  21 - MISSING_SOURCE_TOPICS   - One or more source topics do not exist. Missing topics are indicated in the StatusDetail.
      //  32 - MISSING_INTERNAL_TOPICS - One or more internal topics do not exist. Missing topics are indicated in the StatusDetail.
      //  43 - 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": "Subtopology", "type": "string", "versions": "0+",
        "about": "The subtopology identifier." },
      { "name": "Partitions", "type": "[]int32", "versions": "0+",
        "about": "The partitions of the input topics processed by this member." }
    ]}
    { "name": "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" }
    ]}
  ]
}

StreamsGroupDescribe API

Request Schema

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

Request Validation

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

INVALID_GROUP_ID is returned should the group ID be empty.

Request Handling

When the group coordinator handle a StreamsGroupDescribeRequest request:

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

• Looks up the groups and returns the response.

Response Schema

{
  "apiKey": TBD,
  "type": "response",
  "name": "StreamsGroupDescribeResponse",
  "validVersions": "0",
  "flexibleVersions": "0+",
  // Supported errors:
  // - GROUP_AUTHORIZATION_FAILED (version 0+)
  // - NOT_COORDINATOR (version 0+)
  // - COORDINATOR_NOT_AVAILABLE (version 0+)
  // - COORDINATOR_LOAD_IN_PROGRESS (version 0+)
  // - INVALID_REQUEST (version 0+)
  // - INVALID_GROUP_ID (version 0+)
  // - GROUP_ID_NOT_FOUND (version 0+)
  "fields": [
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
      "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
    { "name": "Groups", "type": "[]DescribedGroup", "versions": "0+",
      "about": "Each described group.",
      "fields": [
        { "name": "ErrorCode", "type": "int16", "versions": "0+",
          "about": "The describe error, or 0 if there was no error." },
        { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
          "about": "The top-level error message, or null if there was no error." },
        { "name": "GroupId", "type": "string", "versions": "0+", "entityType": "groupId",
          "about": "The group ID string." },
        { "name": "GroupState", "type": "string", "versions": "0+",
          "about": "The group state string, or the empty string." },
        { "name": "GroupEpoch", "type": "int32", "versions": "0+",
          "about": "The group epoch." },
        { "name": "AssignmentEpoch", "type": "int32", "versions": "0+",
          "about": "The assignment epoch." },
        { "name": "TopologyId", "type": "string", "versions": "0+",
          "about": "The ID of the currently initialized topology for this group." }, 
        { "name": "Topology", "type": "[]Subtopology", "versions": "0+",
          "about": "The sub-topologies of the streams application.",
          "fields": [
            { "name": "Subtopology", "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+", "nullableVersions": "0+", "default": "null",
               "about": "The regular expressions identifying topics the topology reads from. null if not provided." },
            { "name": "RepartitionSinkTopics", "type": "[]string", "versions": "0+",
              "about": "The topics the topology writes to." },
            { "name": "StateChangelogTopics", "type": "[]TopicInfo", "versions": "0+",
              "about": "The set of state changelog topics associated with this subtopology. Created automatically." },
            { "name": "RepartitionSourceTopics", "type": "[]TopicInfo", "versions": "0+",
              "about": "The set of source topics that are internally created repartition topics. Created automatically." }
          ]
        },
        { "name": "Members", "type": "[]Member", "versions": "0+",
          "about": "The members.",
          "fields": [
            { "name": "MemberId", "type": "string", "versions": "0+",
              "about": "The member ID." },
            { "name": "MemberEpoch", "type": "int32", "versions": "0+",
              "about": "The member epoch." },
            { "name": "InstanceId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
              "about": "The member instance ID for static membership." },
            { "name": "RackId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
              "about": "The rack ID." },

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

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

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

            { "name": "Assignment", "type": "Assignment", "versions": "0+",
              "about": "The current assignment." },
            { "name": "TargetAssignment", "type": "Assignment", "versions": "0+",
              "about": "The target assignment." }
          ]},
        { "name": "AuthorizedOperations", "type": "int32", "versions": "0+", "default": "-2147483648",
          "about": "32-bit bitfield to represent authorized operations for this group." }
      ]
    }
  ],
  "commonStructs": [
    { "name": "TaskOffset", "versions": "0+", "fields": [
      { "name": "Subtopology", "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": "Subtopology", "type": "string", "versions": "0+",
        "about": "The subtopology identifier." },
      { "name": "Partitions", "type": "[]int32", "versions": "0+",
        "about": "The partitions of the input topics processed by this member." }
    ]},
    { "name": "KeyValue", "versions": "0+", "fields": [
      { "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."
      }
    ]}
  ]
}

...

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

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

StreamsGroupMemberMetadataKey

...

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

The records are evolved in the following way

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

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

...

...