...

Configurable snapshot frequency for share groups

Motivation

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

...

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 snapshotsever triggering an update (count-based) snapshot. 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. 

Together with this also update the lower and upper bound of existing config of share.coordinator.snapshot.update.records.per.snapshot. From (0,500) to (200,1000) to handle high through-puts of share groups. 
Values below 200 will cause the coordinator to write snapshots so often that small update records barely save any disk space.

However, this needs a migration note, as any cluster which has a value between 0 and 199, will fail to start after the upgrade. Clusters should update to a value >=200.

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

...