Current state: "Draft"
Discussion thread: here
JIRA: here
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).
To handle all that per record state, across broker restarts, Kafka writes to an internal topic called __share_group_state.
Two kind of records go there.
- Update records (small and incremental)
- Snapshot records (big, complete)
We need snapshots, because when a broker restarts, replaying millions of updates would take forever, and snapshots are like save points.
Today snapshots are controlled only with this one broker level config : share.coordinator.snapshot.update.records.per.snapshot(is for number of small updates which get written before saving it as a full snapshot.)
If we set it low, too many snapshots are written, and recovery is fast. If we set it to high, we mostly write tiny updates, and less disk is used but recovery is replaying several tiny udpates, might take long.
So it's a trade off. write-cost vs recovery-cost
It's range is 0-500 (https://github.com/apache/kafka/blob/ac031bb4e2c95ce00a90c8be6ca3f7c087fe5fbe/share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorConfig.java#L103)
Actual problem :
Is 500 too low ? Probably for high traffic share groups it is low and they get benefitted from snapshotting less often.
What if we remove the upper limit ? If there is no cap, a misconfigured group can write millions of updates without ever triggering an update (count-based) snapshot. This will fill up the disk, as broker cannot delete old log files.
What if we have one config at broker level : There would be different kinds of share groups with different traffic patterns and with mixed workloads, and as they share the same state topic, it may fit one, but not others.
New group level config : A new config (share.snapshot.update.records.per.snapshot) per group, together with broker's value(as hard ceiling) is ideal. Groups without an override, would inherit broker's config.
This would allow every share group with traffic patterns to handle the snapshots/disk sizes etc very well.
Together with this also update the lower and upper bound of existing config of share.coordinator.snapshot.update.records.per.snapshot. From (0,500) to (200,1000) to handle high through-puts of share groups.
Values below 200 will cause the coordinator to write snapshots so often that small update records barely save any disk space.
However, this needs a migration note, as any cluster which has a value between 0 and 199, will fail to start after the upgrade. Clusters should update to a value >=200.
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.
| Name | share.coordinator.snapshot.update.records.per.snapshot | | Type | INT | | Default | 500 | | Validator | between(200, 1000) (changed) | | Importance | MEDIUM | | Doc | "The number of update records the share coordinator writes between snapshot records, applied as a ceiling across all share groups on this broker. Must be in [200, 1000]. May be overridden per group via the share.snapshot.update.records.per.snapshot group config; per-group values are clamped to this ceiling." | |
The floor value moves from 0 to 200, to disallow values that waste disk by writing too many full snapshots.
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.
A new entry on ConfigResource.Type.GROUP, set via AdminClient.incrementalAlterConfigs.
| 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." | |
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"; |
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorConfig.java
Change:
.define(SNAPSHOT_UPDATE_RECORDS_PER_SNAPSHOT_CONFIG, INT, SNAPSHOT_UPDATE_RECORDS_PER_SNAPSHOT_DEFAULT, between(200, 1000), MEDIUM, SNAPSHOT_UPDATE_RECORDS_PER_SNAPSHOT_DOC) |
The bounds `[200, 1000]` are the proposed initial values
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupConfig.java
Add the new constant alongside existing `share.*` group-level entries . Register it in CONFIG_DEF.
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.
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorShard.java
Today the shard reads the broker-level value directly:
int updatesPerSnapshotLimit = config.shareCoordinatorSnapshotUpdateRecordsPerSnapshot(); |
Replace with a per-group lookup that falls back to the broker default. Reuse the existing `ShareGroupConfigProvider` pattern (`group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/share/ShareGroupConfigProvider.java:42-60`). It already accepts a `groupId` and returns either the group override or the supplied default, and is the established channel from `GroupConfigManager` to coordinator code.
Add a new accessor:
public int snapshotUpdateRecordsPerSnapshotOrDefault(String groupId, int defaultValue); |
Inject through the available builder (the provider is already available in the GroupCoordinatorService)
floor raised from `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
(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.
Required operator action before upgrade (only for clusters that explicitly set the broker config below 200):
1. Run kafka-configs.sh --bootstrap-server <broker> --describe --entity-type brokers --all and inspect the value of share.coordinator.snapshot.update.records.per.snapshot on each broker.
2. If any broker has the value set below 200 (most commonly `0`) either:
- Remove the override so the broker falls back to the default of `500`, or
- Raise the override to a value in `[200, 1000]` that matches the operator's intent.
3. Roll out the upgrade.
Operators who have never overridden the broker config (i.e., it is implicitly `500`) need no action.
Unit Tests - Extend ShareCoordinatorConfigTest, ShareCoordinatorShardTest and GroupConfigTest
Integration Tests - Extend ShareCoordinatorIntegrationTest
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. Per-group override only; keep broker max at 500.
The broker ceiling becomes the effective hard cap for any per-group value
Leaving 500 in place forces operators to raise the broker config anyway to take advantage of per-group tuning.
3. Keep the floor at `0` (raise only the ceiling)
The PR thread surfaced 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.