...

public static final String SHARE_SNAPSHOT_UPDATE_RECORDS_PER_SNAPSHOT_CONFIG = "share.snapshot.update.records.per.snapshot";

Proposed Changes

1. Change broker level lower and upper bounds

...

The bounds `[200, 1000]` are the proposed initial values

2. Define the per-group config in GroupConfig

...

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

...

public int snapshotUpdateRecordsPerSnapshotOrDefault(String groupId, int defaultValue);

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

...

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.

DeprecationNone

• There is no deprecation plan

Migration

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

...

Integration Tests - Extend ShareCoordinatorIntegrationTest

Rejected Alternatives

Here are the rejected alternatives.

1. Raise broker max only; no per-group override

...