DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
| Table of Contents |
|---|
Status
Current state: Under Discussion
Discussion thread: here
JIRA: here
Released: <Kafka Version>
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
Note that this is joint work by Unknown User (lucasbru) and Unknown User (cadonna) .
Motivation
Kafka Streams clients use a protocol to distribute stream processing tasks among Streams clients. Since the inception of Streams, this has worked by piggybacking custom byte payloads on the pre-KIP-848 consumer rebalance protocol. A custom partition assignor on one of the Streams clients would perform task assignment and various other centralized tasks, such as internal topic initialization. While this approach has served its purpose well and the streams partition assignor has achieved impressive flexibility and complexity over the years, it has also had a number of deficiencies:
...
The new group protocol would be guarded by enabling a new version of the group.version feature flag on the broker side.
Updated properties
Configuration | Description | Values |
|---|---|---|
| The list of enabled rebalance protocols. |
This will be added to the default value of this configuration property once this feature is complete. |
New properties
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 consumers that a single consumer 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 | 4 | 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 changelog task 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 offsets too often. |
group.streams.assignor | string | null | The name of the task assignor used for all streams groups. Can be |
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 |
|---|---|---|---|
group.streams.session.timeout.ms | int | 45s | The timeout to detect client failures when using the streams group protocol. |
group.streams.heartbeat.interval.ms | int | 5s | The heartbeat interval given to the members. |
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.max.warmup.replicas | int | 2 | The maximum number of warmup replicas. |
group.streams.num.standby.replicas | int | 0 | The number of standby replicas for each task. |
group.streams.task.offset.interval.ms | int | 60s | The interval in which the changelog task offsets on a client are updated on the broker. The offsets are sent with the next heartbeat after this time has passed. |
group.streams.assignor | string | null | The name of the task assignor used for this streams groups. Can be |
Streams Interfaces
In a future major version, when the classic consumer 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.
Streams Configurations
New configurations
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 |
group.remote.assignor | string | null | The server side assignor to use. |
Deprecations
In a future major version, when the classic consumer protocol will be deprecated, the following configuration options will be deprecated:
...