Log compaction preserves the newest record for each key while removing older records with the same key. It suits keyed updates such as stock prices, where an application restarting needs the latest value for every symbol rather than the full update history. section 2.6.2
Use compaction when the latest keyed state matters more than every historical update:
- Enable it per topic with cleanup.policy=compact; a background log-cleaner pool recopies segments and removes superseded records. section 2.6.2
- Combine compact and delete when a large key space could otherwise let compacted segments keep growing. section 7.4.7
- Treat a null value as a deletion marker when permanently removing a key from a Kafka Streams state store. section 7.4.7
Recovering current prices
A topic receives updates such as AAPL=190, MSFT=410, and later AAPL=192. After compaction, the log retains AAPL=192 and MSFT=410, allowing a restarted application to rebuild current prices. section 2.6.2 section 8.1.1
Listing 7.22 Setting a cleanup policy using Materialized
Map<String, String> changeLogConfigs = new HashMap<>();
changeLogConfigs.put("cleanup.policy", "compact,delete");
builder.stream("input")
.groupByKey()
.count(Materialized.as("counting-store")
.withLoggingEnabled(changeLogConfigs)) #1
.toStream()
.to("output")
Configure a Kafka Streams changelog with compact and delete policies through Materialized.withLoggingEnabled. section 7.4.7
