...

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:

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.coordinator.snapshot.update.records.per.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 (https://github.com/apache/kafka/blob/ac031bb4e2c95ce00a90c8be6ca3f7c087fe5fbe/share-coordinator/src/main/java/org/apache/kafka/coordinator/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 writing any snapshots. 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. 
Broker level upper bound remains at 500. If users need any changes, they could configure both these configs.

This would allow every share group with traffic patterns to handle the snapshots/disk sizes etc very well. 

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

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

...