Status

Current state: "DraftUnder Discussion"

Discussion thread: here 

JIRA: here 

Motivation

Kafka has share groups. As multiple consumers in the same share group can consume from the same partition concurrently, records get individually acknowledged and broker tracks per record state(delivered, acked, in-flight etc).

...

Note : There is another config share.coordinator.cold.partition.snapshot.interval.ms (default 5 mins) which forces snapshotting on a timely basis, but only for share partitions with no updates. So the old log recs of the idle groups would be eligible for cleanup.

Public Interfaces

Updated broker-level config

• : share.coordinator.snapshot.update.records.per.snapshot

...

At very low values, every small state change triggers a full ShareSnapshot instead of a small ShareUpdate, so most of what gets written to the log is large snapshot records. This is a behavior break for any cluster currently set below 200 (including the documented value `0`); See the Migration Plan section which is required before upgrading.

New per-group dynamic config

• : share.snapshot.update.records.per.snapshot

...

| Name | share.snapshot.update.records.per.snapshot |
| Type | INT |
| Default | _unset_ — falls back to share.coordinator.snapshot.update.records.per.snapshot |
| Validator | between(200, broker_value) enforced at apply time; static between(200, 1000) for the ConfigDef itself |
| Importance | MEDIUM |
| Doc | "Number of update records the share coordinator writes between snapshot records for this share group. Must be in [200, broker_ceiling], where broker_ceiling is the current value of share.coordinator.snapshot.update.records.per.snapshot. If unset, the broker-level value is used." |

Java constants

ShareCoordinatorConfig.java already exposes `SNAPSHOT_UPDATE_RECORDS_PER_SNAPSHOT_CONFIG`. The new per-group constant lives in `GroupConfig.java`:

public static final String SHARE_SNAPSHOT_UPDATE_RECORDS_PER_SNAPSHOT_CONFIG = "share.snapshot.update.records.per.snapshot";

Proposed Changes

...

Change broker level lower and upper bounds

share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorConfig.java

...

The bounds `[200, 1000]` are the proposed initial values2.

Define the per-group config

...

in GroupConfig

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupConfig.java

...

Validation that the group value does not exceed the broker ceiling is performed in `validate(Map<String, String> props, ShareGroupConfig defaults)` — the same hook used by other share group configs. If the requested value exceeds the broker ceiling, throw `InvalidConfigurationException` with a message naming both bounds.3.

Propagate the override

...

to ShareCoordinatorShard

share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorShard.java

...

public int snapshotUpdateRecordsPerSnapshotOrDefault(String groupId, int defaultValue);

...

Wire the provider

...

into ShareCoordinatorShard construction

Inject through the available builder (the provider is already available in the GroupCoordinatorService)5.

Documentation

...

• Update the [Kafka Configuration](https://kafka.apache.org/documentation/#configuration) page to reflect the new broker ceiling and to add the new group-level entry under "Share Group Configurations."

...

• Update the share-group operator documentation (`docs/streams/...` equivalent for share groups, exact location to be confirmed during implementation) with a tuning section explaining the snapshot/update tradeoff.

Compatibility, Deprecation, and Migration Plan

Behavior break

 floor  Floor raised from `0` to `200` '0' to '200' -- any cluster currently configured with share.coordinator.snapshot.update.records.per.snapshot set to a value below 200 will fail broker startup config validation after upgrade. Clusters using the default 500 or any value in [200, 500] are unaffected. Clusters that previously set the value to anything between 501 and 1000 — not currently possible since the existing range is [0, 500] — would also pass validation in the new range; this case does not apply on upgrade.

Summary :Summary -

• Values 0–199: fail to start after upgrade (operator must fix)

...

• Values 200–500: unaffected

...

•  Values 501+: don't exist (can't happen)

Behavioral compatibility

(ceiling raise): Ceiling raise -- raising the ceiling from 500 to 1000 is non-breaking on its own. Brokers that retain their existing setting in [200, 500] behave identically to today. Operators can opt into the higher range explicitly.

...

Operators who have never overridden the broker config (i.e., it is implicitly `500`) need no action.

Test Plan

Unit Tests

ShareCoordinatorConfigTest (covers the broker-level bounds change)

ShareCoordinatorShardTest (covers the per-group config)

GroupConfigTest (covers the runtime use of the per-group value)

Integration Tests

 ShareCoordinatorIntegrationTest (covers end-to-end propagation of the per-group config) - Extend ShareCoordinatorConfigTest, ShareCoordinatorShardTest and GroupConfigTestIntegration Tests - Extend ShareCoordinatorIntegrationTest

Rejected Alternatives

Here are the rejected alternatives.1.

Raise broker max only

...

. No per-group override

A misconfigured group can write millions of updates without ever triggering an update (count-based) snapshot.2. single broker-level setting forces all share groups on the cluster to use the same value. Different groups have different write profiles. A low-traffic group benefits from a low value (fast recovery), while a high-traffic group benefits from a high value. Without a per-group override, operators must pick one value that compromises for all groups.
Additionally, since all groups on the same __share_group_state partition share log-pruning behavior, one group set very high can delay pruning for unrelated groups on the same partition.

Per-group override only

...

. Keep broker max at 500. 

The broker ceiling becomes the effective hard cap for any acts as a hard cap on every per-group value (per-group value
Leaving 500 in place forces operators is between(200, broker_value)). If we keep the broker max at 500 means, per-group overrides cannot exceed 500 either, so high-traffic groups cannot benefit from the per-group config. In this case operators would have to raise the broker config anyway to take advantage of per-group tuning.3. ceiling immediately anyway to use the new per-group setting meaningfully. Basically the per-group config is not of much use without the ceiling raise.

Keep the floor at

...

'0; (raise only the ceiling)

The PR thread surfaced consensus that `0` Raising only the ceiling (between(0, 1000)) is of minimum risk and it is intact with the documented 0 semantics avoids any upgrade-time validation failure. But based on the PR thread discussion , there is a consensus that 0 and other very small values waste disk by writing a full snapshot for nearly every state change, and no real-world workload benefits from them..