Versions Compared

Key

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

...

Code Block
languagejava
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

...

Code Block
languagejava
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.

...