Implementing Log Compaction

1. Understanding Compaction Concept

Log compaction in Apache Kafka is a process that ensures the retention of the latest value for each key within a topic, even as older messages are removed due to retention policies. This mechanism is essential for maintaining a compacted view of the data, particularly useful in scenarios where only the most recent state of keys is required.

Aspect	Description
Purpose	Retains the latest message for each key in a topic, discarding older entries with the same key.
Use Case	State stores, configuration data, or any scenario where key-value pairs need to be maintained with the most recent value.
Retention Policy	Compaction is independent of the topic's `retention.ms` or `retention.bytes` settings; it is managed separately.
Message Format	Only `key` and `value` fields are considered for compaction. Messages without keys are not compacted.

Example: Compaction in Action

Assume a topic user-events with the following messages:

{ key: "user1", value: "login: 2023-04-01" }
{ key: "user2", value: "logout: 2023-04-01" }
{ key: "user1", value: "login: 2023-04-02" }
{ key: "user3", value: "login: 2023-04-01" }
{ key: "user2", value: "logout: 2023-04-02" }

After compaction, only the latest value for each key is retained:

{ key: "user1", value: "login: 2023-04-02" }
{ key: "user2", value: "logout: 2023-04-02" }
{ key: "user3", value: "login: 2023-04-01" }

Note: Compaction is a background process and does not affect message production or consumption directly.

Warning: Compaction can increase disk usage temporarily due to the retention of older messages until compaction completes.

2. Enabling Log Compaction

Property	Default Value	Description
log.compaction.enable	`true`	Enables or disables log compaction for a topic.

Example: Enabling Log Compaction on a Topic

kafka-topics.sh --create --topic user-events --partitions 3 --replication-factor 1 --config log.compaction.enable=true

Note: Log compaction is only effective for topics that have a retention.ms or retention.bytes setting that allows for deletion of older messages.

3. Understanding Compaction Process

Log compaction in Apache Kafka is a process that ensures the retention of the latest value for each key within a topic, even as older messages are removed due to retention policies. This is particularly useful for topics that store configuration data, user profiles, or any data where the most recent state is important.

Component	Description	Role in Compaction
Log Cleaner	Background thread responsible for cleaning log segments.	Initiates and manages the compaction process.
Log Segment	Physical file containing messages for a partition.	Undergoes compaction to remove older entries with same keys.
Key-Value Pairs	Messages stored with a key and value.	Compaction retains the latest value for each unique key.
Cleaner Threads	Number of threads used to perform compaction.	Determines parallelism and speed of compaction.

Example: Compaction Configuration

{"compaction": true, "retention.ms": 604800000, "max.compaction.lag.ms": 86400000}

Note: Compaction is enabled by setting compaction to true in the topic configuration.

Compaction Process Overview

Log Cleaner scans log segments periodically.
It identifies duplicate keys and removes older entries.
Only the latest value for each key is retained in the log.
Compaction can be resource-intensive and affects performance.

4. Understanding Tombstone Messages

Tombstone messages in Kafka are used to mark records as deleted, particularly in the context of log compaction. When a record is marked with a tombstone, it indicates that the key for that record should be considered deleted from the topic's log.

Field	Description	Value
key	The key of the record to be deleted	`user-123`
value	Null or empty value indicating deletion	`null`
timestamp	Timestamp when the tombstone was created	`1634567890000`

Example: Producing a Tombstone Message

const kafka = new Kafka({ brokers: ['localhost:9092'] });
const producer = kafka.producer();
await producer.connect();

await producer.send({
  topic: 'user-events',
  messages: [
    {
      key: 'user-123',
      value: null
    }
  ]
});

await producer.disconnect();

Note: Tombstone messages are essential for log compaction to work correctly. They allow Kafka to remove outdated versions of records while maintaining the integrity of the log.

5. Configuring Compaction Lag

Property	Default Value	Description
log.compaction.lag.ms	`5000`	Time in milliseconds that a log compaction can be delayed before it is applied. This setting controls how frequently the log compaction process runs.
log.compaction.lag.ms.min	`0`	Minimum allowed value for `log.compaction.lag.ms`.
log.compaction.lag.ms.max	`9223372036854775807`	Maximum allowed value for `log.compaction.lag.ms`.

Example: Setting Compaction Lag in Kafka Configuration

# Broker-level configuration
log.compaction.lag.ms=10000

# Topic-level override (if supported)
topic.log.compaction.lag.ms=5000

Note: A lower log.compaction.lag.ms value results in more frequent compaction, which can increase CPU and I/O usage. Adjust this setting based on your system's capacity and data retention requirements.

6. Setting Delete Retention

Property	Default Value	Description
log.retention.hours	`168`	Retention period for logs in hours. After this time, logs are deleted based on the retention policy.
log.retention.minutes	`-1`	Retention period for logs in minutes. Overrides `log.retention.hours` if set.
log.retention.ms	`-1`	Retention period for logs in milliseconds. Overrides `log.retention.minutes` if set.

Example: Configuring Log Retention

{
  "log.retention.hours": 72,
  "log.retention.ms": -1
}

Note: Setting log.retention.ms to -1 disables time-based retention, allowing logs to persist indefinitely.

7. Configuring Segment Size

Property	Default Value	Description
log.segment.bytes	`1073741824`	Maximum size of a single log segment file in bytes (1 GB)
log.segment.delete.delay.ms	`60000`	Delay before deleting log segments in milliseconds (1 minute)

Example: Setting Segment Size

{"log.segment.bytes": 2147483648, "log.segment.delete.delay.ms": 120000}

Note: Increasing log.segment.bytes can improve performance by reducing the number of files, but may increase recovery time after failures.

8. Setting Min Cleanable Ratio

Property	Default Value	Description
min.cleanable.ratio	`0.1`	Ratio of log size to be cleaned before compaction is triggered. Lower values mean more frequent cleaning.

Example: Configuring min.cleanable.ratio

{
  "min.cleanable.ratio": 0.2
}

Note: Setting this value too low can lead to excessive compaction overhead, while setting it too high may delay log cleanup and increase storage usage.

9. Configuring Compaction Thread

Property	Default Value	Description
compaction.thread.count	`1`	Number of background threads to use for log compaction. Increasing this value can improve performance when multiple topics are being compacted.
compaction.lag.ms	`5000`	Time in milliseconds that a log compaction thread will wait before checking if a topic needs compaction. Lower values mean more frequent checks but higher CPU usage.
compaction.max.pct.of.log	`0.5`	Maximum percentage of the log that can be compacted in a single pass. This is used to prevent long-running compaction from blocking other operations.

Example: Configuring Compaction Thread Count

{"compaction.thread.count": 2,"compaction.lag.ms": 10000}

10. Setting Compaction Buffer

Property	Default Value	Description
log.cleaner.io.buffer.size	`524288`	Size of the buffer used for cleaning logs. This affects performance and memory usage.
log.cleaner.io.buffer.count	`3`	Number of buffers used for cleaning logs. Increasing this can improve throughput.

Example: Configuring Compaction Buffer

{
  "log.cleaner.io.buffer.size": 1048576,
  "log.cleaner.io.buffer.count": 5
}

Note: Adjusting these values can impact the performance of log compaction. Larger buffers may improve throughput but increase memory consumption.

11. Monitoring Compaction Progress

Example: Check compaction status via Kafka Tool

kafka-run-class.sh kafka.tools.GetOffsetShell \
  --broker-list localhost:9092 \
  --topic my-topic \
  --time -1

Property	Value	Description
log.compaction.enable	`true`	Enables log compaction for a topic
min.compaction.lag.ms	`0`	Minimum age of a message before it can be compacted
max.compaction.lag.ms	`9223372036854775807`	Maximum age of a message before it is eligible for compaction

Note: Compaction progress can be monitored through Kafka's internal metrics and JMX beans.

Compaction Monitoring Metrics

kafka.log:type=Log,name=CompactionRate - Rate of compaction operations per second
kafka.log:type=Log,name=CompactionTimeMs - Time spent in compaction operations
kafka.log:type=Log,name=LogSize - Current size of the log segment