Kafka Log Compaction: How to Enable and Verify State‑Store Topics

Desired Outcome

Configure a Kafka topic to retain only the most recent value for each key (log compaction), verify that older records are removed, and ensure consumers receive the correct state, including handling tombstone deletions.

Prerequisites

• Kafka cluster running 3.x or newer.
• Admin access to the broker or cluster (e.g., kafka-topics.sh with --zookeeper or --bootstrap-server).
• Kafka client installed on the machine where you run diagnostic tools.
• JMX access to the broker for metrics inspection.

Step 1 – Enable Log Compaction

Log compaction is enabled by setting the topic configuration cleanup.policy=compact. You can set it at topic creation or alter an existing topic.

# Create a new topic with compaction enabled
kafka-topics.sh \
  --bootstrap-server <broker> \
  --create \
  --topic state-store \
  --partitions 3 \
  --replication-factor 3 \
  --config cleanup.policy=compact \
  --config min.cleanable.dirty.ratio=0.1

To alter an existing topic:

# Alter cleanup policy for an existing topic
kafka-configs.sh \
  --bootstrap-server <broker> \
  --entity-type topics \
  --entity-name state-store \
  --alter \
  --add-config cleanup.policy=compact

Verify the configuration:

# Describe the topic configuration
kafka-topics.sh \
  --bootstrap-server <broker> \
  --describe \
  --topic state-store

Look for cleanup.policy=compact in the output.

Step 2 – Produce Test Data

Send a series of messages with the same key and different values. Include a tombstone (null value) to test deletion handling.

# Produce messages with the same key
kafka-console-producer.sh \
  --bootstrap-server <broker> \
  --topic state-store \
  --property parse.key=true \
  --property key.separator=,
# In the producer prompt, type:
key1,first
key1,second
key1,third
# Produce a tombstone
key1,
# Send a message with a null key (will not be compacted)
,nullValue

Note: Messages with a null key are exempt from compaction and will remain in the log.

Step 3 – Verify Compaction Has Occurred

Compaction is performed asynchronously by the broker’s Log Cleaner thread. You can confirm it by inspecting the log segments and checking metrics.

3.1 Inspect Log Segments

Use DumpLogSegments to view the contents of each segment file. Replace <log-dir> with the actual directory path for your broker’s log.

# Dump the contents of the topic’s log directory
kafka-run-class \
  org.apache.kafka.tools.DumpLogSegments \
  --files <log-dir>/state-store-*/

After compaction, you should see only the latest record for key1 (value "third") and the tombstone record if it has not yet been cleaned. The null‑key record should remain.

3.2 Monitor LogCleaner Metrics via JMX

Connect to the broker’s JMX port (default 9999) and query the following MBeans:

• kafka.server:type=LogCleaner,name=CleanerThreadRate – shows how often the cleaner runs.
• kafka.server:type=LogCleaner,name=BytesCompacted – cumulative bytes removed.
• kafka.server:type=LogCleaner,name=RecordsCompacted – cumulative records removed.

Tools like jconsole or jmxterm can be used. Example with jmxterm:

jmxterm -l localhost:9999 -n -f
[localhost:9999] get kafka.server:type=LogCleaner,name=BytesCompacted

Step 4 – Common Pitfalls & Recovery Options

• Null Key Messages: These are not compacted. If you see unexpected growth, filter or remove them.
• Tombstone Retention: Tombstones are retained for delete.retention.ms (default 24 h). If consumers need to see the deletion sooner, reduce this value.
• min.cleanable.dirty.ratio: If set too high, the cleaner may not trigger. Lower it (e.g., 0.1) to force compaction earlier.
• Segment Size: Small segments may delay compaction. Ensure segment.ms or segment.bytes are configured to produce larger segments.
• Consumer Lag: If a consumer reads the old value before compaction completes, it may process stale data. Use consumer offsets or a compacted‑only consumer group to avoid this.

Step 5 – Rollback (If Needed)

To disable compaction, reset the policy to delete or leave it unspecified:

# Disable compaction
kafka-configs.sh \
  --bootstrap-server <broker> \
  --entity-type topics \
  --entity-name state-store \
  --alter \
  --add-config cleanup.policy=delete

After disabling, the broker will revert to time‑based or size‑based retention. Existing compacted data will remain until the retention period expires.

Practical Checklist

Task	Command / Action	Verification
Set cleanup.policy to compact	kafka-configs.sh --alter ...	kafka-topics.sh --describe
Produce test messages	kafka-console-producer.sh	Consume with kafka-console-consumer.sh
Inspect segments	kafka-run-class DumpLogSegments	Only latest key present
Check LogCleaner metrics	jmxterm or jconsole	BytesCompacted > 0
Tune delete.retention.ms	kafka-configs.sh --alter	Observe tombstone removal

Follow this guide to confidently enable log compaction for state‑store topics, validate its behavior, and address common complications. Remember to test in a staging environment before applying changes to production.