04 · Log Management, Retention & Compaction

Source: Apache Kafka 4.4.0-SNAPSHOT (git 04bfe7d, 2026-06-15), KRaft mode. Derived from source code, not copied from official documentation.

Chapter 03 described a single UnifiedLog, an ordered sequence of immutable segments. This chapter zooms out to the broker-wide lifecycle that surrounds those logs: LogManager, the registry of every UnifiedLog across all log.dirs (JBOD), which creates and deletes logs, recovers them in parallel on startup, and drives the background schedulers for flushing, checkpointing, and retention. It then covers the two reclamation strategies, size/time retention (cleanup.policy=delete, drop whole old segments) and log compaction (cleanup.policy=compact, keep only the latest record per key), and the asynchronous, crash-safe machinery (.deleted/.cleaned/.swap renames, the LogCleaner thread pool, the SkimpyOffsetMap, and the cleaner-offset checkpoint) that makes both safe in the presence of disk failure, truncation, and exactly-once transactions.

Role & responsibilities

LogManager is documented in its own Javadoc as "the entry point to the kafka log management subsystem ... responsible for log creation, retrieval, and cleaning" (storage/src/main/java/org/apache/kafka/storage/internals/log/LogManager.java:178). All read/write I/O is delegated down to the individual UnifiedLog instances; LogManager owns the cross-cutting concerns:

• Directory ownership & JBOD placement. It validates and locks each directory in log.dirs, tracks which are live vs. offline vs. cordoned, and places each new partition in the directory holding the fewest partitions.
• Startup recovery. A per-directory thread pool loads every partition log in parallel, recovering unflushed segments when the prior shutdown was unclean.
• Background schedulers. Five recurring tasks: retention cleanup, flush of dirty logs, recovery-point checkpointing, log-start-offset checkpointing, and async deletion of whole partition directories.
• Disk-failure handling. Reacting to a LogDirFailureChannel event by taking a whole directory offline without crashing the broker (unless it was the last one).
• Compaction orchestration. Owning the LogCleaner pool and mediating the race between retention and compaction (pausing the cleaner on a partition while retention or truncation runs).

Where it lives in the code

Core concepts & terminology

Live / offline / cordoned dir

A directory is live if validated and lockable; offline once an IOException takes it out of service (until restart); cordoned if administratively excluded from new-partition placement (cordonedLogDirs, LogManager.java:138).

Current vs. future log

currentLogs holds the serving replica; futureLogs holds a copy being built in a different dir during an intra-broker move (a -future directory) that will atomically replace the current log once it catches up (LogManager.java:92-96).

Recovery point

The offset up to which a log is known durably flushed; segments above it are re-validated on unclean restart. Checkpointed to recovery-point-offset-checkpoint.

Log start offset

The lowest offset still readable (raised by DeleteRecords or segment deletion). Checkpointed to log-start-offset-checkpoint so deleted data is not re-exposed after restart.

Clean / dirty / cleanable section

For a compacted log, segments below the cleaner checkpoint are clean; above it is dirty, split into a cleanable prefix and an uncleanable tail (active segment, anything past the last stable offset, anything younger than min.compaction.lag.ms) (LogCleaner.java:53-56).

Tombstone

A record with a non-null key and null value; treated as a delete during compaction. Retained for delete.retention.ms after the batch enters the clean section, then dropped.

Cleanable ratio

cleanableBytes / (cleanBytes + cleanableBytes), the cleaner's "dirtiness" estimate and the selection key for the filthiest log (LogToClean.java:45-54).

Data structures

In-memory state in LogManager

The registry is two concurrent maps plus several coordination structures (LogManager.java:91-142):

Cleaning state in LogCleanerManager

Two maps under one ReentrantLock (LogCleanerManager.java:84-100):

• inProgress: Map<TopicPartition, LogCleaningState>, the per-partition state machine.
• uncleanablePartitions: Map<String, Set<TopicPartition>>, partitions that threw during cleaning, keyed by log dir; excluded until restart or recovery.
• checkpoints: Map<File, OffsetCheckpointFile>, the cleaner-offset-checkpoint per dir, recording the first dirty offset (cleaner point) of each compacted partition.

On-disk: the offset checkpoint file

All three offset checkpoints (recovery point, log-start, cleaner) share one plain-text format (OffsetCheckpointFile.java:40-50): a version line (CURRENT_VERSION = 0), an entry-count line, then one topic partition offset triple per line:

Writes are crash-consistent: checkpoint.write ultimately calls Utils.atomicMoveWithFallback, which writes a temp file, fsyncs, renames, and flushes the parent directory (noted at LogManager.java:1054). The three checkpoint file names are constants: RECOVERY_POINT_CHECKPOINT_FILE, LOG_START_OFFSET_CHECKPOINT_FILE (LogManager.java:85-86), and LogCleanerManager.OFFSET_CHECKPOINT_FILE = "cleaner-offset-checkpoint" (LogCleanerManager.java:66).

The dedupe hash table, SkimpyOffsetMap

Compaction's working memory is a fixed-size open-addressing hash table that stores a cryptographic hash of each key (not the key itself) alongside the latest offset for that key (SkimpyOffsetMap.java:28-88). Each slot is bytesPerEntry = hashSize + 8 bytes; with the default MD5 (16-byte digest) that is 24 bytes per entry, so a per-thread buffer holds memory / 24 slots.

This design trades a tiny false-collision probability (two distinct keys with the same MD5 → one wrongly dropped) for very low memory per key. The Javadoc on CleanerConfig spells out the tradeoff: a higher load factor "will allow more log to be cleaned at once but will lead to more hash collisions" (CleanerConfig.java:69). The buffer is allocated once per Cleaner at config.dedupeBufferSize / config.numThreads bytes (LogCleaner.java:480), capped at 2 GiB.

Architecture & control flow

Startup: validate, lock, recover, schedule

Construction calls createAndValidateLogDirs then lockLogDirs then loadDirectoryIds (LogManager.java:229-232). Validation creates missing dirs, rejects duplicates by canonical path, and routes any failing dir to the failure channel rather than aborting, unless every dir fails, in which case the broker halts (LogManager.java:376-379). Locking acquires an exclusive FileLock on each dir's .lock file (LogManager.java:449-462).

loadLogs (LogManager.java:608) recovers logs per directory, in parallel:

1. For each live dir it spins up a fixed pool of num.recovery.threads.per.data.dir threads named log-recovery-<dirPath>-<n> (LogManager.java:590-591,625).
2. It reads the dir's CleanShutdownFileHandler. If the clean-shutdown marker exists, it is deleted immediately and hadCleanShutdown is cached, so a crash mid-load is correctly treated as unclean next boot (KAFKA-10471, LogManager.java:629-636).
3. It reads the recovery-point and log-start checkpoints (tolerating a corrupt checkpoint by resetting to 0 / first segment).
4. It lists topic-partition subdirectories, skipping the remote-log-index-cache dir and the cluster-metadata topic (LogManager.java:653-660), and submits one loadLog job per partition. With a clean shutdown, segment recovery is skipped entirely; otherwise every segment above the recovery point is re-validated.
5. loadLog (LogManager.java:511) opens each UnifiedLog and routes it: a -delete-suffixed dir → logsToBeDeleted; a -stray dir → logged; a still-present-but-orphaned replica → renamed -stray; otherwise inserted into currentLogs/futureLogs (a duplicate is a fatal IllegalStateException).

Two gauges, remainingLogsToRecover (per dir) and remainingSegmentsToRecover (per recovery thread), let operators watch progress (LogManager.java:741-753). After loading, startupWithConfigOverrides schedules the five tasks and, if cleanerConfig.enableCleaner, builds and starts the LogCleaner (LogManager.java:811-841).

Detailed mechanics, retention (cleanup.policy=delete)

The kafka-log-retention task runs cleanupLogs every log.retention.check.interval.ms (default 5 minutes, ServerLogConfigs.LOG_CLEANUP_INTERVAL_MS_DEFAULT). It first pauses the cleaner on all non-compacted partitions to avoid a race, iterates them calling UnifiedLog.deleteOldSegments() (and the same on any future log), then resumes (LogManager.java:1705-1744).

cleanupLogs():
  deletable = cleaner.pauseCleaningForNonCompactedPartitions()   // non-compact logs, now Paused(1)
  for (tp, log) in deletable:
      total += log.deleteOldSegments()
      futureLogs[tp]?.deleteOldSegments()
  finally: cleaner.resumeCleaning(deletable.keySet())

UnifiedLog.deleteOldSegments() (UnifiedLog.java:1967) dispatches on policy. For cleanup.policy=delete it runs three predicates in order, log-start-offset breach, then size breach, then time breach:

A segment's "age" is largestTimestamp(): the segment's max record timestamp if one exists, else its file lastModified time (LogSegment.java:861-866). Future-dated records are detected and logged but still block deletion (a segment with a future timestamp is never "old enough", UnifiedLog.java:2008-2009).

deletableSegments walks segments oldest-first and stops at the first that fails the predicate (UnifiedLog.java:1879-1914). Two crucial guards:

• High-watermark guard. A segment is deletable only if highWatermark() ≥ upperBoundOffset, never delete data at or above the HW, so the log start offset can never exceed the HW (UnifiedLog.java:1892-1894).
• Never-empty guard. The final segment, if empty, is never returned; and if deletion would remove all segments, deleteSegments first rolls a fresh active segment so the log always has at least one (UnifiedLog.java:1933-1942).

On deletion, deleteSegments advances the (local) log-start offset to the base of the next surviving segment before removing the segment from the index, then calls localLog.removeAndDeleteSegments and prunes producer snapshots (UnifiedLog.java:1944-1955).

The asynchronous .deleted rename

Physical deletion is deferred and crash-safe (LocalLog.deleteSegmentFiles, LocalLog.java:846-877):

The default delay is file.delete.delay.ms = 60000 (ServerLogConfigs.LOG_DELETE_DELAY_MS_DEFAULT, wired as the default for TopicConfig.FILE_DELETE_DELAY_MS_CONFIG at LogConfig.java:224). If asyncDelete is false the unlink runs inline (LocalLog.java:872-876). Any leftover .deleted files after a crash are removed during recovery in LogLoader (LocalLog.java:982-983).

Deleting a whole partition

When a partition is removed from the broker, asyncDelete renames its directory to <topic>-<p>.<uuid>-delete and enqueues it (LogManager.java:1589-1625). The kafka-delete-logs task (deleteLogs, LogManager.java:1415) drains logsToBeDeleted, deleting each only after file.delete.delay.ms has elapsed since it was enqueued, and self-reschedules for the remaining time of the next-due entry (nextDeleteDelayMs, LogManager.java:1397-1406,1436). Before enqueueing it aborts any cleaning of that partition and updates checkpoints (LogManager.java:1597-1617).

Detailed mechanics, compaction (cleanup.policy=compact)

Compaction's contract (LogCleaner.java:50-51): "A message with key K and offset O is obsolete if there exists a message with key K and offset O' such that O < O'." The cleaner keeps, for every key, only the record at its highest offset (plus tombstones within their retention window).

Choosing the filthiest log

Each CleanerThread.doWork() calls tryCleanFilthiestLog; if nothing was cleaned it sleeps log.cleaner.backoff.ms (default 15 s) (LogCleaner.java:512-523). Selection happens in LogCleanerManager.grabFilthiestCompactedLog under the manager lock (LogCleanerManager.java:239):

1. Read all cleaner checkpoints (allCleanerCheckpoints), the per-partition first-dirty offset.
2. For every compacted log not already in-progress and not uncleanable, compute cleanableOffsets → (firstDirtyOffset, firstUncleanableDirtyOffset), and build a LogToClean (which computes cleanBytes, cleanableBytes, and the cleanable ratio).
3. Keep only logs that either need compaction now (forced by max.compaction.lag.ms) with non-zero cleanable bytes, or whose cleanable ratio exceeds the per-topic min.cleanable.dirty.ratio (LogCleanerManager.java:281-283).
4. Pick the maximum cleanable ratio; mark it LOG_CLEANING_IN_PROGRESS and return it.

cleanableOffsets computes the first uncleanable offset as the minimum of three bounds (LogCleanerManager.java:730-742):

• the last stable offset (never clean past the LSO, see transactions below);
• the active segment's base offset (the active segment is always uncleanable);
• if min.compaction.lag.ms > 0, the base offset of the first segment whose largestTimestamp > now − minCompactionLagMs (findFirstUncleanableSegment, LogCleanerManager.java:767-783).

It also self-heals a bad checkpoint: if the checkpointed dirty offset is below the log start or above the log end, it resets firstDirtyOffset to the log start and flags forceUpdateCheckpoint (LogCleanerManager.java:702-722). max.compaction.lag.ms forces cleaning even of a barely-dirty log: maxCompactionDelay measures how long the earliest dirty segment has waited beyond the max-lag and sets needCompactionNow when positive (LogCleanerManager.java:669-680).

Pass 1, build the offset map

Cleaner.doClean (Cleaner.java:159) first calls buildOffsetMap over the dirty section [firstDirtyOffset, firstUncleanableOffset) (Cleaner.java:710). For each dirty segment it iterates batches, and for each non-control, non-aborted record with a key at or above the start offset, it puts key → offset into the SkimpyOffsetMap (Cleaner.java:807-817). Because later offsets overwrite earlier ones, the map ends holding each key's highest offset in the dirty range. It stops adding once the map reaches slots × loadFactor entries and returns "full", so a single clean may cover only part of the dirty range; endOffset = map.latestOffset() + 1 becomes the new cleaner point (Cleaner.java:178,810-814).

Pass 2, recopy segments keeping only survivors

Segments from offset 0 up to endOffset are grouped by groupSegmentsBySize so a group's combined log and index bytes stay under segment.bytes / segment.index.bytes and the offset span stays under Integer.MAX_VALUE, this merges tiny segments and "prevents segment sizes from shrinking too much" with repeated cleanings (Cleaner.java:641-676). Each group is rewritten by cleanSegments into one or more .cleaned segments (Cleaner.java:228), and finally swapped in via UnifiedLog.replaceSegments (Cleaner.java:328).

The retain/discard decision for each record is shouldRetainRecord (Cleaner.java:567-602):

• If record.offset() > map.latestOffset() the record is past the mapped range → always retain (it is beyond what this pass deduplicated).
• Else look up the key. Retain only if this is the latest offset for the key (record.offset() ≥ foundOffset) and the record either has a value or is a tombstone still within its delete-retention window.
• A record with no key is invalid in a compacted log; it is dropped and counted in stats.invalidMessagesRead (Cleaner.java:598-601).

The atomic, crash-safe swap

UnifiedLog.createNewCleanedSegment writes to *.cleaned files. replaceSegments (LocalLog.java:1004-1062) then performs a two-phase swap designed to be recoverable at any crash point (the ordering described at LocalLog.java:974-990):

Tombstones and delete.retention.ms

A tombstone (null value) must survive long enough that a consumer reading from offset 0 still observes the delete before it disappears, otherwise it could see the pre-tombstone value and never learn the key was removed. The cleaner therefore retains a tombstone for delete.retention.ms (default 86 400 000 ms = 24 h, LogConfig.DEFAULT_DELETE_RETENTION_MS) measured from when its batch enters the clean section. For the modern v2 record format this is tracked by stamping a delete horizon into the batch header on the first clean that encounters it; the record is dropped only once currentTime ≥ batch.deleteHorizonMs() (Cleaner.java:589-596). For legacy (< v2) formats a coarser horizon based on the last clean segment's lastModified − deleteRetentionMs is used (Cleaner.java:166-170,260).

Interaction with transactions / markers

The cleaner's transaction handling is summarized at LogCleaner.java:81-96 and enforced in cleanInto (Cleaner.java:366):

• Never clean past the LSO. Because firstUncleanableOffset is bounded by lastStableOffset(), every record the cleaner sees is already committed or aborted, so the transaction index can be consulted up front (LogCleanerManager.java:730-733).
• Aborted-transaction records are dropped unconditionally, ignoring keys (CleanedTransactionMetadata drives discardBatchRecords, Cleaner.java:386-391).
• The last batch of each active producer is retained even if empty (RETAIN_EMPTY) to preserve sequence-number continuity; likewise the batch whose next offset equals the high watermark is retained so the last-offset information survives (Cleaner.java:394-420).
• Transaction markers (control batches) are retained until all records from their transaction are gone and a delete-horizon has passed, the same tombstone-retention logic, applied to markers (Cleaner.java:384-389).

Concurrency & threading

Locks. LogManager.logCreationOrDeletionLock (a plain monitor) serializes create/delete/rename and disk-failure handling. LogCleanerManager.lock (a ReentrantLock with a pausedCleaningCond condition) guards all access to inProgress, uncleanablePartitions, and the checkpoint map (LogCleanerManager.java:93-100). Retention runs under UnifiedLog's own lock while deleting segments (UnifiedLog.java:1840). I/O throttling across all cleaner threads is a single shared Throttler sized to log.cleaner.io.max.bytes.per.second (LogCleaner.java:168).

The cleaning state machine

LogCleaningState is a sealed interface with three cases (LogCleaningState.java:24-72). Transitions are all guarded by the manager lock:

This is how LogManager.truncateTo, truncateFullyAndStartAt, intra-broker moves, and partition deletion safely interleave with compaction: each calls abortAndPauseCleaning(tp), does its work, then resumeCleaning(tp) (LogManager.java:944-1003,1453-1455). If a clean threw an unexpected exception, tryCleanFilthiestLog marks the partition uncleanable so it is skipped until maintainUncleanablePartitions or a restart clears it (LogCleaner.java:542-551; LogCleanerManager.java:640-652).

Configuration reference

Retention / segment-lifecycle (topic-level keys; broker defaults via ServerLogConfigs):

Compaction (topic-level; cleaner defaults in CleanerConfig / LogConfig):

Broker-level cleaner & manager knobs (CleanerConfig.java, reconfigurable set at LogCleaner.java:101-109):

Failure modes, edge cases & recovery

Disk failure

Any class doing log I/O routes an IOException to LogDirFailureChannel.maybeAddOfflineLogDir, which records the dir once and enqueues it onto a bounded queue (LogDirFailureChannel.java:60-65). The single LogDirFailureHandler thread drains it and calls ReplicaManager.handleLogDirFailure → LogManager.handleLogDirFailure (LogManager.java:411), which under logCreationOrDeletionLock:

1. removes the dir from liveLogDirs and directoryIds, and halts the broker if it was the last live dir (LogManager.java:416-419);
2. drops the dir's recovery-point and log-start checkpoint maps;
3. tells the cleaner to stop cleaning that dir (drops its cleaner-offset checkpoint, LogCleanerManager.handleLogDirFailure);
4. removes and closes all current and future logs under that dir, marking those partitions offline;
5. destroys the dir's .lock.

An offline dir stays offline until broker restart (LogDirFailureChannel.java:36).

Truncation during compaction

If a log is truncated below the active segment's base while it is being cleaned, LogManager.truncateTo aborts and pauses cleaning, truncates, then maybeTruncateCleanerCheckpointToActiveSegmentBaseOffset rewinds the cleaner checkpoint if it now points past the data, before resuming (LogManager.java:944-967,1143-1147; LogCleanerManager.maybeTruncateCheckpoint:552-567). Mid-clean, checkDone raises LogCleaningAbortedException and cleanSegments deletes its partial .cleaned output (Cleaner.java:330-341).

Buffer too small for a batch

If a chunk read yields zero complete batches, the cleaner doubles its I/O buffers up to max(maxLogMessageSize, maxIoBufferSize) (a single oversized batch, e.g. compression pushing a set just over max.message.bytes, or a later reduction of that config, is still made to fit), failing only if the message genuinely exceeds the allowed maximum (Cleaner.java:508-617). A first write to an empty cleaned segment is always allowed to avoid an infinite overflow loop (Cleaner.java:469-472).

Unclean shutdown

If the clean-shutdown marker is absent (or was deleted mid-load), every segment above the recovery point is re-validated on startup. The marker is written at shutdown only for dirs that either had it before or whose logs were all recovered this run, and it carries the broker epoch for KRaft fencing (LogManager.java:905-914).

Invariants & guarantees

• No concurrent retention + compaction on a partition. Enforced by the inProgress state machine; retention pauses non-compacted partitions, and compaction marks its target InProgress before touching it.
• Compaction never crosses the LSO. firstUncleanableOffset ≤ lastStableOffset, so the cleaner only ever sees decided (committed/aborted) records.
• Latest value per key survives. The offset map keeps each key's highest offset; shouldRetainRecord retains only that record (or a within-window tombstone).
• Crash-atomic swap. The .cleaned → .swap → live sequence plus descending-offset renames make replaceSegments recoverable at any crash point.
• Deleted data is not re-exposed. The log-start-offset checkpoint is written before/at deletion so a restart does not reveal segments below the start offset.
• Producer sequence continuity. The last batch of each active producer (and the marker at the HW) is retained even when emptied by compaction.

Interactions with other subsystems

• 03 · The Log Storage Engine, UnifiedLog/LocalLog/LogSegment provide the segments this chapter creates, deletes, recopies, and swaps; replaceSegments and deleteSegmentFiles live there.
• 05 · Tiered Storage, when remote copy is enabled, retention switches to local.retention.ms/local.retention.bytes and only deletes local segments already uploaded (UnifiedLog.java:1857-1869,1979-1982).
• 08 · Replication, ISR & High Watermark, the high-watermark guard in deletableSegments and the LSO bound on compaction tie reclamation to replication progress.
• 14 · Transactions & Exactly-Once, the cleaner consults the transaction index, drops aborted records, and retains markers/last-producer batches.
• 12 · Metadata Propagation & Broker Lifecycle, stray-replica detection (isStrayReplica) and intra-broker moves (future logs) are driven by the metadata image and the JBOD directory assignment.
• 13 · Group Coordination, the internal __consumer_offsets topic is compacted; the cleaner must run for offset/group state not to grow unbounded (the deprecation note warns about exactly this, CleanerConfig.java:73-74).
• 07 · Request Processing, DeleteRecords raises the log start offset, which then drives deleteLogStartOffsetBreachedSegments.

Design rationale & evolution

The cleaner is being simplified for Kafka 5.0: log.cleaner.enable is deprecated and the cleaner will always run (LogManager.java:838-841, CleanerConfig.java:53). Reconfiguration of cleaner threads (LogCleaner implements BrokerReconfigurable) shuts down and recreates the pool, which also resurrects any threads that had died (LogCleaner.java:289-301).

Gotchas & operational notes