You are viewing an old version of this page. View the current version.

Compare with Current View Page History

Version 1 Next »

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: "Draft"

Discussion thread: here 

JIRA: here 

Configurable snapshot frequency for share groups

Motivation

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

The bound originated from [PR #21291](https://github.com/apache/kafka/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:

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

  1. Updated broker-level config: `share.coordinator.snapshot.update.records.per.snapshot`


| **Name** | `share.coordinator.snapshot.update.records.per.snapshot` (unchanged) |
| **Type** | `INT` (unchanged) |
| **Default** | `500` (unchanged) |
| **Validator** | `between(0, 10000)` — proposed initial ceiling. A follow-up benchmarking ticket (filed as a sub-task of KAFKA-20070, see *Test Plan*) may revise this number before vote based on empirical data. |
| **Importance** | `MEDIUM` (unchanged) |
| **Doc** | "The maximum number of update records the share coordinator writes between snapshot records, applied as a ceiling across all share groups on this broker. 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. New per-group dynamic config: `share.snapshot.update.records.per.snapshot`

A new entry on `ConfigResource.Type.GROUP`, set via `AdminClient.incrementalAlterConfigs` like any other share-group dynamic config (modeled on `share.record.lock.duration.ms`).


| **Name** | `share.snapshot.update.records.per.snapshot` |
| **Type** | `INT` |
| **Default** | _unset_ — falls back to `share.coordinator.snapshot.update.records.per.snapshot` |
| **Validator** | `between(0, broker_value)` enforced at apply time; static lower bound `atLeast(0)` |
| **Importance** | `MEDIUM` |
| **Doc** | "Number of update records the share coordinator writes between snapshot records for this share group. If unset, the broker-level `share.coordinator.snapshot.update.records.per.snapshot` is used. Values exceeding the broker-level setting are rejected." | 


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";
```


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:

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

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

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

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

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.


  • No labels