Using Log Compaction for Stateful Stream Processing

The ability to reconstruct application state after a failure or a scaling event is an operational necessity for any streaming platform. Log compaction turns an append-only Kafka log into a key-indexed snapshot of the latest value for every key, which lets stateful processors rebuild their state directly from Kafka instead of from an external database. For platform engineers and SREs running production clusters, the operational details matter: how the cleaner interacts with segment layout and disk I/O, how to configure it so it keeps pace with write throughput, and how to monitor it before a falling-behind cleaner exhausts a disk.

This page covers log compaction from an operational angle: how the cleaner works, how to configure compacted topics safely for high-throughput state stores, how to validate compaction health from JMX, and how compacted topics back Kafka Streams state stores. It assumes a working knowledge of Kafka’s storage layer. For a refresher on how topics, partitions, and segments form the foundation of data modeling, see Topics, Partitions & Data Modeling.

How Log Compaction Works Under the Hood

Log compaction is a continuous, partition-level operation run by the LogCleaner thread pool on each broker. The cleaner works on a partition’s closed (non-active) segments, retaining only the latest value for each unique key and discarding older records for that key. Records with a null value—tombstones—are also retained for a configurable window so that consumers see the deletion before the record disappears.

Each partition’s log is split into a tail (already compacted) and a head (the “dirty” portion written since the last clean), divided by the cleaner point. When the cleaner selects a partition, it scans the dirty head and builds an in-memory offset map of the latest offset for each key. It then recopies the log from the cleaner point, keeping only records whose offset matches the map entry for their key and skipping superseded records, into fresh segments that replace the originals. Compaction never reorders records and never changes an offset, so existing consumer offsets remain valid; consumers simply observe gaps where superseded records were removed.

This design has direct operational consequences. First, compaction is I/O-intensive: it reads the dirty portion of a log and writes replacement segments before the originals are deleted, so a heavily-written compacted topic can roughly double disk I/O during a cleaning pass. Second, the cleaner is best-effort. If dirty data accumulates faster than the cleaner can recopy it, the log grows and compaction falls behind. The kafka.log:type=LogCleanerManager,name=max-dirty-percent gauge is the key signal for this condition; watch it to catch a saturated cleaner before disk exhaustion.

The cleaner chooses which log to clean next by dirty ratio—the fraction of the log (by bytes) that lies in the dirty head. A log becomes eligible once its dirty ratio reaches the broker default of 0.5, set by log.cleaner.min.cleanable.ratio. Setting it lower means more frequent, smaller, more I/O-hungry cleans; setting it higher means fewer cleans but more wasted space (at 0.5, up to half the log can be duplicates). The thread pool size, log.cleaner.threads, bounds how many logs are cleaned concurrently; its default is 1. On brokers with multiple data directories, raising it (a common starting point is one thread per data disk) lets cleaning proceed in parallel across disks.

# Broker configuration for log compaction
log.cleaner.threads=4
log.cleaner.min.cleanable.ratio=0.5
log.cleaner.backoff.ms=15000
log.cleaner.io.max.bytes.per.second=1500000000

Note the broker key is log.cleaner.min.cleanable.ratio. The per-topic override (below) is min.cleanable.dirty.ratio—the two have deliberately different names, which is a frequent source of confusion.

log.cleaner.io.max.bytes.per.second throttles the cleaner so that the sum of its read and write I/O stays below this average. Its default is effectively unlimited (Double.MAX_VALUE), so an unthrottled cleaner can saturate disk and starve normal reads and writes. The value is a plain number of bytes per second (no 1.5G-style suffix); the example above caps the cleaner’s combined read+write throughput at 1.5 GB/s. Tune it to your storage subsystem rather than the figure here. log.cleaner.backoff.ms (default 15000) is how long a cleaner thread sleeps when no log is eligible, preventing a busy-wait.

Configuring Compacted Topics for Production

A compacted topic needs cleanup.policy=compact. Beyond that, a handful of topic-level configs govern compaction timeliness and data safety; the important ones are segment.ms, min.compaction.lag.ms, and delete.retention.ms.

min.compaction.lag.ms (default 0) is the minimum time a record stays in the uncompacted head before it can be compacted away. Raising it gives slow or briefly-offline consumers a window to read intermediate values before they are collapsed. A typical production value is tens of minutes to a few hours. Too low and a lagging consumer may miss intermediate states it depends on; too high and the uncompacted tail—and disk usage—grows. (Its counterpart, max.compaction.lag.ms, default unbounded, sets an upper bound that forces compaction of even mostly-clean logs; useful for GDPR-style deletion deadlines.)

# Create a compacted topic with production-safe settings
kafka-topics.sh --create \
  --bootstrap-server kafka1:9092 \
  --topic user-profiles \
  --partitions 12 \
  --replication-factor 3 \
  --config cleanup.policy=compact \
  --config min.compaction.lag.ms=3600000 \
  --config delete.retention.ms=86400000 \
  --config segment.ms=3600000 \
  --config min.cleanable.dirty.ratio=0.5

delete.retention.ms (default 86400000, 24 h) controls how long tombstones survive after the segment containing them becomes eligible for cleaning. Tombstones are records with a non-null key and a null value that signal a key’s deletion; consumers must see them to remove the key from their local state, so this window should exceed your maximum expected consumer downtime plus the time for a full state rebuild.

segment.ms (default 7 days) forces a segment roll after the given time even if segment.bytes hasn’t been reached. The cleaner only compacts closed segments, so smaller, more frequently-rolled segments mean the head is cleaned sooner and large segments don’t accumulate stale records. One hour (3600000) is a reasonable starting point for a high-throughput compacted topic—at the cost of more segment files.

Compaction is per-partition, so all records for a key must hash to the same partition for their old versions to be removed. A key that scatters related records across partitions leaves duplicates surviving compaction, defeating its purpose. Get the key right: see Keying Strategies for Ordered Message Processing.

Monitoring Compaction Health and Performance

Effective monitoring tracks broker-level cleaner metrics and per-partition log metrics. The cleaner publishes JMX metrics under kafka.log:type=LogCleanerManager and kafka.log:type=LogCleaner.

The headline metric is kafka.log:type=LogCleanerManager,name=max-dirty-percent, the highest dirty ratio across all compacted logs the broker manages. As it climbs toward 1.0 the cleaner is losing ground and disk usage will grow; sustained values above ~0.5–0.6 (the cleanable threshold) mean the cleaner is consistently saturated. The recopy metric lives under the other type—kafka.log:type=LogCleaner,name=cleaner-recopy-percent—and reflects how much of each thread’s last clean had to be recopied; a high value means a lot of live data per pass. Also watch kafka.log:type=LogCleanerManager,name=time-since-last-run-ms; if it grows without bound the cleaner threads are stuck or starved.

JmxTool was moved out of the kafka.tools package and the old alias was removed in Kafka 4.0, so invoke it as org.apache.kafka.tools.JmxTool:

# Read a cleaner gauge once over JMX
kafka-run-class.sh org.apache.kafka.tools.JmxTool \
  --object-name kafka.log:type=LogCleanerManager,name=max-dirty-percent \
  --jmx-url service:jmx:rmi:///jndi/rmi://kafka1:9999/jmxrmi \
  --one-time true

Per-partition size is exposed as kafka.log:type=Log,name=Size,topic=<t>,partition=<p> (total bytes of all segments for that partition). There is no per-partition dirty-ratio MBean—dirty ratio is a broker-internal cleaner concept surfaced in aggregate via max-dirty-percent—so use Size plus log-dir inspection to spot a partition whose log is growing abnormally. A single hot, ever-growing partition usually points at skewed keys; revisit your layout with Partitioning Strategies for High-Throughput Topics.

kafka-log-dirs.sh reports per-partition segment sizes and log-directory usage (its JSON contains size, offsetLag, and isFuture per partition—there is no dirty-ratio field):

# Per-partition sizes and log-dir usage for specific topics
kafka-log-dirs.sh --describe \
  --bootstrap-server kafka1:9092 \
  --topic-list user-profiles,user-sessions

Disk I/O is the other health axis. Watch utilization with iostat on brokers hosting compacted topics; if disks sit above ~80% busy during cleaning, either raise log.cleaner.io.max.bytes.per.second (only if the storage has headroom) or add brokers to spread the load.

# Monitor disk I/O on broker nodes
iostat -x 5

Reasonable alerts: max-dirty-percent sustained above 0.9 for more than 15 minutes, and time-since-last-run-ms climbing past a few cleaner cycles.

Integrating Compacted Topics with Kafka Streams

Kafka Streams backs its state stores with compacted topics. A stateful operation—an aggregation, a join, a windowed computation—is materialized in a local store (RocksDB by default) and durably backed by a compacted changelog topic, so a task that migrates to another instance can rebuild its store by replaying the changelog.

This matters for capacity planning. A changelog generally has the same partition count as the store it backs, and on a rebalance a newly-assigned task replays its changelog from the earliest offset to rebuild local state. Restore time scales with the compacted changelog’s size and the throughput of the broker serving it, so keeping changelogs well-compacted directly shortens recovery.

You can override the configs Streams uses when it creates internal (changelog and repartition) topics by adding topic configs under StreamsConfig.topicPrefix(...), which prepends the topic. prefix that Streams recognizes (KIP-173). A bare props.put("min.compaction.lag.ms", ...) is not picked up as a topic config and is silently ignored—use the prefix:

// Kafka Streams config, including defaults for internal topics
Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "user-profile-aggregator");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka1:9092,kafka2:9092");
props.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 1);
props.put(StreamsConfig.topicPrefix("min.compaction.lag.ms"), "3600000");
props.put(StreamsConfig.topicPrefix("delete.retention.ms"), "86400000");

NUM_STANDBY_REPLICAS_CONFIG is the lever for fast recovery. A standby replica keeps a warm copy of a store by consuming the same changelog; on failure Streams prefers to move the task to an instance that already holds a standby, avoiding a full changelog replay and cutting recovery time for large stores from minutes to near-instant.

For very large stores (tens of GB and up), weigh compacted-topic-backed state against an external store: compacted topics remove an external dependency but consume broker disk and can lengthen rebalances when a full restore is needed. Standbys, smaller segment.ms, and a healthy cleaner are the main mitigations.

Troubleshooting Common Compaction Issues

When compaction falls behind the symptoms are subtle until disk is nearly full, so diagnose from metrics rather than from the alert that fires last.

Increasing disk usage despite a stable write rate. The cleaner can’t keep up with dirty-data generation. Check max-dirty-percent; if it’s pinned high, the cleaner is saturated. Confirm log.cleaner.threads isn’t left at its default of 1 on a multi-disk broker—each data disk benefits from a dedicated cleaner thread—and check that the cleaner isn’t I/O-throttled below what the disks can sustain.

# Inspect a broker's cleaner-related configs (including defaults and sources)
kafka-configs.sh --bootstrap-server kafka1:9092 \
  --entity-type brokers --entity-name 0 \
  --describe --all | grep cleaner

Consumers missing updates after a restart. Usually min.compaction.lag.ms is too small: a consumer down longer than that window can have intermediate values compacted away before it reads them. Raise it to comfortably exceed worst-case downtime, and confirm delete.retention.ms is long enough for tombstones to reach every consumer.

High CPU during cleaning. Building the offset map is CPU- and memory-bound, especially for logs with many distinct keys, and the map size is capped by log.cleaner.dedupe.buffer.size. If the cleaner can’t map a whole dirty head at once it cleans in smaller increments, which raises overhead. Give it more buffer, or roll smaller segments (segment.bytes/segment.ms) so each pass handles fewer keys. KIP-280 added version/timestamp-based compaction, changing which record “wins” rather than the map’s footprint—useful when out-of-order writes would otherwise let a stale value survive.

Compaction stuck on one partition. Often a single oversized segment that takes a long time to clean, or an active segment that never rolls. Use kafka-log-dirs.sh to find the partition’s segment sizes, then shorten segment.ms/segment.bytes for that topic so the head rolls and becomes cleanable.

# Per-partition segment sizes on a specific broker
kafka-log-dirs.sh --describe \
  --bootstrap-server kafka1:9092 \
  --topic-list problematic-topic \
  --broker-list 0

Operational Best Practices and Capacity Planning

Disk sizing. A compacted topic’s steady-state size is roughly the live key set plus the uncompacted tail: ≈ (unique keys × avg value size) + (write rate × min.compaction.lag.ms). The second term is the head the cleaner hasn’t collapsed yet. Provision at least ~30% headroom for spikes and for the temporary double-write during segment replacement.

Partition count. Each partition is cleaned independently, so more partitions allow more parallel cleaning—but every partition adds segment-file and index overhead, and for Streams the changelog partition count is fixed by the store. Size partitions to your processing parallelism, not to the cleaner. The official log compaction documentation covers the design in detail.

Tombstone management. Tombstones occupy space until delete.retention.ms elapses. For delete-heavy workloads they can dominate disk usage, so emit explicit null values only when a key is genuinely removed, and make sure every consumer handles nulls. In Streams, propagating null from an aggregator deletes the key from the store and emits a tombstone downstream:

stream.groupByKey()
    .aggregate(
        () -> new UserProfile(),
        (key, value, aggregate) -> {
            if (value == null) {
                return null; // delete the key: emits a tombstone to the changelog
            }
            return aggregate.merge(value);
        },
        Materialized.as("user-profiles-store")
    );

Load testing. Validate a new compacted topic under representative load before production and watch max-dirty-percent and cleaner I/O while you do. kafka-producer-perf-test.sh can drive volume but cannot model a realistic key distribution: --payload-file supplies raw record payloads (one per line, sampled at random), not keys, and the tool offers no key-distribution control. To exercise compaction you need a purpose-built producer (or pre-generated payloads with embedded keys) that reproduces your real key cardinality and update skew.

# Volume test only — does not control key distribution
kafka-producer-perf-test.sh \
  --topic user-profiles \
  --num-records 10000000 \
  --record-size 1024 \
  --throughput 50000 \
  --producer-props bootstrap.servers=kafka1:9092

Configuration governance. Standardize compacted-topic configs (cleanup.policy=compact plus your lag, retention, and segment settings) as a template and enforce it, since a single missing min.compaction.lag.ms or delete.retention.ms can cause silent data loss for slow consumers. Audit live configs with kafka-configs.sh --describe and flag drift.

Disaster recovery. Because compaction discards superseded records, a compacted topic cannot replay the original event sequence from offset zero—it only yields the latest value per key. If you need event-level point-in-time recovery, keep a parallel non-compacted (or tiered) topic capturing the full history. For most stateful-processing use cases the compacted topic is the source of truth, so its replication factor and any cross-datacenter mirroring must reflect that criticality.

Done well, compacted topics let you rebuild application state entirely from Kafka and retire the external state store—fewer moving parts, and recovery that is a changelog replay rather than a cross-system restore. That self-contained state model is one of the platform’s defining operational advantages for stateful stream processing.