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

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

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`share.coordinator.snapshot.update.records.per.snapshot` controls how many `ShareUpdate` records the share coordinator writes between full `ShareSnapshot` records on the share-group state topic. It is currently a single broker-level integer with bound `between(0, 500)` (`ShareCoordinatorConfig.java:103`).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 The bound originated from [PR #21291] (https://github.com/apache/kafka/blob/ac031bb4e2c95ce00a90c8be6ca3f7c087fe5fbe/share-coordinator/src/main/java/org/apache/kafka/coordinator/pull/21291), which converted a runtime check into a declarative `ConfigDef` range. The discussion in that PR (chia7712, AndrewJSchofield, smjn, majialoong) reached an explicit conclusion that the [0, 500] bound, while safe, was chosen without benchmark data and is likely too conservative for high-write share-group workloads. Two follow-up concerns were raised:

1. The current ceiling causes frequent snapshots and write amplification on heavy workloads. The maximum should be evaluated and likely raised, with bound informed by empirical data.
2. A purely broker-level setting cannot cater to mixed workloads on the same cluster: one over-tuned group could block log pruning for the entire `__share_group_state` partition. A per-group override, gated by the broker-level value as an upper guardrail, is the natural shape (chia7712: *"the server-level config should serve as a safety guardrail (an upper bound) for all groups to prevent a single misconfigured group from blocking the cleanup of the entire partition"*).

Snapshot frequency on the share-state topic is a tradeoff:

- **Lower values** (e.g., 0–50): every state change writes a full `ShareSnapshot`. This causes write amplification and grows the log faster than necessary, since update records are tiny relative to snapshots.
- **Higher values** (e.g., several thousand): updates dominate the log and snapshots are rare. Recovery on coordinator failover replays many updates, increasing replay time. Old segments cannot be pruned until a new snapshot covers their key range, so an over-large value keeps stale records around (smjn on PR #21291: *"high values stall cleanup while low values cause repeated snapshots"*).

The current bound `[0, 500]` is a conservative default with no benchmark backing. Operators with heavy share-group throughput cannot tune past 500 today, and within a cluster every share group is forced to use the same value regardless of its individual write profile. This KIP:

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 cleanup1. Raises the broker-level upper bound from 500 to a proposed `10,000`, with a follow-up benchmarking sub-task tracked under KAFKA-20070 that may revise this ceiling before vote.
2. Introduces a per-share-group override `share.snapshot.update.records.per.snapshot`, configurable through `incrementalAlterConfigs` on `ConfigResource.Type.GROUP`, capped at the current broker-level value. Groups without an override transparently inherit whatever the broker config is currently set to.

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

The lower bound of `0` is preserved for backwards compatibility. The semantics of `0` ("snapshot every write") are documented but otherwise unchanged. Disallowing `0` is explicitly out of scope (see *Rejected Alternatives*).     2. 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.

New per-group dynamic config

...

• share.snapshot.update.records.per.

...

• snapshot

A new entry on `ConfigResourceConfigResource.Type.GROUP`GROUP, set via `AdminClient.incrementalAlterConfigs` like any other share-group dynamic config (modeled on `share.record.lock.duration.ms`).| **Name** | `shareAdminClient.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

3. Java constants

`ShareCoordinatorConfig.java` already exposes `SNAPSHOT_UPDATE_RECORDS_PER_SNAPSHOT_CONFIG`. The new per-group constant lives in `GroupConfig.java`:```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

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.

Proposed Changes

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:

...

.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

`NEW_MAX` is set during DISCUSS once benchmark data lands (see *Test Plan*).

...

Define the per-group config in GroupConfig

group in `GroupConfig`**File**: `group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupConfig.java`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`). 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.### C.

Propagate the override to ShareCoordinatorShard

share 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:

...

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

...

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.

...

Wire the provider

...

into ShareCoordinatorShard construction

`ShareCoordinatorShard` is constructed via its `Builder` in `ShareCoordinatorService`. Inject the `ShareGroupConfigProvider` through the available 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

...

GroupCoordinatorService)

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 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 - Cross-reference KIP-932 (the originating share-groups KIP) noting that the [0, 500] text in KIP-932 is superseded by this KIP.

Compatibility, Deprecation, and Migration Plan

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

— would also pass validation in the new range; this case does not apply on upgrade.

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

Deprecation

• There is no deprecation plan

Migration

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

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

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.

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)

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.