How a log-based queue stores data: segments, offsets, retention, compaction, replication

The interviewer says: "You chose a log for the order events. How does it store 7 days of data on disk and still read fast? What happens to a message if a server dies?"

A weak answer says "Kafka is fast" or "it keeps data in memory". A strong answer says that a partition is a set of files that are only appended to. It says how a read finds its place, how old data goes away, and how copies decide what a crash can lose.

At SDE-2 level you should explain appending and retention. At SDE-3 level you should explain compaction and the acknowledgement settings, with numbers.

Think of a shop that writes every sale in numbered notebooks. Only the newest notebook is open. Each notebook has the number of its first entry on the cover. When a notebook is full, a new one starts. After 7 days the shop throws away whole old notebooks. A few tabs inside each notebook point to page numbers, so finding entry 730,000 does not need a page-by-page search.

A partition is an ordered log with offsets (lesson 87). On disk it is a directory of segment files. Only the newest, the active segment, is written. Every write goes to the end of it. A segment is named by the offset of its first message. When it reaches a size limit or an age limit, a new one starts. The numbers below are the defaults of a Kafka-style system, which is the example in this lesson.

Why appending is fast

• Sequential writes. A disk is far faster when it writes in one long run than in many small places. The Kafka design page gives 600 MB per second for sequential writes and 100 kB per second for random writes. Its example is an array of six disks.
• The page cache. The operating system keeps recent file data in memory, so recent reads often need no disk read. The broker relies on this cache and does not build its own.
• Batching. Many messages travel and are written in one request.
• No copy through the program. A consumer's read can go from the file cache to the network without being copied into the program. The design page says this saves copies (the call is named sendfile).

The broker does not force a flush to disk after each message by default. It leaves that to the operating system, and relies on copies for safety. If a power cut hit every copy at the same moment, messages not yet written to disk could be lost.

Finding an offset

A read at offset N has two steps.

1. Find the segment. It is the file with the largest first offset that is not above N. File names are sorted, so a binary search finds it.
2. Use the index. Each segment has a small index file that maps an offset to a byte position. The index is sparse: Kafka adds an entry about every 4,096 bytes of data (default). The read jumps to the nearest entry at or before N and scans forward at most 4,096 bytes. In Kafka's source code, each entry is 8 bytes. It holds a 4-byte offset counted from the start of the segment and a 4-byte position in the file.

Retention

Retention removes old data. Cleanup works one file at a time, so it deletes whole segments, by age (default 7 days) or by size per partition (default no limit). In Kafka's source code, a segment goes by age only when its newest message is older than the limit. So an old message can stay about one segment longer than the limit. The segment that is being written is never deleted until it rolls. Deleting a file is cheap.

Compaction

Compaction is a different cleanup. It keeps at least the latest message for each key, and removes older messages with the same key. It runs in the background on closed segments, so for a while older messages with the same key still exist. Messages that stay keep their offsets. It is not the merge of sorted files in lesson 74.

A delete is a message with a key and no value. It is a tombstone. It is kept for a time (default 24 hours) so that readers see it, and then it is removed. With compaction, the size follows the number of keys, not the number of updates. It fits a topic that holds the current state of each user.

Replication

Each partition has one leader and followers that copy it (lessons 65 and 67). The in-sync replicas are the leader and the followers that are caught up. A message is committed when all in-sync replicas have it. Consumers are given only committed messages, so they never read a message that a leader crash could take back. The offset just after the last committed message is called the high watermark.

The producer chooses when it hears "saved". This is the acks setting. With acks=0, the producer does not wait at all. With acks=1, the leader answers after its own write. With acks=all (the default now), it answers when all in-sync replicas have the message. Kafka's promise is that a committed message is not lost while at least one in-sync replica is alive. A leader-only message can be lost if the leader dies before the followers copy it.

The setting min.insync.replicas (default 1) is the smallest in-sync group that may accept a write when acks=all. If the group is smaller, the broker refuses the write with an error, and the producer must retry or report a failure. With the default of 1, "all" can mean only the leader. A third setting decides if a replica that is behind may become leader when no in-sync replica is left. The default is no, which keeps data and loses availability.

A partition of an order topic.

Assumptions (made up to teach with): 2 MB per second into the partition, in messages of 1,000 bytes. A segment is 1 GiB (1,073,741,824 bytes), the default limit. Retention is 7 days. A follower is 50 ms behind. A disk write takes 1 ms and a trip between servers takes 0.5 ms (lesson 65). An index entry is 8 bytes (from Kafka's source code, see Sources).

Step 1: segments. A segment fills in 1,073,741,824 / 2,000,000 = 537 seconds, about 9 minutes. In 7 days: 2,000,000 x 86,400 x 7 = 1,209,600,000,000 bytes. Segments: 1,209,600,000,000 / 1,073,741,824 = 1,126.5, so 1,127 files. With 3 copies: 3 x 1.2096 TB = 3.63 TB.

Step 2: a read. For the predict box, the file that starts at 500,000 holds offset 730,000. It has the largest start not above 730,000. With 1,127 files, a binary search takes at most 11 steps. In a segment, the index has 1,073,741,824 / 4,096 = 262,144 entries, and 262,144 x 8 = 2 MiB. Finding the entry takes at most 18 steps. Then the read scans at most 4,096 / 1,000 = 4 messages.

A partition is split into segment files named by their first offset: one starts at 0, one at 500,000 and one at 1,000,000. The last one is the active segment where writes go. A read at offset 730,000 goes to the segment that starts at 500,000, because it is the file with the largest start that is not above 730,000.

Step 3: disk work. The partition writes 2 MB per second. A disk with 600 MB per second of sequential writes could take 600 / 2 = 300 such partitions. With 100 kB per second of random writes, it could not take one.

Step 4: retention. A segment closes every 9 minutes. Retention removes the oldest of the 1,127 files by deleting it, with no rewrite. A segment goes only when its newest message is older than 7 days, so its oldest message stays past 7 days. The extra time is up to about one segment, 9 minutes here.

Step 5: compaction. A topic holds a profile for each of 10,000,000 users, 500 bytes each. Assumption: 50 updates a user a year. If every update is kept: 10,000,000 x 50 x 500 = 250 GB. With compaction: 10,000,000 x 500 = 5 GB. Small example with offsets 0 to 5: (u1,A), (u2,B), (u1,C), (u3,D), (u2,none), (u1,E). After compaction: offset 3 (u3,D), offset 4 (u2,none), offset 5 (u1,E). The tombstone at offset 4 goes after 24 hours.

Step 6: a leader dies. The partition takes 5,000 messages per second. With acks=1, up to 5,000 x 0.05 = 250 acknowledged messages are only on the leader. If it dies now, they are lost. With acks=all, 0 are lost while one in-sync copy lives. The cost: 1 ms with acks=1, and 1 + 0.5 + 1 = 2.5 ms with acks=all.

Step 7: the minimum. There are 3 copies. Two followers fall behind for an hour. With min.insync.replicas=1, the leader alone accepts writes and says "saved". If its disk then dies, an hour of data is gone, although acks=all was set. With min.insync.replicas=2, writes are refused in that hour. Assumption: each server is up 99.9% of the time, independently. Writes work if at least 2 of 3 are up: 99.9997% of the time, about 0.13 minutes down in a 43,200-minute month. Needing all 3 (min.insync.replicas=3) gives 99.7%, about 129 minutes down.

What a company wrote

LinkedIn (Jay Kreps, 27 Apr 2014) wrote that replication is the normal case, and a topic without copies is the special case with one copy. In a test with 3 copies on 3 servers, copying in the background gave 786,980 records per second, against 821,557 with no copies. That is 4% less. When the producer waited for the copies, it sent 421,823 records per second, about half (the company's figures). Link: https://engineering.linkedin.com/kafka/benchmarking-apache-kafka-2-million-writes-second-three-cheap-machines (opens in a new tab)

Copies with acks=all solve lost acknowledged messages (250 against 0). They cost 1.5 ms more for each write and refused writes when too few copies are in sync. A longer retention solves late readers and replays, and costs disk (3.63 TB for 7 days). I would change to acks=1 only for data that a new message replaces, such as a location ping. I would use compaction when readers need the current state, and retention when they need the history.

"A log queue is fast because it keeps messages in memory." It appends to files in long runs and lets the operating system cache them. At 2 MB per second, one disk with fast sequential writes could take 300 partitions.

"With acks=all, no acknowledged message can be lost." The in-sync group can shrink to the leader. With a minimum of 1, "all" then means one server. In step 7, an hour of data was at risk.

"Compaction deletes messages older than the retention time." That is retention. Compaction keeps the latest message for each key, however old, and removes older ones with the same key.

1. How is a segment file named? What does retention delete, and what does compaction keep? What is a tombstone?
2. A partition takes 4 MB per second. A segment is 1,073,741,824 bytes and retention is 3 days. (a) How often does a segment roll? (b) How many segments exist after 3 days? (c) With one index entry for each 4,096 bytes, how many entries does one segment have?
3. A topic has 3 copies, acks=all and min.insync.replicas=1. Two followers are down for an hour. Then the leader's disk fails. What is lost, and what do you change?