This page is meant as a template for writing a KIP. To create a KIP choose Tools->Copy on this page and modify with your content and replace the heading with the next KIP number and a description of your issue. Replace anything in italics with your own description.
Current state: "Draft"
Discussion thread: here
JIRA: here
Configurable snapshot frequency for share groups
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.
2. New per-group dynamic config: share.snapshot.update.records.per.snapshot
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." | |
3. 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"; |
4 . No other public surface changes
- No RPC schema changes. `WriteShareGroupState` is unchanged; the per-group override is read from `GroupConfigManager` server-side, not transmitted over the wire.
- No new metrics in v1. (Optional follow-up: a `share-snapshot-write-rate` metric tagged by group — out of scope.)
- No CLI changes; `kafka-configs.sh --entity-type groups --entity-name <gid> --alter --add-config share.snapshot.update.records.per.snapshot=N` works through the existing GROUP entity-type plumbing.
Describe the new thing you want to do in appropriate detail. This may be fairly extensive and have large subsections of its own. Or it may be a few sentences. Use judgement based on the scope of the change.
### A. Raise the broker-level upper bound
**File**: `share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorConfig.java:103`
Change:
```java
.define(SNAPSHOT_UPDATE_RECORDS_PER_SNAPSHOT_CONFIG, INT,
SNAPSHOT_UPDATE_RECORDS_PER_SNAPSHOT_DEFAULT,
between(0, NEW_MAX),
MEDIUM, SNAPSHOT_UPDATE_RECORDS_PER_SNAPSHOT_DOC)
```
`NEW_MAX` is set during DISCUSS once benchmark data lands (see *Test Plan*).
### B. Define the per-group config in `GroupConfig`
**File**: `group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupConfig.java`
Add the new constant alongside existing `share.*` group-level entries (around line 60, near `SHARE_RECORD_LOCK_DURATION_MS_CONFIG`). Register it in `CONFIG_DEF` with `Type.INT`, default sentinel (`-1` to indicate "unset"), validator `atLeast(-1)`, importance `MEDIUM`. Add a public final field and a getter mirroring the pattern used for `shareRecordLockDurationMs` (`GroupConfig.java:124`).
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.
### C. Propagate the override to `ShareCoordinatorShard`
**File**: `share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorShard.java:654`
Today the shard reads the broker-level value directly:
```java
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:
```java
public int snapshotUpdateRecordsPerSnapshotOrDefault(String groupId, int defaultValue);
```
Call it in `generateShareStateRecord` keyed off `key.groupId()`. The fallback path is the broker-level value; no behavior change for groups that have not set the override.
### D. Wire the provider into `ShareCoordinatorShard` construction
`ShareCoordinatorShard` is constructed via its `Builder` in `ShareCoordinatorService`. Inject the `ShareGroupConfigProvider` through the builder (the provider is already available in the `GroupCoordinatorService` boot sequence; pass it across the share-coordinator module boundary in the same way `ShareCoordinatorConfig` is passed today). No constructor-signature break for external code, since the `Builder` is package-private.
### E. 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.
- Cross-reference KIP-932 (the originating share-groups KIP) noting that the [0, 500] text in KIP-932 is superseded by this KIP.
### Backwards compatibility
- **Wire compatibility**: unaffected. No RPC, protocol, or storage changes.
- **API compatibility**: unaffected. Only additive `ConfigDef` and `GroupConfig` entries.
- **Behavioral compatibility**: brokers that do not change the broker-level value see no behavior change. A broker upgraded to a release containing this KIP that retains its previous setting (`<= 500`) behaves identically to today. Operators who explicitly raise the broker value or set a per-group override opt into the new behavior.
- **Mixed-version clusters**: while the controller is on an older release that does not understand the new group config name, `incrementalAlterConfigs` requests with `share.snapshot.update.records.per.snapshot` are rejected with `InvalidConfigurationException`. Operators must upgrade the controller before relying on the per-group override. This matches how every prior `GroupConfig` addition has rolled out.
### Deprecation
None. No existing behavior is removed or renamed.
### Migration
No user action required. Operators wanting the new ceiling must explicitly raise `share.coordinator.snapshot.update.records.per.snapshot`. Per-group overrides are opt-in via `kafka-configs.sh`.
### Benchmark workstream
The proposed ceiling of `10,000` is an initial value chosen as 20× the current `500` ceiling — large enough to relieve the pressure described in the PR thread without entering territory where coordinator failover replay time becomes operationally painful. **A dedicated benchmarking sub-task will be filed under KAFKA-20070** to validate this choice with empirical data; the ceiling may be adjusted up or down before the KIP vote based on the results.
The benchmark sub-task should run two scenarios on a representative share-coordinator cluster:
1. **Sustained-write workload**: high `WriteShareGroupState` QPS against a fixed group, varying the per-group override across `{500, 2000, 5000, 10000, 25000, 50000}`. Measure: (a) `__share_group_state` log growth rate, (b) coordinator failover replay time, (c) write amplification ratio (snapshot-bytes / update-bytes).
2. **Pruning-lag workload**: same as above but track time from "old key state superseded" to "log segment containing it eligible for deletion." This is the metric that operationalizes chia7712's "unbounded growth" concern.
The final ceiling is the largest value for which (b) replay time and (a) log growth rate stay within currently observed operational SLOs. If benchmarks show `10,000` is unsafe, the KIP will be revised downward (e.g. to `5,000`) before vote; if benchmarks show headroom, it may be revised upward.
### Unit tests
- `ShareCoordinatorConfigTest`: assert validator accepts `0`, `500`, `NEW_MAX`; rejects `-1`, `NEW_MAX + 1`.
- New `GroupConfigTest` cases: per-group value `0` and any positive value `<= broker_ceiling` accepted; values `> broker_ceiling` rejected with the bound named in the message.
- `ShareCoordinatorShardTest`: when a group has the override set to `N`, `generateShareStateRecord` writes a `ShareSnapshot` after exactly `N` updates and a `ShareUpdate` before. When unset, falls back to broker default. Pre-existing tests at `ShareCoordinatorShardTest.java:757,888` continue to pass unchanged.
### Integration test
Extend `ShareCoordinatorIntegrationTest` (or add a new one) to:
1. Create two share groups on the same coordinator partition.
2. Set group A's override to a small value (e.g., 5), leave group B at the broker default.
3. Drive writes to both. Assert from the log that A produces snapshots ~5× more frequently than B.
### System test
Not required for v1. A backfill/restart system test asserting failover replay time as a function of the override is a reasonable follow-up.
### 1. Raise broker max only; no per-group override
Considered. Smallest KIP, matches AndrewJSchofield's lighter framing. **Rejected** because it does not address chia7712's central concern: a single misconfigured group can block `__share_group_state` log pruning across all groups on its partition. Without a per-group knob, operators cannot safely tune for one workload without imposing the tuning on everyone.
### 2. Per-group override only; keep broker max at 500
Considered. **Rejected** because the broker ceiling becomes the effective hard cap for any per-group value (the per-group config is `between(0, broker_value)`). Leaving 500 in place forces operators to raise the broker config anyway to take advantage of per-group tuning, so coupling the two changes in one KIP is simpler.
### 3. Disallow `0` as a value (`atLeast(1)` or chia7712's `between(200, 1000)`)
Considered (PR #21291: AJ "we could consider disallowing it," chia7712 proposed `between(200, 1000)`). **Rejected** for this KIP because (a) the PR thread did not reach consensus, (b) treating `0` as misconfiguration is a behavior change orthogonal to the bound-evaluation work, and (c) doing it here would expand the KIP's risk surface. If benchmark data shows `0` causes pathological write amplification we will file a follow-up KIP.
### 4. Extend KIP-1240 or KIP-932 with this evaluation
Explicitly **rejected** by AndrewJSchofield on PR #21291: *"I would not increase the scope of KIP-1240. We can easily put in a specific KIP for this once we have done the work."* This KIP is that specific KIP.
### 5. Defer the new ceiling entirely (leave the number TBD until benchmarks complete)
Considered. **Rejected** because reviewers reading the KIP need a concrete proposal to react to. Instead, the KIP proposes `10,000` as the working ceiling and commits to the benchmarking sub-task under KAFKA-20070 as a gate before vote. The number will be revised in either direction if data warrants.
### 6. Transmit the per-group value in `WriteShareGroupState` requests
Considered as an alternative propagation mechanism. **Rejected** because the existing `GroupConfigManager` → `ShareGroupConfigProvider` channel already does exactly this for other share-group configs (`share.record.lock.duration.ms` etc.) and reusing it costs zero new RPC schema work.