Digging into Kafka's Source 7: Log Storage — Segments, Indexes, Retention and Compaction

What this post covers

This is the seventh post in the "Digging into Kafka's Source" series. In part 6 we followed a Produce request through KafkaApis → ReplicaManager → Partition → UnifiedLog.append() until it was written with FileChannel.write() into the active segment's .log file. This time we go down into those files: which files make up one partition's log on disk, and when they are created and removed.

A zoom into zone E (log storage) of the map from part 2. The vertical line in the middle is the path a write request travels down; the blue boxes on top and on the side are background work that runs regardless of requests.

There are four questions to answer.

• What are the 00000000000000003830.log, .index and .timeindex files in a partition directory, and exactly when is a new segment created (log.segment.bytes, segment.ms)?
• When a request asks for offset 35, how does the broker find the byte in the file where that record lives? Why doesn't the index list every record?
• Who checks retention.ms and retention.bytes, when, and in what order are segments deleted?
• What does cleanup.policy=compact keep and what does it remove? And what does a broker re-read after it suddenly dies and comes back?

All line numbers in this post refer to this commit. Part 1 shows how to check out the same commit and build the Docker image used for the examples. Until a few years ago the classes in this post were Scala code in the core module (with names like kafka.log.LogSegment); they have all moved to Java in the storage module. core/src/main/scala/kafka/log/LogManager.scala is a 72-line shell. So there is no Scala in this post.

I also compared against 4.4.0, the latest stable release most people run. LogSegment, LogSegments, OffsetIndex, TimeIndex, LazyIndex, LogFileUtils, LogLoader, LogCleaner, LogCleanerManager and SkimpyOffsetMap are the same files as in the 4.4.0 tag. There are three differences, pointed out in the sections where they matter (future timestamps in section 4, overflow during cleaning in section 5, and a size check on follower appends that this post does not cover).

UnifiedLog.java alone is 2,806 lines, and LogManager.java is 1,895. Tiered storage (moving old segments to remote storage such as S3) and multiple log directories (JBOD) are mixed into the same files. This post picks only the main path a newcomer can follow; what I left out is listed by name and location in "What this post leaves out" at the end.

Terms to know first

A log in Kafka means "a line of records that is only ever appended to". It shares its name with log files such as the server's server.log, which is confusing; in this post "log" means the data of one partition. One partition has one log, which is one UnifiedLog object in the source and one <topic>-<partition> directory (e.g. seg-0) under log.dirs on disk.

A segment is one bundle of files that a log has been cut into. If a log were a single file, removing only its old beginning would be hard. So a new file is started every so many bytes or so much time, and deletion removes whole segments. Each segment has a .log (the record batches themselves), a .index (offset → file position), a .timeindex (time → offset) and a .txnindex (the list of aborted transactions, part 12). The last segment, the one being written now, is called the active segment, and moving on to a new segment is called a roll.

The base offset is the offset of the first record in a segment, and a segment's file name is its base offset. 00000000000000003830.log is the segment that holds records from offset 3830 on.

A record batch, as we saw in part 6, is a group of records the producer sends together, and the broker writes that batch into .log almost unchanged. What matters in this post are the first two fields of the batch header: the first 8 bytes are the batch's first offset (baseOffset), and the next 4 bytes are the batch's length. With just these two values you can skip to the next batch without decoding any records.

The page cache is where the operating system keeps the contents of disk files in memory. When a program calls write(), the data first goes into the page cache, and the OS writes it to disk later. Calling fsync (in Java, FileChannel.force()) waits until it actually reaches the disk. As part 6 showed, with default settings Kafka does not fsync on every write.

mmap (a memory-mapped file) is an OS feature that maps a file directly into a process's address space. In Java you use the MappedByteBuffer returned by FileChannel.map(): a getInt() on the buffer is a file read, and underneath it looks straight at the page cache. Kafka opens .index and .timeindex this way (section 2).

Retention and compaction are the two ways old data is cleared away. Compaction is not the same thing as part 6's compression.type (shrinking bytes with gzip or zstd). Retention (cleanup.policy=delete, the default) deletes whole segments that are too old or push the log over a size limit; compaction (cleanup.policy=compact) keeps only the last record of each key and removes the rest.

1. The objects and files that make up one log

Four layers: UnifiedLog → LocalLog → LogSegments → LogSegment

The middle column of the figure above is the stack of objects.

LogSegment's fields are exactly those four files.

    private final FileRecords log;
    private final LazyIndex<OffsetIndex> lazyOffsetIndex;
    private final LazyIndex<TimeIndex> lazyTimeIndex;
    private final TransactionIndex txnIndex;
    private final long baseOffset;
    private final int indexIntervalBytes;
    private final long rollJitterMs;
    private final Time time;

storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegment.java lines 79–86

LogSegments is a ConcurrentSkipListMap so that "which segment holds offset N?" can be answered quickly. floorSegment(offset) returns the segment with the largest base offset that is ≤ offset (LogSegments.java lines 220–222). For offset 5000 that is the segment with base offset 3830. It's the very first thing LocalLog.read() does on the read path (part 8, LocalLog.java line 481). Because the map is safe for concurrent access, readers can find a segment without taking UnifiedLog's lock. The active segment is simply the last entry in the map (activeSegment() is lastSegment().get(), lines 322–324).

File names are 20-digit base offsets

File names are made in LogFileUtils.

    public static String filenamePrefixFromOffset(long offset) {
        NumberFormat nf = NumberFormat.getInstance();
        nf.setMinimumIntegerDigits(20);
        nf.setMaximumFractionDigits(0);
        nf.setGroupingUsed(false);
        return nf.format(offset);
    }

storage/src/main/java/org/apache/kafka/storage/internals/log/LogFileUtils.java lines 102–108

The comment gives the reason in one line: "pad out the offset number with zeros so that ls sorts the files numerically". The largest long has 19 digits, so 20 digits fit any offset. Then .log, .index, .timeindex or .txnindex is appended, and sometimes .deleted, .cleaned or .swap on top of that (the constants on lines 32–58 of the same file). Going the other way, offsetFromFileName() runs Long.parseLong() on everything before the first . (lines 79–81). That is what lets a restarting broker rebuild its segments from nothing but the directory listing (section 6).

Example 1: 12,000 records in 1 MiB segments

To see this for real, I made the segments small. This example was run in a Docker container (Eclipse Temurin JDK 25) with Kafka 4.5.0-SNAPSHOT built as-is from the reference commit (ab1e816411). The example folder is ex/u07-segments. I started the broker with log.segment.bytes=1048576 and sent 12,000 records with 250-byte values from a tiny producer that sends ten records per batch.

// Written for this post (not part of Kafka's source)
// Sends records to a one-partition topic ten at a time (send 10, flush → one batch of 10 records).
        // ... (bootstrap.servers, serializers) ...
        props.put(ProducerConfig.LINGER_MS_CONFIG, "1000");   // hold records until flush() is called
        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "none");
        String value = "v".repeat(250);                       // fixed 250-byte value
        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            for (int i = 0; i < count; i++) {
                producer.send(new ProducerRecord<>(topic, 0, String.format("k-%06d", i), value));
                if (i % 10 == 9) producer.flush();             // every 10 records go out as one batch
            }
            producer.flush();
        }

[Shortened from the example ex/u07-segments/Fill.java]

=== files in the partition directory (size name)
      1528 00000000000000000000.index
   1045973 00000000000000000000.log
      2292 00000000000000000000.timeindex
      1528 00000000000000003830.index
   1045973 00000000000000003830.log
        56 00000000000000003830.snapshot
      2292 00000000000000003830.timeindex
      1528 00000000000000007660.index
   1045973 00000000000000007660.log
        56 00000000000000007660.snapshot
      2292 00000000000000007660.timeindex
  10485760 00000000000000011490.index
    139281 00000000000000011490.log
        56 00000000000000011490.snapshot
  10485756 00000000000000011490.timeindex
         8 leader-epoch-checkpoint
        43 partition.metadata
=== roll lines in server.log
Rolled new log segment at offset 3830 in 9 ms. (org.apache.kafka.storage.internals.log.LocalLog)
Rolled new log segment at offset 7660 in 3 ms. (org.apache.kafka.storage.internals.log.LocalLog)
Rolled new log segment at offset 11490 in 2 ms. (org.apache.kafka.storage.internals.log.LocalLog)

Times like in 9 ms differ from run to run. A batch's size can also vary by a byte or two because of the variable-length fields that hold timestamp deltas, so in another run 00000000000000007660.log was 1,045,982 bytes, and in yet another the first batch was 2,740 bytes, shifting every later position and index entry by 9 bytes (29 → 5471 and so on). The calculations below use the values of the run shown here. Four things to read here:

• There are four segments, named 0, 3830, 7660 and 11490. Each holds 3,830 records (383 batches).
• No .log exceeds 1 MiB (1,048,576). 1,045,973 is 383 batches × 2,731 bytes. Taking a 384th batch would have made it 1,048,704 bytes, so that batch moved to a new segment before being written. Section 2 looks at that decision.
• The active segment's (11490) .index is 10 MiB, while the rolled segments' .index files are 1,528 bytes. Index files start at their maximum size and are cut down to the real size when the segment rolls (section 2). That's why, in part 1's example, "three records, yet a 10 MB .index".
• .snapshot is the producer state snapshot taken at every roll (for idempotent producers' duplicate check, part 12), leader-epoch-checkpoint records leader epochs (a number that grows each time the leader changes) and the offset where each began (part 9), and partition.metadata holds the topic id. The first segment (0) has no .snapshot because snapshots are taken at roll time with the new segment's base offset (UnifiedLog.roll(), section 2).

Example 1's folder redrawn segment by segment. File sizes are exactly the output above; offset ranges follow from the file names (base offsets) and the 12,000 records.

Easy to trip over: log.segment.bytes can't go below 1 MiB

At first I thought of a value like log.segment.bytes=102400 to cut segments even finer, but config validation rejects it. Both the broker and the topic config are defined with atLeast(1024 * 1024).

            .define(ServerLogConfigs.LOG_SEGMENT_BYTES_CONFIG, INT, DEFAULT_SEGMENT_BYTES, atLeast(1024 * 1024), HIGH, ServerLogConfigs.LOG_SEGMENT_BYTES_DOC)

storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java line 160

The topic-level segment.bytes has the same floor (line 198). The default DEFAULT_SEGMENT_BYTES is 1024 * 1024 * 1024, i.e. 1 GiB (line 130). There is a separate internal setting, internal.segment.bytes, that test code uses for smaller segments (lines 152–153, documented as "testing only"), and segmentSize() uses it when set (lines 384–387). The example uses the official minimum of 1 MiB.

2. When does a segment move on — rolling

Five conditions

The roll decision sits inside UnifiedLog.append(), right after offsets are assigned. It's the maybeRoll() that part 6 only mentioned as "a new segment when this one is full". Holding lock, it asks the active segment shouldRoll(new RollParams(…)); if the answer is yes it calls roll() and returns the new segment, otherwise it returns the current one (UnifiedLog.java lines 2166–2202). RollParams is just a record carrying the six values the decision needs (RollParams.java lines 22–23). The decision itself is five lines of LogSegment.shouldRoll().

    public boolean shouldRoll(RollParams rollParams) throws IOException {
        boolean reachedRollMs = timeWaitedForRoll(rollParams.now(), rollParams.maxTimestampInMessages()) > rollParams.maxSegmentMs() - rollJitterMs;
        int size = size();
        return size > rollParams.maxSegmentBytes() - rollParams.messagesSize() ||
            (size > 0 && reachedRollMs) ||
            offsetIndex().isFull() || timeIndex().isFull() || !canConvertToRelativeOffset(rollParams.maxOffsetInMessages());
    }

storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegment.java lines 167–173

The roll decision is made before writing the batch, so it amounts to "if it looks like it will overflow, move on first".

One by one:

1. Size — current size > segment.bytes − this batch's size, i.e. "writing this batch would overflow". In example 1, 1,045,973 > 1,048,576 − 2,731 = 1,045,845 became true and the 384th batch went to a new segment. So a segment never exceeds segment.bytes. A batch that is itself bigger than segment.bytes couldn't fit even into an empty segment, so append() rejects it with RecordBatchTooLargeException before the roll check (UnifiedLog.java lines 1242–1245). Exempting follower replication writes (AppendOrigin.REPLICATION) from that check is a condition 4.4.0 did not have.
2. Time — segment.ms (on the broker the default is log.roll.hours=168, 7 days — DEFAULT_SEGMENT_MS, LogConfig.java line 131). The key point is that the clock is the records' timestamps, not the wall clock. If the first batch has a timestamp, timeWaitedForRoll() returns this batch's max timestamp − the first batch's timestamp; only when there is none does it use now − segment creation time (lines 709–716). An empty segment never rolls on time (size > 0). rollJitterMs is a random number drawn from segment.jitter.ms (default 0) that spreads rolls out so that many partitions don't all roll at the same instant (LogConfig.randomSegmentJitter(), lines 399–404). On a compacted topic with max.compaction.lag.ms set, maxSegmentMs() returns the smaller of the two (lines 406–411, which ties into section 5).
3. The offset index is full — isFull() from section 3 below.
4. The time index is full — the time index must write one last entry at roll time, so it keeps one slot free and reports "full" one entry early (TimeIndex.isFull() is entries() >= maxEntries() - 1, TimeIndex.java lines 120–124).
5. Relative offset overflow — the indexes store offsets as the difference from the base offset (a 4-byte int, section 3). If the difference would exceed Integer.MAX_VALUE it can't be stored, so a new segment starts (AbstractIndex.toRelative(), lines 552–558). Putting 2.1 billion records into one segment is rare, but it can happen when replicating a log that compaction has left with huge gaps between offsets (the comment in maybeRoll(), lines 2183–2192).

Easy to trip over: if nobody writes, segment.ms passing doesn't roll anything

shouldRoll() is only called from inside append(). So if you write to a topic with segment.ms=5000 and wait a minute, the active segment stays as it is until the next record arrives. That's why the compaction example in section 5 waits six seconds and then writes one more record. (The exception: when the retention check decides "even the active segment has to go", it rolls from there — section 4.)

What roll() does

    public LogSegment roll(Optional<Long> expectedNextOffset) throws IOException {
        synchronized (lock) {
            long nextOffset = expectedNextOffset.orElse(0L);

            LogSegment newSegment = localLog.roll(nextOffset);
            // ... (comment: align the producer state snapshot with the new segment's base offset) ...
            producerStateManager.updateMapEndOffset(newSegment.baseOffset());
            // We avoid potentially-costly fsync call, since we acquire UnifiedLog#lock here
            // which could block subsequent produces in the meantime.
            // flush is done in the scheduler thread along with segment flushing below
            Optional<File> maybeSnapshot = producerStateManager.takeSnapshot(false);
            updateHighWatermarkWithLogEndOffset();
            // Schedule an asynchronous flush of the old segment
            scheduler().scheduleOnce("flush-log", () -> {
                maybeSnapshot.ifPresent(f -> flushProducerStateSnapshot(f.toPath()));
                flushUptoOffsetExclusive(newSegment.baseOffset());
            });
            return newSegment;
        }
    }

Shortened from storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java lines 2221–2244

This is where part 6's "the other flush points are in part 7" leads. A roll does fsync the old segment, but not while holding lock — it hands that to a kafka-scheduler thread. The comment says why: fsync can be slow, and while lock is held the next Produce for the same partition has to wait. flushUptoOffsetExclusive() calls flush() (= FileChannel.force(true) on .log plus MappedByteBuffer.force() on the indexes, LogSegment.java lines 628–636) on the segments from the recovery point up to the new segment, and when it's done LocalLog.markFlushed() raises the recovery point to the new segment's base offset (lines 2274–2291). Section 6 shows what the recovery point is for.

The other periodic flushes are the kafka-log-flusher task that LogManager starts (log.flush.scheduler.interval.ms) and the topic's flush.ms, but both default to Long.MAX_VALUE (ServerLogConfigs.java line 109, LogConfig.java line 207), so with default settings they effectively never run. Think of the default fsync as happening at roll time and at clean shutdown.

localLog.roll() handles the files (LocalLog.java lines 634–699). The new base offset is max(expected offset, LEO); it calls onBecomeInactiveSegment() on the old active segment and then opens four new files with LogSegment.open(). onBecomeInactiveSegment() is four lines:

    public void onBecomeInactiveSegment() throws IOException {
        timeIndex().maybeAppend(maxTimestampSoFar(), shallowOffsetOfMaxTimestampSoFar(), true);
        offsetIndex().trimToValidSize();
        timeIndex().trimToValidSize();
        log.trim();
    }

storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegment.java lines 681–686

It writes "the largest timestamp in this segment" as the last time index entry (section 4 uses this value to measure retention), and shrinks both indexes from their 10 MiB to the number of entries actually used. The 1,528 bytes (= 191 entries × 8 bytes) and 2,292 bytes (= 191 entries × 12 bytes) in example 1 are the result. The same is done to the active segment on clean shutdown (UnifiedLog.prepareActiveSegmentForClose(), lines 952–959). At the end of example 1, after the shutdown, the active segment's indexes look like this:

=== after a clean shutdown: the active segment's index files are trimmed
       200 00000000000000011490.index
    139281 00000000000000011490.log
        56 00000000000000011490.snapshot
       300 00000000000000011490.timeindex

25 entries × 8 = 200, 25 entries × 12 = 300. On the next startup LogLoader grows the active segment's indexes back to the maximum size (resizeIndexes(config.maxIndexSize), LogLoader.java line 212).

3. From offset to file position — the sparse index

One index entry per 4 KB

Here is the second half of LogSegment.append(), which part 6 cut short.

    public void append(long largestOffset,
                       MemoryRecords records) throws IOException {
        if (records.sizeInBytes() > 0) {
            // ... (trace log) ...
            int physicalPosition = log.sizeInBytes();

            ensureOffsetInRange(largestOffset);

            // append the messages
            long appendedBytes = log.append(records);
            // ... (trace log) ...

            for (RecordBatch batch : records.batches()) {
                long batchMaxTimestamp = batch.maxTimestamp();
                long batchLastOffset = batch.lastOffset();
                if (batchMaxTimestamp > maxTimestampSoFar()) {
                    maxTimestampAndOffsetSoFar = new TimestampOffset(batchMaxTimestamp, batchLastOffset);
                }

                if (bytesSinceLastIndexEntry > indexIntervalBytes) {
                    offsetIndex().append(batchLastOffset, physicalPosition);
                    timeIndex().maybeAppend(maxTimestampSoFar(), shallowOffsetOfMaxTimestampSoFar());
                    bytesSinceLastIndexEntry = 0;
                }
                var sizeInBytes = batch.sizeInBytes();
                physicalPosition += sizeInBytes;
                bytesSinceLastIndexEntry += sizeInBytes;
            }
        }
    }

Shortened from storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegment.java lines 250–280

After writing to the file it walks the batches and writes an entry only when more than indexIntervalBytes has piled up since the last index entry. indexIntervalBytes is the topic setting index.interval.bytes (broker: log.index.interval.bytes), default 4096 (ServerLogConfigs.java line 97). So the offset index doesn't list every record — not even every batch. An index like this is called a sparse index. An entry is (this batch's last offset, the file position where this batch starts).

Example 1's dump is exactly that calculation.

=== first batches of the first segment (baseOffset lastOffset position size)
baseOffset 0 lastOffset 9 position 0 size 2731
baseOffset 10 lastOffset 19 position 2731 size 2731
baseOffset 20 lastOffset 29 position 5462 size 2731
baseOffset 30 lastOffset 39 position 8193 size 2731
baseOffset 40 lastOffset 49 position 10924 size 2731
baseOffset 50 lastOffset 59 position 13655 size 2731
=== first entries of the offset index (.index)
Dumping /tmp/kafka/1/data/seg-0/00000000000000000000.index
offset: 29 position: 5462
offset: 49 position: 10924
offset: 69 position: 16386
offset: 89 position: 21848
=== number of index entries per segment
00000000000000000000.index: 191 offset entries, 191 time entries
00000000000000003830.index: 191 offset entries, 191 time entries
00000000000000007660.index: 191 offset entries, 191 time entries
00000000000000011490.index: 25 offset entries, 25 time entries

Only the needed columns of the kafka-dump-log.sh output are kept. Before the first batch bytesSinceLastIndexEntry is 0, before the second 2,731, before the third 5,462. At the third batch (offsets 20–29, position 5462), 5,462 > 4,096 becomes true, the first entry 29 → 5462 is written, and the counter goes back to 0. After that there is one entry every two batches, so a segment of 383 batches has 191 entries: 3,830 records, 191 index entries, 1.5 KB.

Easy to trip over: the key is the batch's last offset, the value is the batch's first position

The 29 in 29 → 5462 is the last offset of the batch holding 20–29, but 5462 is that batch's first position. At first glance they don't seem to match, but all the index promises is "to find offset 29 or above, scan from position 5462", and since that batch's first offset (20) is below 29, the promise always holds. 4.4.0's LogSegment.append() has the same shape.

Lookup: one binary search, then a scan over batch headers

Turning an offset into a position is the two lines of LogSegment.translateOffset().

    LogOffsetPosition translateOffset(long offset, int startingFilePosition) throws IOException {
        OffsetPosition mapping = offsetIndex().lookup(offset);
        return log.searchForOffsetFromPosition(offset, Math.max(mapping.position(), startingFilePosition));
    }

storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegment.java lines 398–401

1. OffsetIndex.lookup(offset) binary-searches the index for the largest entry ≤ the target offset. If there is none it returns (base offset, 0), i.e. the start of the file (OffsetIndex.java lines 97–106).
2. FileRecords.searchForOffsetFromPosition() steps through batches from that position until it finds the one containing the target offset (FileRecords.java lines 395–425). To step over a batch it reads only the front of the header (first offset, length, magic byte — HEADER_SIZE_UP_TO_MAGIC) and skips by the length (FileLogInputStream.java lines 64–94). It never decodes records.

The sparser the index, the more batches step ③ scans. With default settings that's a little over 4 KB at most.

The end of example 1: looking it up with Kafka's own classes

I followed these two steps by calling Kafka's internal classes (OffsetIndex, FileRecords) directly. After stopping the broker, the same segment files are opened read-only.

// Written for this post (not part of Kafka's source)
// Follows the two steps of LogSegment.translateOffset() using Kafka's internal classes.
        // ... (usage comment, imports, start of main) ...
        // file names come from LogFileUtils — the base offset zero-padded to 20 digits
        File indexFile = LogFileUtils.offsetIndexFile(dir, baseOffset);
        File logFile = LogFileUtils.logFile(dir, baseOffset);
        // ...
        // open read-only (writable=false); maxIndexSize is not used for reading
        OffsetIndex index = new OffsetIndex(indexFile, baseOffset, -1, false);
        System.out.println("index entries: " + index.entries());
        try (FileRecords records = FileRecords.open(logFile, false)) {
            for (int i = 2; i < args.length; i++) {
                long target = Long.parseLong(args[i]);
                OffsetPosition hint = index.lookup(target);
                FileRecords.LogOffsetPosition found = records.searchForOffsetFromPosition(target, hint.position());
                // ... (print target, hint and found on one line) ...
            }
        }

[Shortened from the example ex/u07-segments/Lookup.java]

=== offset -> file position, the way LogSegment.translateOffset() does it
index file: 00000000000000000000.index, log file: 00000000000000000000.log
index entries: 191
target 0 -> index says (offset 0, position 0) -> batch at position 0 (baseOffset 0, size 2731)
target 35 -> index says (offset 29, position 5462) -> batch at position 8193 (baseOffset 30, size 2731)
target 49 -> index says (offset 49, position 10924) -> batch at position 10924 (baseOffset 40, size 2731)
target 50 -> index says (offset 49, position 10924) -> batch at position 13655 (baseOffset 50, size 2731)
target 3829 -> index says (offset 3829, position 1043242) -> batch at position 1043242 (baseOffset 3820, size 2731)

• 0 is smaller than the index's first entry (29), so the answer is "the start of the file".
• 35 starts from 29 → 5462, exactly as in the figure, skips one batch and lands at 8193.
• 49 matches an entry exactly, so that position (10924, the batch holding 40–49) is the answer right away. 50 starts from the same entry and skips one batch.
• 3829 (the last offset in this segment) also reaches the last entry in one binary search.

Indexes in mmap, searched from the "warm" end first

Index files are opened in AbstractIndex.

        try {
            /* pre-allocate the file if necessary */
            if (newlyCreated) {
                if (maxIndexSize < entrySize())
                    throw new IllegalArgumentException("Invalid max index size: " + maxIndexSize);
                raf.setLength(roundDownToExactMultiple(maxIndexSize, entrySize()));
            }

            long length = raf.length();
            MappedByteBuffer mmap = createMappedBuffer(raf, newlyCreated, length, writable, entrySize());

storage/src/main/java/org/apache/kafka/storage/internals/log/AbstractIndex.java lines 108–117

A new index preallocates its file length to segment.index.bytes (broker: log.index.size.max.bytes, default 10 MiB — ServerLogConfigs.java line 93) rounded down to a multiple of the entry size, and FileChannel.map()s the whole thing (lines 471–487). Writing an entry is two mmap.putInt() calls (OffsetIndex.append(), lines 150–151), and reading is getInt(). An offset index entry is 8 bytes (relative offset 4 + position 4, OffsetIndex.java line 56); a time index entry is 12 bytes (timestamp 8 + relative offset 4). With the defaults, the offset index holds 10,485,760 / 8 = 1,310,720 entries. A 1 GiB segment with one entry per 4 KB needs at most a little over 260 thousand, so with default settings the index rarely fills up first.

Writing offsets as 4-byte relative offsets (the difference from the base offset) is another space saver. OffsetIndex's class comment (lines 44–47) explains that with base offset 50, offset 55 is stored as 5.

The binary search has one small trick, and the long comment in AbstractIndex (lines 327–383) explains why. In short: the index only grows at the end, and lookups (from in-sync followers and consumers that aren't lagging) also go almost to the end. But a plain binary search touches page 0, the middle page and so on every time, and when the index grows by a page the set of touched pages changes, so long-unused pages have to be read back from disk (page faults) — the comment says latency jumped to around a second. So it looks at the last 8 KB (the "warm" section) first.

        int firstHotEntry = Math.max(0, entries - 1 - warmEntries());
        // check if the target offset is in the warm section of the index
        if (compareIndexEntry(parseEntry(idx, firstHotEntry), target, searchEntity) < 0) {
            return binarySearch(idx, target, searchEntity,
                searchResultType, firstHotEntry, entries - 1);
        }

        // check if the target offset is smaller than the least offset
        if (compareIndexEntry(parseEntry(idx, 0), target, searchEntity) > 0) {
            return switch (searchResultType) {
                case LARGEST_LOWER_BOUND -> -1;
                case SMALLEST_UPPER_BOUND -> 0;
            };
        }

        return binarySearch(idx, target, searchEntity, searchResultType, 0, firstHotEntry);

storage/src/main/java/org/apache/kafka/storage/internals/log/AbstractIndex.java lines 498–513

warmEntries() is 8192 / entrySize(), which is 1,024 entries for the offset index (lines 384–386). If the target is within the last 1,024 entries, the search stays inside them, so it always touches the same two or three pages, and those pages stay in the page cache. "Leave it to the page cache" shows up in many places in this post as the reason Kafka is fast despite using disks; this code goes a step further and shapes its access pattern so that the page cache works well.

LazyIndex: not opening every index at startup

The reason LogSegment's index fields are LazyIndex<OffsetIndex> rather than OffsetIndex is in the class comment (lines 29–41). When a broker with a very large number of segments starts or shuts down, mmap-ing every index would take a long time, so it holds only the file name and opens the index on the first get(). get() returns the index right away if it's already open; otherwise it takes a lock, checks once more and creates the OffsetIndex or TimeIndex (LazyIndex.java lines 160–179). Old segments don't open their indexes until they are read or deleted.

The time index: time → offset

.timeindex serves offsetsForTimes() (a consumer's "read from this time", kafka-get-offsets.sh --time). An entry (T, O) means "the largest timestamp seen up to offset O is T", and entries are appended only when the timestamp grows (TimeIndex.maybeAppend(), line 203). So the entries' timestamps always increase and can be binary-searched. A lookup gets an offset from the time index, turns that offset into a position with the offset index, and scans the file (LogSegment.findOffsetByTimestamp(), lines 752–759). The first time index entries from example 1:

=== first entries of the time index (.timeindex)
Dumping /tmp/kafka/1/data/seg-0/00000000000000000000.timeindex
timestamp: 1794024676890 offset: 29
timestamp: 1794024676913 offset: 49
timestamp: 1794024676935 offset: 69

The timestamps differ from run to run (this run was on 2026-11-07). They are written at the same moment as the offset index entries (bytesSinceLastIndexEntry > indexIntervalBytes), so the offset column matches.

Example: shrink the index and the index triggers the roll

In the same example I sent 2,000 records to a topic seg2 with segment.index.bytes=400.

=== a second topic seg2 with segment.index.bytes=400 (index files hold only a few entries)
Created topic seg2.
sent 2000 records to seg2
00000000000000000000.log 00000000000000000650.log 00000000000000001300.log 00000000000000001950.log 
00000000000000000000.index: 32 offset entries, 32 time entries
00000000000000000650.index: 32 offset entries, 32 time entries
00000000000000001300.index: 32 offset entries, 32 time entries
00000000000000001950.index: 2 offset entries, 2 time entries

The segments are only 177 KB, yet it rolled every 650 records. With 400 bytes, the offset index holds 50 entries and the time index holds 33 (400 rounded down to a multiple of 12 is 396 bytes). The time index keeps one slot free, so it is "full" at 32 entries, and the batch after the 32nd entry (offset 649) caused a roll. The time index filled up before the offset index. We've just seen the third and fourth conditions of shouldRoll() actually fire.

4. Deleting old segments — retention

Who and when: the kafka-log-retention task

The retention check is not done by request handler threads but by periodic tasks that LogManager schedules at startup. After loading the logs, startupWithConfigOverrides() schedules cleanupLogs() under the name kafka-log-retention to run every retentionCheckMs, followed by the other tasks in the table below (LogManager.java lines 811–834).

These tasks run in KafkaScheduler's thread pool, whose threads are named kafka-scheduler-N (KafkaScheduler.java line 95). The broker creates the pool with background.threads (default 10, ServerConfigs.java line 64) threads (BrokerServer.scala line 216). The roll's flush-log and the delete-file task below run in the same pool.

The defaults are on lines 85, 109, 116 and 120 of ServerLogConfigs.java. Every task first runs after log.initial.task.delay.ms (default 30 s, line 172 — documented as "testing only").

cleanupLogs() calls deleteOldSegments() only on logs whose cleanup.policy does not contain compact (lines 1703–1742). For logs with compact (including compact,delete), the cleaner thread from section 5 does the retention check instead (LogCleaner.java lines 566–579).

deleteOldSegments(): three conditions in turn

    public int deleteOldSegments() throws IOException {
        int deletedSegments;
        try {
            if (config().delete) {
                deletedSegments = deleteLogStartOffsetBreachedSegments() +
                        deleteRetentionSizeBreachedSegments() +
                        deleteRetentionMsBreachedSegments();
            } else if (config().compact) {
                deletedSegments = deleteLogStartOffsetBreachedSegments();
            } else {
                // ... (empty cleanup.policy — only the start offset, unless tiered storage) ...
            }
        } finally {
            // ... (update the retention-size-in-percent metric) ...
        }
        return deletedSegments;
    }

Shortened from storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java lines 1981–2010

With the delete policy it checks three things in this order.

Here is the retention.ms check.

        DeletionCondition shouldDelete = (segment, nextSegmentOpt) -> {
            long anchorTimestamp = segment.largestTimestamp();
            if (startMs < anchorTimestamp) {
                if (remoteLogEnabledAndRemoteCopyEnabled) {
                    anchorTimestamp = segment.lastModified();
                    futureTimestampLogger.warn("{} contains future timestamp(s), using lastModified time {} as the retention anchor", segment, anchorTimestamp);
                } else {
                    futureTimestampLogger.warn("{} contains future timestamp(s), making it ineligible to be deleted", segment);
                }
            }
            boolean delete = startMs - anchorTimestamp > retentionMs;
            // ... (debug log) ...
            return delete;
        };

Shortened from storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java lines 2022–2036

The reference point is segment.largestTimestamp(), i.e. the largest timestamp of the records in the segment (the file's modification time only when records have no timestamps, LogSegment.java lines 856–861). Not the time the file was created. So a segment is deleted only once its newest record is older than retention.ms. If a producer sent records stamped with a future time, startMs < anchorTimestamp holds and that segment is never deleted — only a warning is logged. The branch that measures from the file's modification time instead when tiered storage is on is not in 4.4.0; it came into trunk with commit f52e331cae (KAFKA-20609). Without tiered storage it behaves the same as 4.4.0.

How far: from the front, and only below the high watermark

The three methods differ only in their condition (DeletionCondition); the choosing is done by deletableSegments().

            while (segmentOpt.isPresent()) {
                LogSegment segment = segmentOpt.get();
                Optional<LogSegment> nextSegmentOpt = nextOption(segmentsIterator);
                boolean isLastSegmentAndEmpty = nextSegmentOpt.isEmpty() && segment.size() == 0;
                long upperBoundOffset = nextSegmentOpt.map(LogSegment::baseOffset).orElseGet(this::logEndOffset);
                // We don't delete segments with offsets at or beyond the high watermark to ensure that the log start
                // offset can never exceed it.
                boolean predicateResult = highWatermark() >= upperBoundOffset && predicate.execute(segment, nextSegmentOpt);

                // ... (tiered storage: roll the active segment ahead of time) ...
                if (predicateResult && !isLastSegmentAndEmpty && isSegmentEligibleForDeletion(nextSegmentOpt, upperBoundOffset)) {
                    deletable.add(segment);
                    segmentOpt = nextSegmentOpt;
                } else {
                    segmentOpt = Optional.empty();
                }
            }

Shortened from storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java lines 1901–1921

• It starts at the oldest segment and stops at the first one where the condition is false. It never plucks out a single segment from the middle; a log only ever shrinks from the front.
• The whole segment must lie below the high watermark (the boundary all ISR members have received, part 9). As the comment says, this keeps the log start offset from overtaking the high watermark.
• An empty last segment is never picked. But a non-empty active segment can be. In that case deleteSegments() notices that every segment would go and rolls an empty segment first ("we must always have at least one segment", lines 1947–1956). A log must always have at least one segment.

Deleting in two steps: .deleted, then file.delete.delay.ms

deleteSegments() first raises the log start offset to the base offset of the first surviving segment (lines 1960–1965), then hands the chosen segments to LocalLog.removeAndDeleteSegments(). That removes them from the LogSegments map (line 404) — but the files are not deleted right away.

        for (LogSegment segment : segmentsToDelete) {
            if (!segment.hasSuffix(DELETED_FILE_SUFFIX)) {
                segment.changeFileSuffixes("", DELETED_FILE_SUFFIX);
            }
        }

        Runnable deleteSegments = () -> {
            // ... (log line, exception handling) ...
                        for (LogSegment segment : segmentsToDelete) {
                            segment.deleteIfExists();
                        }
            // ...
        };
        if (asyncDelete) {
            scheduler.scheduleOnce("delete-file", deleteSegments, config.fileDeleteDelayMs);
        } else {
            deleteSegments.run();
        }

Shortened from storage/src/main/java/org/apache/kafka/storage/internals/log/LocalLog.java lines 904–926

It appends .deleted to the names of the four files and leaves the actual deletion to a delete-file task after file.delete.delay.ms (default 60 s, ServerLogConfigs.java line 105). The reason is in the comment on removeAndDeleteSegments() (lines 385–386). The moment a segment leaves the map, new reads can't find it, but a request that was already reading it may still be running (for example a Fetch from part 8 that is sending the file to a socket with zero-copy). Waiting a while before deleting makes this safe without synchronizing with that request. If the broker dies in between, LogLoader deletes the leftover .deleted files on the next startup (LogLoader.java lines 279–281).

Once steps 5–7 are done the segment is gone from the log; steps 8–10 are disk housekeeping.

Example 2: a topic with retention.ms=20 s

The example folder is ex/u07-retention. I started the broker so the retention check runs every 2 seconds (log.retention.check.interval.ms=2000) and first runs after 2 seconds (log.initial.task.delay.ms=2000, a testing-only setting), sent the same 12,000 records as example 1 to a topic with retention.ms=20000, segment.bytes=1048576 and file.delete.delay.ms=10000, and printed the list of .log files whenever it changed.

=== produce 12000 records
sent 12000 records to ret
earliest offset: ret:0:0
=== watch the .log files (t = seconds after producing)
t=3s: 00000000000000000000.log 00000000000000003830.log 00000000000000007660.log 00000000000000011490.log 
t=17s: 00000000000000000000.log.deleted 00000000000000003830.log 00000000000000007660.log 00000000000000011490.log 
t=20s: 00000000000000000000.log.deleted 00000000000000003830.log.deleted 00000000000000007660.log 00000000000000011490.log 
t=22s: 00000000000000000000.log.deleted 00000000000000003830.log.deleted 00000000000000007660.log.deleted 00000000000000011490.log.deleted 00000000000000012000.log 
t=24s: 00000000000000003830.log.deleted 00000000000000007660.log.deleted 00000000000000011490.log.deleted 00000000000000012000.log 
t=28s: 00000000000000007660.log.deleted 00000000000000011490.log.deleted 00000000000000012000.log 
t=30s: 00000000000000012000.log 
earliest offset: ret:0:12000

t is seconds since the producer finished, and the values shift a little from run to run. Times like in 8 ms and the lastModifiedTime/largestRecordTimestamp values in the log lines below also differ from run to run. Here are just this partition's lines from server.log (with the leading time, class name and partition prefix removed):

=== server.log lines about ret-0 (trimmed)
Rolled new log segment at offset 3830 in 8 ms.
Rolled new log segment at offset 7660 in 3 ms.
Rolled new log segment at offset 11490 in 2 ms.
Incremented log start offset to 3830 due to segment deletion
Deleting segment LogSegment(baseOffset=0, size=1045973, lastModifiedTime=1794024815819, largestRecordTimestamp=1794024815939) due to log retention time 20000ms breach based on the largest record timestamp in the segment
Incremented log start offset to 7660 due to segment deletion
Deleting segment LogSegment(baseOffset=3830, size=1045973, lastModifiedTime=1794024818444, largestRecordTimestamp=1794024818462) due to log retention time 20000ms breach based on the largest record timestamp in the segment
Rolled new log segment at offset 12000 in 2 ms.
Incremented log start offset to 12000 due to segment deletion
Deleting segment LogSegment(baseOffset=7660, size=1045973, lastModifiedTime=1794024820273, largestRecordTimestamp=1794024820278) due to log retention time 20000ms breach based on the largest record timestamp in the segment
Deleting segment LogSegment(baseOffset=11490, size=139281, lastModifiedTime=1794024820474, largestRecordTimestamp=1794024820477) due to log retention time 20000ms breach based on the largest record timestamp in the segment
Deleting segment files LogSegment(baseOffset=0, size=1045973, lastModifiedTime=1794024815819, largestRecordTimestamp=1794024815939)
Deleted log 00000000000000000000.log.deleted.
Deleting segment files LogSegment(baseOffset=3830, size=1045973, lastModifiedTime=1794024818444, largestRecordTimestamp=1794024818462)
Deleted log 00000000000000003830.log.deleted.
Deleting segment files LogSegment(baseOffset=7660, size=1045973, lastModifiedTime=1794024820273, largestRecordTimestamp=1794024820278), LogSegment(baseOffset=11490, size=139281, lastModifiedTime=1794024820474, largestRecordTimestamp=1794024820477)
Deleted log 00000000000000007660.log.deleted.
Deleted log 00000000000000011490.log.deleted.

• Each segment was deleted at a different time. Sending 12,000 records took the producer a few seconds, so each segment has a different largest timestamp (largestRecordTimestamp). Segment 0's last record crossed 20 seconds first. The log line's "based on the largest record timestamp" is exactly the reference point in the code above.
• "Incremented log start offset" is printed before "Deleting segment". deleteSegments() raises the start offset first, and the reason for deleting (reason.logReason()) is printed inside removeAndDeleteSegments() (LocalLog.java line 403). The change of kafka-get-offsets.sh --time -2 (earliest offset) from 0 to 12000 is that start offset.
• The active segment (11490) was deleted too. Right before it there's Rolled new log segment at offset 12000 — the "roll an empty segment first when everything would go" we just saw. That's why 00000000000000012000.log remains even with no records, and the next record will get offset 12000.
• About 10 seconds after .deleted was added (file.delete.delay.ms=10000), Deleting segment files and Deleted log appeared and the files were gone.

Easy to trip over: it's not "deleted as soon as retention passes"

Between a record crossing retention.ms and it actually disappearing from disk there are three layers of delay. ① The newest record in the segment holding it must cross the limit (segments are 1 GiB / 7 days by default); ② the next retention check (every 5 minutes by default) must come; ③ after the rename to .deleted, another file.delete.delay.ms (60 s by default) must pass. Consumers stop seeing it at ②, when the log start offset moves; disk space comes back at ③.

5. Only the last value per key — compaction

The cleaner thread and the "dirty" section

Topics with cleanup.policy=compact are handled by LogCleaner. log.cleaner.threads (default 1, CleanerConfig.java line 38) CleanerThreads each run under the name kafka-log-cleaner-thread-<number> (LogCleaner.java lines 205–209 and 468), and every round does this:

        @Override
        public void doWork() {
            boolean cleaned = tryCleanFilthiestLog();
            if (!cleaned) {
                try {
                    pause(config.backoffMs, TimeUnit.MILLISECONDS);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }

            cleanerManager.maintainUncleanablePartitions();
        }

storage/src/main/java/org/apache/kafka/storage/internals/log/LogCleaner.java lines 503–515

"Pick the filthiest log and clean it; if there's nothing to clean, rest for log.cleaner.backoff.ms (default 15 s, CleanerConfig.java line 43)." A log is split into a front part that has already been cleaned (the clean section) and a back part that hasn't (the dirty section), and the boundary (firstDirtyOffset) is written to the cleaner-offset-checkpoint file in the data directory (LogCleanerManager.java line 66). "Filthiness" is cleanable dirty bytes ÷ (clean bytes + cleanable bytes) (LogToClean.java lines 53–54), and among the logs where it exceeds min.cleanable.dirty.ratio (default 0.5, LogConfig.java line 137) the one with the highest ratio wins (grabFilthiestCompactedLog(), lines 239–297). A log whose max.compaction.lag.ms has passed, so it must be compacted now, is a candidate even with a low ratio (line 282).

The whole dirty section isn't cleaned. There are three boundaries it must not cross:

        long firstUncleanableDirtyOffset = Stream.of(
                        // we do not clean beyond the last stable offset
                        Optional.of(log.lastStableOffset()),

                        // the active segment is always uncleanable
                        Optional.of(log.activeSegment().baseOffset()),

                        // the first segment whose largest message timestamp is within a minimum time lag from now
                        minCompactionLagMs > 0 ? findFirstUncleanableSegment(log, firstDirtyOffset, now, minCompactionLagMs) : Optional.<Long>empty()
                )
                .flatMap(Optional::stream)
                .min(Long::compare)
                .orElseThrow(() -> new IllegalStateException("No uncleanable offset found"));

storage/src/main/java/org/apache/kafka/storage/internals/log/LogCleanerManager.java lines 730–742

The active segment is never cleaned — you can't swap out the file being written to. The other boundaries are the last stable offset (LSO, the point before any unfinished transaction, part 12) and min.compaction.lag.ms. So the most common reason for "compaction isn't happening" is that the active segment hasn't rolled (the same story as the "easy to trip over" in section 2).

One cleaning: build an offset map, then copy again

Cleaner.doClean() has two steps (Cleaner.java lines 159–209).

① Build the offset map (buildOffsetMap()). It reads the dirty section from the start and puts "key → the last offset where that key appeared" into a map.

                        try (CloseableIterator<Record> recordsIterator = batch.streamingIterator(decompressionBufferSupplier, maxRecordBodySize)) {
                            for (Record record : (Iterable<Record>) () -> recordsIterator) {
                                if (record.hasKey() && record.offset() >= startOffset) {
                                    if (map.size() < maxDesiredMapSize) {
                                        map.put(record.key(), record.offset());
                                    } else {
                                        return true;
                                    }
                                }
                                stats.indexMessagesRead(1);
                            }
                        }

storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java lines 815–826

When the same key shows up again, put() overwrites its value, so at the end each key holds its last offset. If the map fills up (maxDesiredMapSize) it stops there and this round cleans only up to that point.

The map is a SkimpyOffsetMap. As the name says it is a "skimpy" hash table: it doesn't store keys, only the key's MD5 hash (16 bytes) and the offset (8 bytes).

    public SkimpyOffsetMap(int memory, String hashAlgorithm) throws NoSuchAlgorithmException {
        this.bytes = ByteBuffer.allocate(memory);

        this.digest = MessageDigest.getInstance(hashAlgorithm);

        this.hashSize = digest.getDigestLength();
        this.bytesPerEntry = hashSize + 8;
        this.slots = memory / bytesPerEntry;

        this.hash1 = new byte[hashSize];
        this.hash2 = new byte[hashSize];
    }

storage/src/main/java/org/apache/kafka/storage/internals/log/SkimpyOffsetMap.java lines 77–88

One ByteBuffer is divided into 24-byte slots; on a collision it shifts the 4-byte window it reads from the hash by one byte to try another slot (13 tries for a 16-byte MD5), and after that moves one slot at a time to find a free one (positionOf(), lines 205–209). However long a key is, it takes 24 bytes, and since no objects are created there's no GC pressure either. Its memory is log.cleaner.dedupe.buffer.size (default 128 MiB, CleanerConfig.java line 40) divided by the number of cleaner threads (LogCleaner.java line 472). With defaults that's 128 MiB ÷ 24 = 5,592,405 slots, filled only up to a load factor of 0.9 (log.cleaner.io.buffer.load.factor, line 42), so one round handles a little over 5 million keys. The design accepts the risk that two different keys with the same MD5 would be treated as one, in exchange for space (class comment, lines 29–30).

② Copy again (cleanSegments()). The segments from the start of the log (including the clean section) up to the cleaning boundary are grouped by size (groupSegmentsBySize()), and for each group a new segment file (.cleaned) is created into which only the records to keep are copied. Whether to keep each record is decided here:

        boolean pastLatestOffset = record.offset() > map.latestOffset();
        if (pastLatestOffset) {
            return true;
        }

        if (record.hasKey()) {
            ByteBuffer key = record.key();
            long foundOffset = map.get(key);
            /* First,the message must have the latest offset for the key
             * then there are two cases in which we can retain a message:
             *   1) The message has value
             *   2) The message doesn't have value but it can't be deleted now.
             */
            boolean latestOffsetForKey = record.offset() >= foundOffset;
            boolean legacyRecord = batch.magic() < RecordBatch.MAGIC_VALUE_V2;

            boolean shouldRetainDeletes;
            if (!legacyRecord) {
                shouldRetainDeletes = batch.deleteHorizonMs().isEmpty() || currentTime < batch.deleteHorizonMs().getAsLong();
            } else {
                shouldRetainDeletes = retainDeletesForLegacyRecords;
            }

            boolean isRetainedValue = record.hasValue() || shouldRetainDeletes;
            return latestOffsetForKey && isRetainedValue;
        } else {
            stats.invalidMessage();
            return false;
        }

storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java lines 577–605

• Records beyond the range the map has seen are kept (outside this round).
• With a key, only records at or after that key's last offset are kept. This is where old values disappear.
• A record whose value is null is called a tombstone, meaning "delete this key". Even when it is the last value, a tombstone isn't removed right away; it goes in a cleaning after the delete deadline written into its batch (deleteHorizonMs) has passed. The deadline is written into the batch at the first cleaning by adding delete.retention.ms (default 1 day, LogConfig.java line 134) (the LogCleaner class comment, lines 72–77). This gives late consumers time to see "this key was deleted".
• Records without a key are dropped in a compacted topic.

When copying is done, log.replaceSegments() swaps the new segments in for the old ones (line 329). It renames .cleaned → .swap and puts them in the map, sends the old segments through section 4's two-step .deleted path, and finally drops .swap to give them their proper names (LocalLog.java lines 1079–1109). If the broker dies halfway, at startup it sees the .swap files and finishes the swap (LogLoader.java lines 131–193).

One difference from 4.4.0: when the cleaned output overflows the relative offset of a single new segment (the fifth condition from section 2; in this commit, also when the recompressed output would exceed the int size limit), 4.4.0 split the original segment, aborted this cleaning and redid it later. This commit's cleanSegments() closes the cleaned segment at the overflow point and keeps going in a new .cleaned, swapping several in at once (lines 289–312).

Putting together what we have seen so far as the life of one segment:

The parentheses next to a state name are the suffix its files carry in that state. Thanks to these suffixes, wherever the broker dies, LogLoader can delete the leftovers or finish a swap on the next startup just from the file names (LogLoader.java lines 261–314).

Example 3: write the same keys three times, and delete one

The example folder is ex/u07-compaction. I started the broker with the cleaner resting only 1 second (log.cleaner.backoff.ms=1000) and used the console producer to send ten key:value lines to a topic with cleanup.policy=compact, segment.ms=5000 and min.cleanable.dirty.ratio=0.01. The last line is a tombstone, its value turned into null by null.marker=NULL.

=== produce 3 keys x 3 updates, then a tombstone (null value) for user2
=== segment 0 before cleaning (offset key payload)
offset: 0 key: user1 payload: a1
offset: 1 key: user2 payload: b1
offset: 2 key: user3 payload: c1
offset: 3 key: user1 payload: a2
offset: 4 key: user2 payload: b2
offset: 5 key: user3 payload: c2
offset: 6 key: user1 payload: a3
offset: 7 key: user2 payload: b3
offset: 8 key: user3 payload: c3
offset: 9 key: user2 
=== wait 6s (> segment.ms), then one more record so that the active segment rolls
00000000000000000000.log
00000000000000000010.log
=== wait for the cleaner thread
Cleaner 0: Beginning cleaning of log cmp-0
Cleaner 0: Building offset map for log cmp-0 for 1 segments in offset range [0, 10).
Cleaner 0: Cleaning log cmp-0 (cleaning prior to Sat Nov 07 04:13:49 UTC 2026, discarding tombstones prior to upper bound deletion horizon Thu Jan 01 00:00:00 UTC 1970)...
Cleaner 0: Swapping in cleaned segment(s) [LogSegment(baseOffset=0, size=110, ...)] for segment(s) [LogSegment(baseOffset=0, size=199, ...)] in log Log(dir=/tmp/kafka/1/data/cmp-0, ...)
=== segment 0 after cleaning (offset key payload)
00000000000000000000.log
00000000000000000010.log
offset: 6 key: user1 payload: a3
offset: 8 key: user3 payload: c3
offset: 9 key: user2 
=== consume from the beginning
Offset:6	user1	a3
Offset:8	user3	c3
Offset:9	user2	null
Offset:10	user4	d1
=== cleaner checkpoint (topic partition firstDirtyOffset)
0
1
cmp 0 10

The last line of the cleaner log (log-cleaner.log) is long, so I shortened it with ...; the time differs from run to run.

• The console producer sent the ten records (offsets 0–9) as one batch into segment 0. After waiting six seconds, writing user4:d1 rolled on time and created segment 10, so segment 0 stopped being active and became eligible for cleaning.
• The offset map was built for the range [0, 10), i.e. up to the active segment (10). The boundary being the active segment's base offset is the firstUncleanableDirtyOffset above.
• After cleaning, what remains is 6 (user1=a3), 8 (user3=c3) and 9 (the user2 tombstone). The segment went from 199 to 110 bytes (in a rerun it was 208 bytes before cleaning — because of variable-length fields in the batch). The surviving records' offsets did not change. Offsets 0–5 and 7 remain as holes, and the consumer receives 6, 8, 9 and 10.
• user2's tombstone survived the first cleaning (it goes only after delete.retention.ms). user2's old values b1, b2 and b3 all had offsets below the tombstone, so they were all removed.
• cmp 0 10 was written to cleaner-offset-checkpoint; the next cleaning's dirty section starts at offset 10. The file's first two lines, 0 and 1, are the checkpoint file format's version number and entry count (OffsetCheckpointFile's class comment, OffsetCheckpointFile.java lines 41–50). Section 6's recovery-point-offset-checkpoint uses the same format.

![One compaction. Segment 0's dirty section 0, 10) holds 0 user1=a1, 1 user2=b1, 2 user3=c1, 3 user1=a2, 4 user2=b2, 5 user3=c2, 6 user1=a3, 7 user2=b3, 8 user3=c3 and 9 user2=null (a tombstone). Segment 10 is the active segment holding 10 user4=d1 and is never cleaned; firstUncleanableDirtyOffset is the active segment's base offset. ① buildOffsetMap() fills a SkimpyOffsetMap from the MD5 of each key to its last offset: user1→6, user2→9, user3→8. ② cleanSegments() keeps a record only if its offset is at least map(key). ③ After copying into .cleaned and swapping in through .swap as .log, 6 user1=a3, 8 user3=c3 and 9 user2=null remain. ④ cmp 0 10 is written to cleaner-offset-checkpoint, so next time the dirty section starts at 10. Drawn with the actual values from example 3. The condition in ② is simplified; tombstones and transactional records get the extra conditions from the code above.

Easy to trip over: compaction does not guarantee "only the last value is there"

Reading a compacted topic from the beginning can still show the same key several times, because the active segment and the not-yet-cleaned dirty section still hold old values (had we read before cleaning in example 3, user1 would have appeared three times). What compaction guarantees is that "at least the last value of each key survives", so readers of such topics (for example the group coordinator, which rebuilds group state by reading __consumer_offsets, part 11) are written to overwrite when they meet the same key again.

6. Coming back from a crash — the recovery point and LogLoader

The recovery point: "fsynced up to here"

As part 6 showed, Kafka doesn't fsync on every write. So if the broker process dies with kill -9 or the server loses power, the tail that lived only in the page cache can be lost or left half-written. That's why LocalLog keeps a recovery point — "everything before this offset has been written to disk". As section 2 showed, it rises when a roll's flush-log task finishes (markFlushed(), LocalLog.java lines 243–249), and the kafka-recovery-point-checkpoint task writes it to the recovery-point-offset-checkpoint file in the data directory (LogManager.checkpointLogRecoveryOffsets(), lines 1008–1014). On clean shutdown, each log's active segment is tidied and flushed (prepareActiveSegmentForClose() → flush(true), lines 881–883), the checkpoints are written, and finally a .kafka_cleanshutdown file is left in the data directory (lines 903–912).

At startup: re-read only the segments after the recovery point

At startup LogManager.loadLogs() first checks for .kafka_cleanshutdown; if it's there, it deletes it and remembers that the shutdown was clean (lines 630–635). As the comment says, it's deleted so that if the broker dies again while loading the logs, the next startup treats it as an unclean shutdown. Then for each partition directory LogLoader rebuilds the segments from the file list (loadSegmentFiles(), lines 357–391) and, after an unclean shutdown, recovers.

    RecoveryOffsets recoverLog() throws IOException {
        // If we have the clean shutdown marker, skip recovery.
        if (!hadCleanShutdown) {
            Collection<LogSegment> unflushed = segments.values(recoveryPointCheckpoint, Long.MAX_VALUE);
            // ... (progress counters) ...
            while (unflushedIter.hasNext() && !truncated) {
                LogSegment segment = unflushedIter.next();
                logger.info("Recovering unflushed segment {}. {} recovered for {}.", segment.baseOffset(), numFlushed / numUnflushed, topicPartition);
                int truncatedBytes;
                try {
                    truncatedBytes = recoverSegment(segment);
                } catch (InvalidOffsetException | IOException ioe) {
                    // ... (empty the segment) ...
                }
                if (truncatedBytes > 0) {
                    // we had an invalid message, delete all remaining log
                    // ... (delete every remaining segment) ...
                    truncated = true;
                    // ...
                } else {
                    numFlushed += 1;
                    // ...
                }
            }
        }

Shortened from storage/src/main/java/org/apache/kafka/storage/internals/log/LogLoader.java lines 456–492

Only the segments after the recovery point are re-read through recoverSegment() → LogSegment.recover(). recover() empties the indexes, validates every batch from the start (batch.ensureValid() checks the CRC) and rebuilds .index and .timeindex with the same rule as in section 3.

            for (RecordBatch batch : log.batches()) {
                batch.ensureValid();
                ensureOffsetInRange(batch.lastOffset());

                // The max timestamp is exposed at the batch level, so no need to iterate the records
                if (batch.maxTimestamp() > maxTimestampSoFar()) {
                    maxTimestampAndOffsetSoFar = new TimestampOffset(batch.maxTimestamp(), batch.lastOffset());
                }

                // Build offset index
                if (validBytes - lastIndexEntry > indexIntervalBytes) {
                    offsetIndex().append(batch.lastOffset(), validBytes);
                    timeIndex().maybeAppend(maxTimestampSoFar(), shallowOffsetOfMaxTimestampSoFar());
                    lastIndexEntry = validBytes;
                }
                validBytes += batch.sizeInBytes();

storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegment.java lines 490–505

When it hits a corrupt batch it catches the exception (lines 514–517) and truncates the file before it (validBytes, line 522). If any bytes were cut, recoverLog() deletes every segment after it. It's rolling a log with a broken tail back to "just before it broke". OffsetIndex's comment (line 42) notes that index files carry no checksum and are rebuilt if damaged — this loop is that rebuild.

The further forward the recovery point, the faster the startup. Hence the design of fsyncing the old segment at every roll to push the recovery point forward.

Example 4: kill -9 versus a clean shutdown

The example folder is ex/u07-recovery. To make the result the same on every run, I started the broker so the recovery point checkpoint is written every second (log.flush.offset.checkpoint.interval.ms=1000, first run also after 1 second — log.initial.task.delay.ms=1000), sent 12,000 records to a segment.bytes=1048576 topic and waited 3 seconds. Then I killed the broker with kill -9 and started it again, and then shut it down cleanly with SIGTERM and started it again.

=== create topic rec (segment.bytes=1048576) and produce 12000 records
Created topic rec.
sent 12000 records to rec
00000000000000000000.log 00000000000000003830.log 00000000000000007660.log 00000000000000011490.log 
=== kill -9 the broker
clean shutdown marker: 0
recovery-point-offset-checkpoint for rec-0: rec 0 11490
=== restart #1 (after kill -9)
Recovering 1 logs from /tmp/kafka/1/data since no clean shutdown file was found
Recovering unflushed segment 11490. 0 recovered for rec-0.
Completed load of Log(dir=/tmp/kafka/1/data/rec-0, topicId=kJRxw38gQUKneMDde2_5Ag, topic=rec, partition=0, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=12000) with 4 segments, local-log-start-offset 0 and log-end-offset 12000 in 167ms (1/1 completed in /tmp/kafka/1/data)
=== stop with SIGTERM (clean shutdown)
clean shutdown marker: .kafka_cleanshutdown
recovery-point-offset-checkpoint for rec-0: rec 0 12000
=== restart #2 (after clean shutdown)
Skipping recovery of 1 logs from /tmp/kafka/1/data since clean shutdown file was found
Completed load of Log(dir=/tmp/kafka/1/data/rec-0, topicId=kJRxw38gQUKneMDde2_5Ag, topic=rec, partition=0, highWatermark=0, lastStableOffset=0, logStartOffset=0, logEndOffset=12000) with 4 segments, local-log-start-offset 0 and log-end-offset 12000 in 143ms (1/1 completed in /tmp/kafka/1/data)
=== records are all there
rec:0:12000

The topic id and times (ms) differ from run to run. Only the relevant lines of server.log are shown.

• Right before kill -9 the recovery point was 11490, the base offset of the last rolled segment. The three rolls each fsynced the old segment and pushed the recovery point 3830 → 7660 → 11490, while the active segment (11490–11999) had not been fsynced yet.
• On restart there was no .kafka_cleanshutdown, so recovery ran — and only one line, Recovering unflushed segment 11490, appeared. Segments 0, 3830 and 7660 were not re-read. (0 recovered is the integer division numFlushed / numUnflushed, so it always prints 0 until the last segment.) Only the process died inside the container, so the page cache survived and all 12,000 records are there with nothing truncated. Had the server lost power, part of what lay after the recovery point could have been lost; that share is recovered from other brokers through replication (part 9).
• After the clean SIGTERM shutdown, .kafka_cleanshutdown appeared and the recovery point became 12000, the same as the LEO. The next startup said Skipping recovery and re-read no segments at all.

Easy to trip over: highWatermark=0 right after loading

Both times it prints Completed load of Log(… highWatermark=0 …), even after the clean shutdown. LogLoader restores only segments, the LEO and the recovery point; the high watermark is set afterwards by ReplicaManager from its own replication-offset-checkpoint file and the replication state. That path is in part 9. There's no need to read this line's highWatermark as data loss — the final kafka-get-offsets.sh returning 12000 proves it.

What this post leaves out

• Tiered storage (old segments in remote storage): the remoteLogEnabled() / localLogStartOffset branches in UnifiedLog, the Remote* classes in storage/, RemoteIndexCache.java — left out of the whole series
• The transaction index .txnindex: TransactionIndex.java, LogSegment.updateTxnIndex() (lines 348–357) — part 12
• Producer state snapshots .snapshot: ProducerStateManager.takeSnapshot() — part 12
• The leader epoch cache and truncation (truncateTo()): epoch/LeaderEpochFileCache.java, LogSegment.truncateTo() (lines 561–592) — part 9
• The rest of the read path (UnifiedLog.read() → LocalLog.read() → LogSegment.read(), FetchDataInfo, zero-copy): LocalLog.java lines 467–529 — part 8
• Splitting segments whose offsets overflow: LocalLog.splitOverflowedSegment() (LocalLog.java line 957), LogLoader.retryOnOffsetOverflow()
• Deleting topic/partition directories (the -delete suffix, the kafka-delete-logs task): LogManager.deleteLogs() (line 1413)
• Multiple log directories (JBOD) and moving partitions between them (-future directories, ReplicaAlterLogDirsThread) — left out of the whole series
• log.preallocate (preallocating segment files): LogConfig.initFileSize() (lines 413–418), FileRecords.openChannel()
• Cleaner throttling and statistics: log.cleaner.io.max.bytes.per.second (Throttler), CleanerStats, the kafka.log:type=LogCleaner metrics

Summary

What's next

In this post we saw how records are laid out in files, up to translateOffset() turning one offset into a file position. The next post is about the side that takes that position and actually reads. A Fetch request from a consumer or a follower goes through ReplicaManager.fetchMessages() to LogSegment.read(); when there isn't enough to read, it waits in purgatory for up to fetch.max.wait.ms; and the file slice it found goes straight from the page cache to the socket via FileChannel.transferTo() — zero-copy. The reason we "wait a moment before deleting" in section 4 shows up again there. It continues in part 8: The Read Path — Fetch Requests, Long Polling and Zero-Copy.

• Part 2: The Big Picture — Clients, Brokers, Controllers and Partitions
• Part 6: How the Broker Accepts Records — The Record Batch Format and the Produce Path
• Part 9: Replication — The ISR, the High Watermark and the Wait Behind acks=all — the high watermark, leader epochs and log truncation
• Part 12: Idempotence and Transactions — How Exactly-Once Is Built, and Wrapping Up the Series — .snapshot and .txnindex