Versions Compared

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.

...

2. Define the per-group config in GroupConfig

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 `CONFIGCONFIG_DEF` DEF with `Type.INT`, default sentinel (`-1` to indicate "unset"), validator `atLeast(-1)`, importance `MEDIUM`.

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.

### C3. Propagate the override to `ShareCoordinatorShard` ShareCoordinatorShard

share**File**: `share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorShard.java:654`

Today the shard reads the broker-level value directly:

Code Block
language

...

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

Code Block
language

...

java
public int snapshotUpdateRecordsPerSnapshotOrDefault(String groupId, int defaultValue);

...


4. Wire the provider into ShareCoordinatorShard construction

5.

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

...