Heuristic:Apache Kafka Coordinator Loading Commit Interval
| Knowledge Sources | |
|---|---|
| Domains | Optimization, Coordinator_Runtime |
| Last Updated | 2026-02-09 12:00 GMT |
Overview
The coordinator loader commits offsets every 16,384 records as a measured trade-off between loading speed and memory consumption during partition loading.
Description
When a coordinator partition is loaded, `CoordinatorLoaderImpl` reads records from the log and replays them. During this process, it periodically commits the last written and committed offsets. The frequency of these commits directly impacts performance: committing too often slows down loading (especially for simple workloads), while committing too rarely allows excessive temporary data to accumulate in memory (especially for workloads that create collections participating in `CoordinatorPlayback` snapshotting). The value of 16,384 was empirically chosen as a balanced trade-off. Additionally, a two-tier commit strategy is used: the loader commits when reaching the high watermark OR at regular intervals, preventing both memory bloat and consistency issues.
Usage
Consider this heuristic when tuning coordinator partition loading performance or when debugging slow coordinator startups. If loading is too slow, the commit interval might be too small. If memory usage spikes during loading, the interval might be too large. The code explicitly instructs developers to run `GroupCoordinatorShardLoadingBenchmark` before changing this value.
The Insight (Rule of Thumb)
- Action: Use `DEFAULT_COMMIT_INTERVAL_OFFSETS = 16384` as the commit interval during coordinator loading.
- Value: 16,384 offsets between commits.
- Trade-off: Smaller values increase commit frequency (slower loading, lower memory). Larger values decrease commit frequency (faster loading, higher memory).
- Validation: Always run `GroupCoordinatorShardLoadingBenchmark` before changing this value.
- Two-tier Strategy: Commit at the high watermark boundary OR every `commitIntervalOffsets` records, whichever comes first.
Reasoning
The 16,384 value was determined through benchmarking against two competing workload profiles:
Simple workloads (no snapshotting collections): These benefit from fewer commits, as each commit has overhead. Larger intervals let the loader process more records without interruption.
Complex workloads (many temporary collections for snapshotting): These benefit from more frequent commits, as uncommitted data accumulates in memory and must be held until the next snapshot point.
The two-tier commit strategy (lines 298-314) handles a subtle race condition: the high watermark continues to advance during loading, so the loader must commit when it catches up to the watermark and at regular intervals to prevent unbounded memory growth.
Code evidence from `CoordinatorLoaderImpl.java:53-65`:
/**
* The interval between updating the last committed offset during loading, in offsets. Smaller
* values commit more often at the expense of loading times when the workload is simple and does
* not create collections that need to participate in CoordinatorPlayback snapshotting.
* Larger values commit less often and allow more temporary data to accumulate before the next
* commit when the workload creates many temporary collections that need to be snapshotted.
*
* The value of 16,384 was chosen as a trade-off between the performance of these two workloads.
*
* When changing this value, please run the GroupCoordinatorShardLoadingBenchmark to evaluate
* the relative change in performance.
*/
public static final long DEFAULT_COMMIT_INTERVAL_OFFSETS = 16384;
Two-tier commit logic from `CoordinatorLoaderImpl.java:298-314`:
// Note that the high watermark can be greater than the current offset but as we load more records
// the current offset will eventually surpass the high watermark. Also note that the high watermark
// will continue to advance while loading.
currentOffset = batch.nextOffset();
long currentHighWatermark = log.highWatermark();
if (currentOffset >= currentHighWatermark) {
coordinator.updateLastWrittenOffset(currentOffset);
if (currentHighWatermark > lastCommittedOffset) {
coordinator.updateLastCommittedOffset(currentHighWatermark);
lastCommittedOffset = currentHighWatermark;
}
} else if (currentOffset - lastCommittedOffset >= commitIntervalOffsets) {
coordinator.updateLastWrittenOffset(currentOffset);
coordinator.updateLastCommittedOffset(currentOffset);
lastCommittedOffset = currentOffset;
}