The Log: Advanced

The intermediate lesson ended with a promise: a compacted topic keeps the latest value for every key. It said nothing about when. Right after a producer writes a new value for key u42, the partition holds the old value and the new one, and a reader that starts at offset 0 sees both. Removing the old one is the job of the log cleaner, a pool of threads inside the broker, and it runs on its own schedule under rules you set per topic.

The cleaner sees each partition as 3 regions. The active segment, where new records are appended, is never touched. Everything before it is either clean, meaning the cleaner has already passed over it and it holds at most 1 value per key, or dirty, meaning records have arrived since the last pass. The boundary is 1 offset per partition, saved in a file named cleaner-offset-checkpoint in each log directory, so it survives a restart.

Rewriting segments costs a full read and a full write of their bytes, so the cleaner runs on a partition only once its dirty region has grown enough to pay for that. The measure is the dirty ratio: dirty bytes divided by clean plus dirty bytes, with the active segment left out. Say the clean segment in the table holds 100 MB and the 2 dirty segments hold 60 MB and 40 MB. The ratio is 100 divided by 200, which is 0.5. The topic setting min.cleanable.dirty.ratio defaults to 0.5, so this partition qualifies by a hair. When a cleaner thread wakes, it looks across every compacted partition on the broker, takes the eligible one with the highest ratio, cleans it, then looks again; if nothing qualifies it sleeps for log.cleaner.backoff.ms, 15 seconds by default.

When a pass may happen is bounded by 2 topic settings. min.compaction.lag.ms, default 0, keeps records out of the dirty region until their timestamps are that old: a reader that needs every change from the last 10 minutes sets it to 600000 and knows nothing younger is deduplicated. The cleaner stops at the first segment whose largest timestamp is younger than the lag, so 1 record stamped a year ahead stalls its segment and every later one. max.compaction.lag.ms, with a default so large it never fires, forces the opposite. Once the oldest dirty record is older than it, the partition becomes eligible whatever its ratio, so a topic that receives a trickle of writes still compacts inside a known window. A partition with a ratio of 0.1 and a 2 day old dirty record is cleaned because of the max lag; a partition with a ratio of 0.9 whose dirty records are all 3 minutes old waits because of the min lag.

A pass has 2 phases. First the cleaner reads the dirty region into an offset map that holds, for each key, the highest offset at which that key appears. The map lives in memory sized by log.cleaner.dedupe.buffer.size, 128 MB by default; when the dirty region holds more keys than fit, the pass covers only the prefix that fits and the rest waits for the next pass. Second it reads the clean and dirty segments from the start, copies into a new file each record whose offset is the one in the map for its key, and swaps the new file in for the group it replaced. Offsets never change. A reader positioned at offset 1500 resumes at 1500 even if 1500 itself was removed: the fetch returns the first surviving record after it.

A tombstone survives the pass that first reads it. That pass writes a deletion horizon beside it, and a later pass, once delete.retention.ms has elapsed past that horizon, drops it.

Every record carries 1 timestamp: a 64-bit count of milliseconds since the epoch, stored in the record itself on disk. Whose clock it came from is the topic setting message.timestamp.type, which has 2 values. CreateTime, the default, keeps what the producer sent: the producer's own clock at the moment the record was built, unless the application set an explicit value such as the time a sensor took its reading. LogAppendTime discards that and stamps the record with the broker's clock as the record is appended to the log.

Beside each segment's offset index sits a second file ending in .timeindex, a list of (timestamp, offset) pairs. The broker appends an entry about every index.interval.bytes of data, 4096 bytes by default, and only if the record's timestamp is larger than every timestamp already in that segment's index. The entries therefore increase even on a CreateTime topic whose records do not.

On the command line, kafka-get-offsets.sh --bootstrap-server broker1:9092 --topic clicks --time 1762250405000 prints that offset for every partition of the topic, and a consumer gets the same answer from its offsetsForTimes call. In place of a time you can pass 3 words: earliest returns the first offset still in the log, latest the offset after the last record, and max-timestamp the offset of the record with the largest timestamp.

Time retention deletes a closed segment once it is old enough, and the broker decides age with the segment's largest timestamp: a segment is eligible when now minus its largest timestamp exceeds retention.ms, checked every log.retention.check.interval.ms, 5 minutes by default. Under CreateTime that largest timestamp is whatever the most forward-running producer clock claimed. One record stamped 3 years ahead raises its segment's largest timestamp by 3 years, and the segment stays on disk until the real clock catches up to that stamp plus the retention.

Deletion walks the segments from the front and stops at the first one that is not old enough. Segment 20000 is stamped 3 years in the future, so it is never old enough, and segment 30000 behind it, which on its own is 8 days old and past the 7 day retention, is kept as well. Every segment appended after the future record is kept with it. One record of a few hundred bytes has turned a 7 day topic into one that keeps everything, and the disk fills at the partition's full write rate until somebody notices. Size retention, retention.bytes, still applies, because it counts bytes rather than reading stamps.

The broker refuses records that claim a time too far from its own clock, and 2 topic settings draw the window: message.timestamp.after.max.ms, default 3600000, 1 hour, rejects a record stamped further ahead than that, and message.timestamp.before.max.ms, with a default so large it never fires, would reject one stamped further behind. A rejected record fails the write with InvalidTimestampException, so the producer sees it and the log stays clean. With the defaults, the worst a future stamp can do is hold a segment 1 hour longer. The damage in the table needs the after window widened, which teams do when applications stamp records with a scheduled time, such as a reminder due next month, and then forget that retention reads the same field. Both settings apply only under CreateTime; LogAppendTime makes the broker's clock the stamp, so there is nothing to check.

The before window defaults to open because replays are routine: a producer re-reads 3 days of events from a file and writes them with their original timestamps. Under CreateTime those stamps stay, and if the topic's retention is 24 hours, every segment the replay fills is already older than the retention the moment it closes. The next retention check, within 5 minutes, deletes it. A reader that is 10 minutes behind the replay finds the offsets it wanted gone and starts again from the earliest offset the log still has. The replay looked like it worked, and the data was in the log for less time than it took to write it. The fixes are to raise retention.ms for the replay's duration, or to run the replay into a topic set to LogAppendTime, or to keep the original time inside the value and let the record's stamp be the time of the replay.

The same field feeds 2 more mechanisms. Time-based rolling measures from a segment's first stamp, so a segment that opens with a future-stamped record never rolls on segment.ms while new stamps trail it; it rolls on segment.bytes or waits. The time index records only stamps larger than any before them in the segment, so after the future record at offset 20000 no entry is added, and a lookup for any time from 2 Mar 2027 to 2030 picks the segment holding 20000, scans to it, and returns 20000. A reader that asks for 'everything since yesterday' is handed more than 10 days of data in place of 1.

LogAppendTime would not repair this: it stamps only later appends, and segment 20000 keeps its 2030 stamp. Set message.timestamp.after.max.ms back to 1 hour, then run kafka-delete-records.sh --bootstrap-server broker1:9092 --offset-json-file delete.json, the file naming partition 0 at offset 30000. The log's first offset moves to 30000, and the next check deletes segment 20000, whose other records were already past retention, with segment 30000 behind it, so lookups read real stamps again. A compact-only topic refuses the command; there, or where current records share the bad segment, copy the records into a new LogAppendTime topic.

Enabling it adds a second retention to the topic. retention.ms and retention.bytes keep their meaning: how much the topic holds in total, local and remote together. local.retention.ms and local.retention.bytes say how much of that stays on the broker's disk; their default, -2, means the same as the total, which is the same as not tiering at all. A topic with retention.ms of 90 days and local.retention.ms of 1 day holds 1 day of segments on fast disk and all 90 days in the store. Reads inside the last day are served from local files; reads older than that wait for a fetch from the store.

Step 3 happens shortly after the roll, and it is the step everything waits on. The local copy is deleted at step 4 only if step 3 completed, so a store that is unreachable for an hour means an hour of segments piling up on the broker past local.retention.ms, which is the right failure: data is never dropped to meet a local target. The active segment is never copied, whatever its age, because it is still being written. Each remote segment is known to the broker by a record in the internal topic __remote_log_metadata, saying which segment lives where and which offsets it covers, so a broker that restarts rebuilds its view of the store from that topic rather than listing the store.

A fetch that asks for an offset older than the local log's first offset becomes a remote read. The broker looks up the segment in its metadata, finds the position inside the segment using the remote copy of the offset index, which it caches locally under remote.log.index.file.cache.total.size.bytes, downloads the needed part of the segment, and returns records from it. The first such read on a cold segment takes the store's latency, tens to hundreds of milliseconds, in place of a local disk read; remote.fetch.max.wait.ms caps how long the broker waits on the store before answering the fetch with what it has. A lookup by time picks the remote segment by the largest timestamps in __remote_log_metadata, then waits on the store for its time index and records. A reader replaying from 60 days ago runs at the store's throughput, and a reader 10 minutes behind never touches it.

Turning it off starts with remote.log.copy.disable=true, which stops new copies while remote reads keep working, and a topic can stay that way while its remote data is needed. remote.storage.enable=false then makes it local, and the broker accepts that only alongside remote.log.delete.on.disable=true, which deletes the remote segments, so a reader positioned in one must be moved first.

Follow a 3 MB record. The producer serializes it, compares 3145728 bytes with max.request.size, and raises RecordTooLargeException without sending anything. Raise max.request.size to 4 MiB and the producer sends it; the broker compares the batch with the topic's max.message.bytes if one is set, else its own message.max.bytes, and answers with the error MESSAGE_TOO_LARGE, which the producer surfaces as the same exception. Set max.message.bytes=4194304 on the topic as well and the record lands in the log. The consumer then needs nowhere near 3 MB of configuration to read it: since a fetch always returns at least the first batch of the first partition that has data, however large, the record arrives in a fetch of its own. What the consumer does need is memory for it, and a handler written for a payload 30 times its usual size.

The ceiling behind all of these is the broker's socket.request.max.bytes, 104857600, 100 MiB, the largest request of any kind it reads from a connection. A record that needs the limits raised toward that is a sign the record is the wrong unit, which is what the 2 patterns below are for.

The pattern's cost is 2 lifetimes to keep in step. The record expires under retention.ms; the object expires under whatever rule the store applies. If the object is deleted first, a reader replaying the topic finds a pointer to nothing, so the store's retention must be at least the topic's, and a reader must treat a missing object as a case to handle rather than a crash.