Status

Current state: "Under Discussion"

Discussion thread: here

JIRA: here

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

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)

Measurements on a local 4.3.0 cluster show __share_group_state update records are small (~154 bytes average), so the records-based default of 500 corresponds to only ~75 KB between snapshots. A bytes-based 20 MB default (matching KRaft) is a more predictable, workload-independent trigger.

Problems

Records-based control is workload-dependent. Different share groups have different state-batch sizes. A records-based threshold means "snapshot every N updates," but those N updates can be 5 KB on one workload and 500 KB on another, so disk impact is unpredictable.

One config at the broker level cannot fit all share groups. Different groups have different traffic patterns. With mixed workloads sharing the same state topic, one value may fit one group but not others.

Is 500 too low ? Probably for high traffic share groups it is low and they get benefitted from snapshotting less often. It is conservative for high-traffic workloads

What if we remove the upper limit ? If there is no cap, a misconfigured group can write millions of updates without ever 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.

Proposal

Switch to bytes-based control (similar to metadata.log.max.record.bytes.between.snapshots in MetadataLogConfig.java)

- Broker-level: share.coordinator.max.record.bytes.between.snapshots, default 20 MB, (matching KRaft's well-tested default.)

- Per-group: share.max.record.bytes.between.snapshots, configurable per share group. This would allow every share group with different traffic patterns to handle the snapshots/disk sizes etc very well.

The existing records-based config (share.coordinator.snapshot.update.records.per.snapshot) is deprecated with a warning at broker startup if set, and removed in a future release.

Bytes-based control gives operators predictable disk impact, independent of workload, and follows the same pattern Kafka already uses for the metadata log. (also in kip 770)

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.

Public Interfaces

New broker-level config

share.coordinator.max.record.bytes.between.snapshots

| Name | share.coordinator.max.record.bytes.between.snapshots |
| Type | LONG |
| Default | 20 * 1024 * 1024 (20 MB)  |
| Validator | atLeast(1)  |
| Importance | HIGH  |
| Doc | "The maximum number of bytes of update records the share coordinator writes between snapshot records, applied as a ceiling across all share groups on this broker. May be overridden per group via share.max.record.bytes.between.snapshots; per-group values are clamped to this ceiling."

New per-group dynamic config

share.max.record.bytes.between.snapshots

A new entry on ConfigResource.Type.GROUP, set via AdminClient.incrementalAlterConfigs.

| Name | share.max.record.bytes.between.snapshots |
| Type | LONG |
| Default | unset — falls back to share.coordinator.max.record.bytes.between.snapshots |
| Validator | atLeast(1). Values exceeding the current broker-level ceiling are rejected at incrementalAlterConfigs time with InvalidConfigurationException.   |
| Importance | LOW |
| Doc | "Specifies update-record bytes between snapshots for this share group. If unset, falls back to share.coordinator.max.record.bytes.between.snapshots. Values above the broker-level ceiling are rejected at the time of the alter request."

Deprecated Config

share.coordinator.snapshot.update.records.per.snapshot will be marked @Deprecated. If set on broker startup, a warning is logged. The config is honored for the deprecation period; if both records-based and bytes-based configs are set, the bytes-based config wins and a warning is logged. Removed in Kafka 5.0.

Resolution happens once at broker startup, in ShareCoordinatorConfig. If share.coordinator.max.record.bytes.between.snapshots is explicitly set, the shard runs in bytes mode. If only the records-based config is explicitly set, the shard runs in records mode.

Resolution of set vs. default. Because the bytes-based config has a 20 MB default, its resolved value alone cannot tell us whether an operator set it. We use AbstractConfig.originals().containsKey(...) — the established idiom in the codebase (see ProducerConfig idempotence resolution and GroupConfig's own Optional getters. At broker startup ShareCoordinatorConfig resolves the snapshot mode once:

1. If share.coordinator.max.record.bytes.between.snapshots is in originals() → bytes mode.
2. Else if share.coordinator.snapshot.update.records.per.snapshot is in originals() → records mode (deprecated path).
3. Else (neither explicitly set) → bytes mode at the 20 MB default.
4. If both are in originals() → bytes mode wins and a WARN is logged.

Java constants

Add the new bytes-based config in ShareCoordinatorConfig.java

public static final String MAX_RECORD_BYTES_BETWEEN_SNAPSHOTS_CONFIG = "share.coordinator.max.record.bytes.between.snapshots";

Add the new bytes-based per-group config in GroupConfig.java

public static final String SHARE_MAX_RECORD_BYTES_BETWEEN_SNAPSHOTS_CONFIG = "share.max.record.bytes.between.snapshots";

New metric

A new per-partition metric is added to the share coordinator, following the same convention as the existing last-pruned-offset metric (ShareCoordinatorMetrics):

│ Name        │ bytes-since-snapshot                                                                                                                             │
│ Group       │ share-coordinator-metrics                                                                                                                        │
│ Type        │ Value (gauge)                                                                                                                                    │
│ Tags        │ topic, partition                                                                                                                                 │
│ Description │ "The number of uncompressed update-record bytes written to this __share_group_state partition since the last snapshot record. Reset to zero when a snapshot   │
│             │ is written."                                                                                                                                     │

Full MBean name:

kafka.server:type=share-coordinator-metrics,topic=__share_group_state,partition=([0-9]+)

bytes-since-snapshot is exposed as a Value (gauge) attribute on this MBean (similar to last-pruned-offset)

Semantics : The value tracks accumulated update bytes for the share partitions hosted on a given __share_group_state partition. It rises as ShareUpdate records are written and drops after a ShareSnapshot is emitted.

The metric is tagged per __share_group_state partition, not per share group or per share partition. Because multiple share partitions (and groups) can map to the same state-topic partition, the value aggregates their in-flight update bytes.

Proposed Changes

Broker level config

share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorConfig.java

Add MAX_RECORD_BYTES_BETWEEN_SNAPSHOTS_CONFIG constant, register it in CONFIG_DEF, validate apply-time clamp to broker value.

Per-group config

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupConfig.java

Add SHARE_MAX_RECORD_BYTES_BETWEEN_SNAPSHOTS_CONFIG constant, register it in CONFIG_DEF, validate apply-time clamp to broker value.

Add bytes counter

In ShareCoordinatorShard.java

similar to snapshotUpdateCount, create another variable snapshotUpdateBytes
increment the update. (around line 282) and the existing snapshotUpdateCount increment stays.

// In replay(...), SHARE_UPDATE branch — full record is in scope here:
  case SHARE_UPDATE:
      long updateBytes = recordSerde.recordSize(record);   // see "Serialized size" below
      handleShareUpdate((ShareUpdateKey) key, (ShareUpdateValue) messageOrNull(value), updateBytes);
      break;

- Accumulate in handleShareUpdate (the existing snapshotUpdateCount increment at ~line 282 stays):

 snapshotUpdateBytes.compute(mapKey, (k, v) -> v == null ? updateBytes : v + updateBytes);

- Check the byte limit in the write path (around line 662, via the snapshotThresholdReached(key) helper):

 if (snapshotUpdateBytes.getOrDefault(key, 0L) >= maxBytesBetweenSnapshots) { ... }

- Also update tombstone handling to clear the new map (similar to snapshotUpdateCount.remove(mapKey)).

Also update tombstone handling. (similar to snapshotUpdateCount.remove(mapKey))

Serialized size

The serialized size in bytes of the full CoordinatorRecord (key + value) produced for a share update, as measured via ShareCoordinatorRecordSerde. This matches what the runtime appends to the __share_group_state log and is computed in ShareCoordinatorShard.replay(...) where the full record is in scope. Compression is not accounted for; the threshold is on uncompressed record bytes.

long updateBytes = recordSerde.recordSize(record); and we add recordSize to CoordinatorRecordSerde

Note - Both counters are maintained throughout the deprecation period. Keeping both counters preserves exact records-based behavior at the cost of one extra map write per update.

Propagate the override to ShareCoordinatorShard

share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorShard.java

Today the shard reads the broker-level value directly:

int updatesPerSnapshotLimit = config.shareCoordinatorSnapshotUpdateRecordsPerSnapshot();

This needs to change in two places: (1) the look-up path needs to consult the per-group config first and fall back to the broker default, and (2) the unit changes from records (int) to bytes (long).

Add a new accessor:

Accessor (group-coordinator side). Add maxRecordBytesBetweenSnapshotsOrDefault(String groupId, long brokerDefault) on ShareGroupConfigProvider, following the existing accessor pattern:

 public long maxRecordBytesBetweenSnapshotsOrDefault(String groupId, long brokerDefault) {
      return manager.groupConfig(groupId)
          .flatMap(GroupConfig::shareMaxRecordBytesBetweenSnapshots)
          .orElse(brokerDefault);
  }

This requires a corresponding Optional<Long>shareMaxRecordBytesBetweenSnapshots() getter on GroupConfig.

Interface (share-coordinator side). ShareGroupConfigProvider lives in group-coordinator, so share-coordinator does not reference it. Instead, share-coordinator defines the ShareGroupConfigLookup interface (see Module dependency), and the shard depends only on that:

public interface ShareGroupConfigLookup {
      long maxRecordBytesBetweenSnapshotsOrDefault(String groupId, long brokerDefault);
  }

Inject the lookup into ShareCoordinatorShard

The shard is constructed via its inner Builder (ShareCoordinatorShard.java).

The Builder constructor takes ShareCoordinatorConfig. Add a new optional setter alongside the existing ones (withTime, withCoordinatorMetrics, withTopicPartition, etc.):

Add an optional setter:

private ShareGroupConfigLookup shareGroupConfigLookup;                                                                                     
public Builder withShareGroupConfigLookup(ShareGroupConfigLookup lookup) {
    this.shareGroupConfigLookup = lookup;
    return this;
  }

And wire it through the build() method to a new field on the shard itself:

private final ShareGroupConfigLookup shareGroupConfigLookup;

Update both ShareCoordinatorShard constructors (ShareCoordinatorShard.java around line 176) to accept and store the lookup, and pass it through build().

Wire the lookup into ShareCoordinatorService.Builder

Add a withShareGroupConfigLookup(ShareGroupConfigLookup) setter to ShareCoordinatorService.Builder, store it as a field. Update the CoordinatorShardBuilderSupplier.

Pass the lookup in from BrokerServer.scala

BrokerServer constructs an adapter that implements ShareGroupConfigLookup, and passes it via .withShareGroupConfigLookup(...) on ShareCoordinatorService.Builder

Use the lookup at the snapshot-decision sites

ShareCoordinatorShard decides between writing a ShareSnapshot and a ShareUpdate at two sites, both governed by the same threshold. Both are updated to consult the startup-resolved mode via a single helper.

Write path — generateShareStateRecord (ShareCoordinatorShard.java)

// Before
  int updatesPerSnapshotLimit = config.shareCoordinatorSnapshotUpdateRecordsPerSnapshot();
  ...
  if (snapshotUpdateCount.getOrDefault(key, 0) >= updatesPerSnapshotLimit) { ... write snapshot ... }

  // After
  if (snapshotThresholdReached(key)) { ... write snapshot ... }

Reset path — handleShareSnapshot (ShareCoordinatorShard.java)

The same helper decides when the accumulated counter for key should be reset after a snapshot record is replayed:

  if (snapshotThresholdReached(key)) {
      // reset the active counter for this key
  }

snapshotThresholdReached(SharePartitionKey key) encapsulates mode selection so both sites stay in sync. The mode is resolved once at startup (see Deprecated config resolution); the helper does not re-resolve per call:

private boolean snapshotThresholdReached(SharePartitionKey key) {
      if (snapshotMode == SnapshotMode.BYTES) {
          long brokerLimit = config.shareCoordinatorMaxRecordBytesBetweenSnapshots();
          long limit = shareGroupConfigLookup.maxRecordBytesBetweenSnapshotsOrDefault(key.groupId(), brokerLimit);
          return snapshotUpdateBytes.getOrDefault(key, 0L) >= limit;
      } else { // RECORDS (deprecated path)
          return snapshotUpdateCount.getOrDefault(key, 0) >= config.shareCoordinatorSnapshotUpdateRecordsPerSnapshot();
      }
  }

- In bytes mode, the per-group override (share.max.record.bytes.between.snapshots) is consulted first via shareGroupConfigLookup, falling back to the broker-level share.coordinator.max.record.bytes.between.snapshots. The per-group value is already bounded at alter time to not exceed the broker ceiling.
- In records mode (only the deprecated config is set), the threshold is the broker records limit; there is no per-group records override.

shareGroupConfigLookup is the ShareGroupConfigLookup interface defined in share-coordinator and is required — ShareCoordinatorShard.Builder.build() rejects a null lookup, mirroring the existing null-checks in ShareCoordinatorService.Builder.build() (ShareCoordinatorService.java: around line 190). This avoids a silent no-op if the wiring in BrokerServer is ever omitted.

Emit the bytes-since-snapshot metric

New metric is wired through the existing ShareCoordinatorMetrics / ShareCoordinatorMetricsShard

Module dependency

share-coordinator needs to look up per-group config overrides, which live in group-coordinator.

Config lookup interface - share-coordinator defines a small interface, ShareGroupConfigLookup. BrokerServer constructs an adapter over the existing ShareGroupConfigProvider and passes it in, so the provider is adapted into the share-coordinator interface at the broker composition root.

share-coordinator gains no new module dependency, preserving the layering that coordinator-common was created to protect.

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.

Compatibility, Deprecation, and Migration Plan

Cluster-roll backward compatibility

During a rolling upgrade, yes brokers run mixed versions. Each broker's shard independently decides snapshot cadence, from its own local config. A newer broker replaying records written by an older broker interprets them identically. So cadence may differ in the mid-roll, but there is no on-disk format change and no coordination required I see. A broker that hasn't yet had the new config set, will simply use the 20 MB default.

Behavioral compatibility:

- The default snapshot trigger changes from 500 records to 20 MB of update bytes. For typical workloads this means snapshots are written less frequently, reducing write amplification.
- Brokers that set share.coordinator.snapshot.update.records.per.snapshot continue to use the records-based threshold during the deprecation period. A warning is logged at startup.
- The new per-group config is opt-in.

Deprecation:

share.coordinator.snapshot.update.records.per.snapshot will be marked @Deprecated in this release and removed in a future major release. Operators are encouraged to migrate to the bytes-based config.

Migration:

Operators who don't touch any config get the new 20 MB default.

Operators who explicitly set the records-based config can either:
- Remove the records-based config and accept the new 20 MB default, or
- Set the new bytes-based config (share.coordinator.max.record.bytes.between.snapshots) to a value that matches their previous tuning intent.

Operators who set the records-based config to 0 (snapshot every write) cannot perfectly preserve that behavior under the bytes-based config (which has atLeast(1)); they should set the new config to a small value like 1 if they need snapshots after every update.

The deprecated records-based config continues to work during the deprecation period; if both records-based and bytes-based configs are set, the bytes-based config takes precedence and a warning is logged.

Test Plan

Unit Tests

- ShareCoordinatorConfigTest — bytes-based broker bounds, deprecation warning, both-configs-set precedence.

- ShareCoordinatorShardTest — counter accumulates serialized bytes, snapshot triggers at threshold, both read sites stay consistent, per-group override propagated

- GroupConfigTest — per-group bytes config validation, apply-time clamp to broker ceiling.

Performance investigation

Before the deprecated records-based config is removed, we will measure the snapshot-cadence change (snapshots become substantially less frequent for small update records — ~154 bytes measured) and its effect on recovery time and retained-log size, using an out-of-order-ack workload that exercises the snapshot threshold. We will also benchmark the per-update overhead (extra counter + record-size computation) against trunk before merge, and publish migration guidance.

Integration Tests

ShareCoordinatorIntegrationTest (covers end-to-end propagation of the per-group config)

Rejected Alternatives

Here are the rejected alternatives.

Raise broker max only. No per-group override

A single broker-level setting forces all share groups on the cluster to use the same value. Different groups have different write profiles. A low-traffic group benefits from a low value (fast recovery), while a high-traffic group benefits from a high value. Without a per-group override, operators must pick one value that compromises for all groups.
Additionally, since all groups on the same __share_group_state partition share log-pruning behavior, one group set very high can delay pruning for unrelated groups on the same partition.

Per-group override only. Keep broker max at 500.

The broker ceiling acts as a hard cap on every per-group value (per-group between(1, broker_value)). If we keep the broker max at 500 means, per-group overrides cannot exceed 500 either, so high-traffic groups cannot benefit from the per-group config. In this case operators would have to raise the broker ceiling immediately anyway to use the new per-group setting meaningfully. Basically the per-group config is not of much use without the ceiling raise.

Keep records-based control, just adjust bounds.

Records-based control is workload-dependent — the same N records can be very different bytes on different workloads, making disk impact unpredictable. Bytes-based control mirrors KRaft's well-established metadata.log.max.record.bytes.between.snapshots and gives operators a metric they can reason about directly

Space shortcuts

Child pages

Status

Motivation

Problems

Proposal

Public Interfaces

New broker-level config

New per-group dynamic config

Deprecated Config

Java constants

New metric

Proposed Changes

Broker level config

Per-group config

Add bytes counter

Propagate the override to ShareCoordinatorShard

Documentation

Compatibility, Deprecation, and Migration Plan

Cluster-roll backward compatibility

Behavioral compatibility:

Deprecation:

Migration:

Test Plan

Unit Tests

Integration Tests

Rejected Alternatives

Raise broker max only. No per-group override

Per-group override only. Keep broker max at 500.

Keep records-based control, just adjust bounds.

Space shortcuts

Child pages

KIP-1349: Bytes-based configurable snapshot frequency for share groups

Status

Motivation

Problems

Proposal

Public Interfaces

New broker-level config

New per-group dynamic config

Deprecated Config

Java constants

New metric

Proposed Changes

Broker level config

Per-group config

Add bytes counter

Propagate the override to ShareCoordinatorShard

Documentation

Compatibility, Deprecation, and Migration Plan

Cluster-roll backward compatibility

Behavioral compatibility:

Deprecation:

Migration:

Test Plan

Unit Tests

Integration Tests

Rejected Alternatives

Raise broker max only. No per-group override

Per-group override only. Keep broker max at 500.

Keep records-based control, just adjust bounds.