DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
We make core assignment options configurable centrally on the broker, without relying on each clients configuration. This allows tuning a streams group without redeploying the streams application. The three core assignment options will be introduced on the broker-side: acceptable.recovery.lag, num.warmup.replicas and , num.standby.replicas, and rack.aware.assignment.tags. They can be configured both globally on the broker, and dynamically for specific streams groups through the IncrementalAlterConfigs and DescribeConfigs RPCs.
We store the last used assignment configuration in the group metadata on the broker. This way, we can detect if an assignment configuration is dynamically changed, and we can trigger reassignment immediately.
...
| Code Block | ||||
|---|---|---|---|---|
| ||||
{
"apiKey": TBD,
"type": "response",
"name": "StreamsGroupHeartbeatResponse",
"validVersions": "0", // "0" in 4.2/4.3 release; bumped to "0-1" in 4.4 release
"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": [
{ "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+",
"about": "The member id is always generated by the streams consumer."},
{ "name": "MemberEpoch", "type": "int32", "versions": "0+",
"about": "The member epoch." },
{ "name": "HeartbeatIntervalMs", "type": "int32", "versions": "0+",
"about": "The heartbeat interval in milliseconds." },
{ "name": "AcceptableRecoveryLag", "type": "int32", "versions": "0+",
"about": "The maximal lag a warm-up task can have to be considered caught-up." }, // incorrectly implemented as `int32` in 4.2 release; updated to `int64` and request/response version bump to "1" in 4.34 release.
{ "name": "RackAwareAssignmentTags", "type": "[]string", "versions": "1+",
"about": "The assignment tags used by the group coordinator for rack aware standby task assignment." }, // added in 4.4 release
{ "name": "TaskOffsetIntervalMs", "type": "int32", "versions": "0+",
"about": "The interval in which the task changelog offsets on a client are updated on the broker. The offsets are sent with the next heartbeat after this time has passed." },
{ "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": "EndpointInformationEpoch", "type": "int32", "versions": "0+",
"about": "The endpoint epoch set in the response"},
{ "name": "PartitionsByUserEndpoint", "type": "[]EndpointToPartitions", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "Global assignment information used for IQ. Null if unchanged since last heartbeat." ,
"fields": [
{ "name": "UserEndpoint", "type": "Endpoint", "versions": "0+",
"about": "User-defined endpoint to connect to the node" },
{ "name": "ActivePartitions", "type": "[]TopicPartition", "versions": "0+",
"about": "All active partitions available on the node" },
{ "name": "StandbyPartitions", "type": "[]TopicPartition", "versions": "0+",
"about": "All standby paritions available on the node"}
]
}
],
"commonStructs": [
{ "name": "Status", "versions": "0+", "fields": [
// Possible status codes
// 0 - STALE_TOPOLOGY - The topology epoch supplied is lower than 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 - INCORRECTLY_PARTITIONED_TOPICS - One or more topics are incorrectly partitioned, that is, they are not copartitioned 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.
// Incorrectly partitioned topics are indicated in the StatusDetail.
// 3 - 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.
// 4 - SHUTDOWN_APPLICATION - A client requested the shutdown of the whole application.
// 5 - ASSIGNMENT_DELAYED - No assignment was provided because assignment computation was delayed.
{ "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": "uint16", "versions": "0+",
"about": "port of the endpoint" }
]}
]
} |
...
Name | Type | Default | Doc |
|---|---|---|---|
group.streams.session.timeout.ms | int | 45s | The timeout to detect client failures when using the streams group protocol. |
group.streams.min.session.timeout.ms | int | 45s | The minimum session timeout. |
group.streams.max.session.timeout.ms | int | 60s | The maximum session timeout. |
group.streams.heartbeat.interval.ms | int | 5s | The heartbeat interval given to the members. |
group.streams.min.heartbeat.interval.ms | int | 5s | The minimum heartbeat interval. |
group.streams.max.heartbeat.interval.ms | int | 15s | The maximum heartbeat interval. |
group.streams.max.size | int | MaxValue | The maximum number of streams clients that a single streams group can accommodate. |
group.streams.acceptable.recovery.lag | long | 10’000 | The maximum acceptable lag (number of offsets to catch up) for a client to be considered caught-up enough to receive an active task assignment. |
group.streams.num.warmup.replicas | int | 2 | The maximum number of warmup replicas. |
group.streams.max.warmup.replicas | int | 20 | Maximum for dynamic configurations of the warmup replica configuration |
group.streams.num.standby.replicas | int | 0 | The number of standby replicas for each task. |
group.streams.max.standby.replicas | int | 2 | Maximum for dynamic configurations of the standby replica configuration |
group.streams.task.offset.interval.ms | int | 60s | 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. |
group.streams.min.task.offset.interval.ms | int | 15s | Minimum for dynamic configurations of the task offset interval. Used to restrict users of the cluster from sending the task changelog offsets too often. |
group.streams.assignor.name | string | null | The name of the task assignor used for all streams groups. Can be |
group.streams.initial.rebalance.delay.ms | int | 3000 | The first rebalance of a group is delayed by this amount to allow more members to join the group. |
group.streams.rack.aware.assignment.tags | string[] | empty list | The client tags used for rack aware standby task assignment. |
Group Configurations
We will add new configurations for the resource type GROUP in DescribeConfigs and IncrementalAlterConfigs to override the default broker configurations dynamically for specific groups.
Name | Type | Default | Doc |
|---|---|---|---|
streams.session.timeout.ms | int | 45sThe timeout to detect client failures when using the streams group protocol. | |
streams.heartbeat.interval.ms | int | 5s | The heartbeat interval given to the members. |
streams.acceptable.recovery.lag | long | 10’000The maximum acceptable lag (number of offsets to catch up) for a client to be considered caught-up enough to receive an active task assignment. | |
streams.num.warmup.replicas | int | 2 | The maximum number of warmup replicas. |
streams.num.standby.replicas | int | 0The number of standby replicas for each task. | |
streams.task.offset.interval.ms | int | 60sThe 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. | |
streams.assignor.name | string | nullThe name of the task assignor used for this streams groups. Can be | |
streams.initial.rebalance.delay.ms | int | 3000The | |
streams.rack.aware.assignment.tags | string[] | empty list first rebalance of a group is delayed by this amount to allow more members to join the group. |
Streams API
In a future major version, when the classic group protocol will be deprecated, the interfaces introduced in KIP-924 would be deprecated, unless an integration with this KIP is proposed in a follow-up KIP.
...