DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
| Code Block | ||||
|---|---|---|---|---|
| ||||
{
"apiKey": TBD,
"type": "request",
"listeners": ["broker"],
"name": "StreamsGroupHeartbeatRequest",
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "GroupId", "type": "string", "versions": "0+", "entityType": "groupId",
"about": "The group identifier." },
{ "name": "MemberId", "type": "string", "versions": "0+",
"about": "The member id generated by the coordinator. The member id must be kept during the entire lifetime of the member." },
{ "name": "MemberEpoch", "type": "int32", "versions": "0+",
"about": "The current member epoch; 0 to join the group; -1 to leave the group; -2 to indicate that the static member will rejoin." },
{ "name": "InstanceId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "null if not provided or if it didn't change since the last heartbeat; the instance Id otherwise." },
{ "name": "RackId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "null if not provided or if it didn't change since the last heartbeat; the rack ID of consumer otherwise." },
{ "name": "RebalanceTimeoutMs", "type": "int32", "versions": "0+", "default": -1,
"about": "-1 if it didn't change since the last heartbeat; the maximum time in milliseconds that the coordinator will wait on the member to revoke its partitions otherwise." },
{ "name": "TopologyId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": null,
"about": "The ID of the topology. Only sent when memberEpoch = 0, must be non-empty. Null otherwise. Used to check if topology corresponds to server-topology. " },
{ "name": "ActiveTasks", "type": "[]TaskIds", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "Currently owned active tasks for this client. Null if unchanged since last heartbeat." },
{ "name": "StandbyTasks", "type": "[]TaskIds", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "Currently owned standby tasks for this client. Null if unchanged since last heartbeat." },
{ "name": "WarmupTasks", "type": "[]TaskIds", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "Currently owned warm-up tasks for this client. Null if unchanged since last heartbeat." },
{ "name": "ProcessId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "Identity of the streams instance that may have multiple consumers. Null if unchanged since last heartbeat." },
{ "name": "UserEndpoint", "type": "stringEndpoint", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "User-defined endpoint for Interactive Queries. Null if unchanged since last heartbeat." },
{ "name": "ClientTags", "type": "[]stringKeyValue", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "Used for rack-aware assignment algorithm. Null if unchanged since last heartbeat." },
{ "name": "TaskOffsets", "type": "[]TaskOffset", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "Cumulative offsets for tasks. Only updated when a warm up task has caught up, and according to the task offset interval. Null if unchanged since last heartbeat." },
{ "name": "TaskEndOffsets", "type": "[]TaskOffset", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "Cumulative end-offsets for tasks. Only updated when a warm up task has caught up, and according to the task offset interval. Null if unchanged since last heartbeat. the task offset interval. Null if unchanged since last heartbeat." },
{ "name": "ShutdownApplication", "type": "bool", "versions": "0+", "default": false,
"about": "Whether all Streams clients in the group should shut down. Can only be defined if memberEpoch = -1" }
],
"commonStructs": [
{ "name": "Endpoint", "versions": "0+", "fields": [
{ "name": "Host", "type": "string", "versions": "0+",
"about": "host of the endpoint" },
{ "name": "Port", "type": "uint16", "versions": "0+",
"about": "port of the endpoint" }
]},
{ "name": "KeyValue", "versions": "0+", "fields": [
{ "name": "Key", "type": "string", "versions": "0+",
"about": "key of the config" },
{ "name": "ShutdownApplicationValue", "type": "boolstring", "versions": "0+",
"default": false,
"about": "Whethervalue all Streams clients in the group should shut down. Can only be defined if memberEpoch = -1of the config" }
],
"commonStructs": [},
{ "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." }
]}
]
} |
...
- When the first member joins the group, the group metadata and the member metadata records for the member are first created.
- Every time a new member joins the group, another member metadata record is added
- Whenever a member updates its member metadata through the heartbeat, a new version of the member metadata record is appended.
- When a member leaves the group, a tombstone is written for the member metadata key.
When the group was empty for a configurable period, as for , lost all members and offsets.retention.minutes expires, the group metadata tombstone is written.
...
- When the group coordinator receives a valid
StreamsGroupInitializerequest, 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 last member leaves the group, tombstones are written to remove the records for the groupgroup lost all members and offsets.retention.minutes expires, the group metadata tombstone is written.
StreamsGroupPartitionMetadataKey
...
The target assignment is stored in N + 1 records, where N is the number of members in the group. When a new target assignment is computed, the group coordinator will compare it with the current target assignment and only write the difference between the two assignments to the __consumer_offsets partition. The assignment must be atomic, so the group coordinator will ensure that all the records are written in a single batch, limiting the size to the maximum batch size (default 1MB), as in consumer groups. When the last a member leaves the group, a new target assignment is calculated and the target assignment member valWhen the group lost all members and offsets.retention.minutes expires, the group metadata tombstone is written to remove the record.
StreamsGroupTargetAssignmentMemberKey
...