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.

Status

Current state: "DraftUnder Discussion"

Discussion thread: here 

JIRA: here 

Configurable snapshot frequency for share groups

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

Code Block

| 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." |

...

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.

Code Block

| 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 must upgrade the controller/broker before relying on the per-group override.

Deprecation

...

Deprecation

• There is no deprecation plan

Migration

Required operator action before upgrade (only for clusters that explicitly set the broker config below 200):

...

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

Test Plan

### 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.

Rejected Alternatives

### 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

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)

Rejected Alternatives

Here are the rejected alternatives.

Raise broker max only. No per-group override

A 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 acts as a hard cap on every per-group value (per-group 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 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)

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. 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.