Apache Kafka and the distributed log
Kafka is an append-only log split into partitions, with readers that own their offsets. Covers key ordering, commit order, replication with acks and min.insync.replicas, and idempotent producers.
Apache Kafka is a distributed, append-only log split into partitions, where consumers track their own read positions (offsets) instead of the broker deleting messages. The detail interviewers expect: ordering is guaranteed only within a partition, and committing the offset after processing gives at-least-once delivery, while committing before gives at-most-once.
Apache Kafka is a distributed, append-only log: producers add records to the end of partitions, and consumers read at positions they track themselves. That single design explains why many consumers can read the same data, why ordering holds only per partition, and why a crash gives duplicates or loss depending on when the offset is committed.
- Kafka guarantees record order only within a single partition; records with the same key hash to the same partition, and changing the partition count remaps keys.
- Within one Kafka consumer group, each partition is assigned to exactly one consumer, so ten consumers on six partitions leave four idle.
- A Kafka consumer that processes a batch and then commits its offset gets at-least-once delivery; committing before processing gives at-most-once and can lose records after a crash.
- With 3 replicas, acks=all and min.insync.replicas=2, a Kafka partition keeps accepting writes with one broker down and rejects writes with NotEnoughReplicas when two are down.
- In Kafka, acks=all means all replicas currently in the in-sync replica set (ISR), so with min.insync.replicas=1 it can behave like acks=1.
- Kafka's idempotent producer (KIP-98) tags records with a producer id and per-partition sequence number so the broker drops retried duplicates, within one producer session.
01A log is the whole idea
Picture a text file where you only ever add lines at the bottom. You never edit a line and you never remove one from the middle. Each line gets a number, its position in the file, and that number never changes. That is a log, and it is the data structure Kafka is built on. The original 2011 paper from LinkedIn describes each partition as a set of segment files and says messages are addressed by their logical offset in the log, not by a message id.
A traditional queue works differently. When a consumer takes a message and acknowledges it, the broker deletes it, so the broker has to remember who holds what. A log makes the opposite choice: the broker keeps the data and forgets the readers. Each reader remembers its own position and moves it forward when it is ready. Two readers can sit at different places in the same log, and a reader that wants to reprocess last Tuesday just moves its position back. The article on Message Queues and Delivery Guarantees covers the queue side and the vocabulary of at-most-once and at-least-once. This one adds what changes when the broker is a log instead.
Take one question: why can a fraud-detection service and a nightly report job both read the same click stream? Nothing is deleted when the first one reads, so the second finds the data waiting. Now take another: why is ordering only per partition? A log has exactly one order, and a topic is several logs side by side. The rest of the article keeps asking what an append-only log with reader-owned positions would do. That also connects to Batch vs Stream Processing: a stream processor needs a source it can re-read after a failure, and a retained log is exactly that.
The code below is a small Python model of Kafka's rules, not a broker. The file timings measure plain files, not Kafka. The rules come from the Kafka 3.0 design documentation, the 2011 paper and KIP-98.
02Topics and partitions: parallelism and ordering
One log on one machine has a ceiling. Kafka splits a topic into partitions, each an independent log that can live on a different broker. Producers pick the partition for each record. When a record has a key, the usual rule is a hash of the key modulo the number of partitions, so the same key always lands in the same log. (Kafka's own default partitioner uses a different hash, murmur2; CRC32 here keeps the model short.)
This is a pure-Python model, not a broker. It sends three records for each of three user keys, interleaved, then moves to four partitions to see where the same keys would go.
import zlib
class Topic:
def __init__(self, n): self.parts = [[] for _ in range(n)]
def part(self, key): return zlib.crc32(key.encode()) % len(self.parts)
def send(self, key, val):
p = self.part(key)
self.parts[p].append(val)
return p
t = Topic(3)
for i in range(3): # interleave three users
for u in ("user-1", "user-2", "user-3"):
t.send(u, f"{u}#{i}")
for p, log in enumerate(t.parts):
print("partition", p, log)
four = Topic(4) # same keys, one more partition
for u in ("user-1", "user-2", "user-3"):
print(u, "3 partitions ->", t.part(u), "| 4 partitions ->", four.part(u))Output of that script:
partition 0 ['user-3#0', 'user-3#1', 'user-3#2']
partition 1 ['user-2#0', 'user-2#1', 'user-2#2']
partition 2 ['user-1#0', 'user-1#1', 'user-1#2']
user-1 3 partitions -> 2 | 4 partitions -> 0
user-2 3 partitions -> 1 | 4 partitions -> 2
user-3 3 partitions -> 0 | 4 partitions -> 0Each user's events are in order inside one partition, even though the sends were interleaved. Between partitions there is no order at all: nothing says user-1's first event came before user-2's. The design documentation states the same rule, a topic is a set of totally ordered partitions.
The last three lines show the trap. Change the partition count from 3 to 4 and user-1 moves from partition 2 to 0, and user-2 from 1 to 2. New records for those users start in a different log while the old ones stay where they were, so a consumer can see a user's newest event before the older ones are processed. Choosing the partition count with headroom before launch is cheaper than repairing this later. Skew is the other trap: if one key produces most of the traffic, its partition becomes a hot spot while the others sit idle, and adding partitions does not help because that key still hashes to one of them.
03Offsets and consumer groups
An offset is a record's position in its partition, and a consumer's position is the offset of the next record it will read. The design documentation points out that this makes the "what has been consumed" state tiny: one integer per partition, which can be checkpointed cheaply. Consumers pull from the broker rather than being pushed to, so a slow one falls behind instead of being flooded, and it can deliberately rewind to an old offset after a bug fix.
A consumer group is a set of consumers that share a topic's work. Inside one group, the broker hands each partition to a single consumer, and it stays with that consumer until the group rebalances (the design documentation describes this rule). So the partition count is the ceiling on parallelism: ten consumers on six partitions leave four idle. A second group reading the same topic gets the whole stream again with its own offsets. Same data, independent positions, and that is the practical difference from a queue.
The commit order decides what a crash costs. The model below reads a ten-record partition in batches of five. The consumer dies while handling m7 (once), restarts, and reads its last committed offset. It is run twice: once committing after the work, once before.
records = [f"m{i}" for i in range(10)]
def run(commit_first, crash_at="m7", batch=5):
committed, handled, crashed = 0, [], False
while committed < len(records):
batch_recs = records[committed:committed + batch]
if commit_first:
committed += len(batch_recs) # offset says "done" up front
try:
for r in batch_recs:
if r == crash_at and not crashed:
crashed = True
raise RuntimeError("consumer died")
handled.append(r)
except RuntimeError:
continue # restart: re-read committed offset
if not commit_first:
committed += len(batch_recs) # offset moves after the work
dups = sorted({r for r in handled if handled.count(r) > 1})
lost = [r for r in records if r not in handled]
return len(handled), dups, lost
for name, flag in (("process then commit", False), ("commit then process", True)):
n, d, l = run(flag)
print(f"{name}: handled {n} records, duplicates={d}, lost={l}")process then commit: handled 12 records, duplicates=['m5', 'm6'], lost=[]
commit then process: handled 7 records, duplicates=[], lost=['m7', 'm8', 'm9']Process-then-commit is at-least-once. The crash lands before the commit for batch m5 to m9, so the restarted consumer reads offset 5 and handles m5 and m6 a second time. Commit-then-process is at-most-once: the offset already claimed the whole batch was done, so m7, m8 and m9 are never handled. In practice most pipelines take the first option and make the handler safe to repeat, for example by writing with an upsert keyed on the record id.
04Storage: segments, retention and why disk is fast
A partition is not one giant file. It is a series of segment files, and the broker appends to the newest one. Each segment file is named after the offset of the first message it contains, so the files are 00000000000, then a larger number, and so on. To read offset N the broker picks the segment whose starting offset is the largest one not above N, using a binary search over the segment list, then uses that segment's offset index to jump close to the right byte position and scans forward from there. The index is sparse: it holds an entry every few kilobytes rather than one per record, which keeps it small enough to stay in memory.
Old data leaves in two ways. With the default delete policy, whole segments are dropped once they pass the retention limit, by age (the 2011 paper mentions seven days) or by total size, and it does not matter whether anyone has read them. A topic can instead be set to log compaction, which keeps at least the latest record for every key and discards older records with the same key. That suits topics that describe current state, such as the latest address per customer, and it is the exception to the idea that Kafka only forgets by age or size.
The design documentation gives two reasons the layout is quick. Kafka leans on the operating system's page cache instead of building a cache inside the process, and it uses the sendfile call so data can go from the cache to the network without being copied through the application. Both only work well because the access pattern is simple. The script below shows how much that pattern matters for plain files: it writes 200,000 records of 100 bytes (20 MB) sequentially, overwrites the same positions in shuffled order, then reads them back both ways.
import os, random, tempfile, time
N, SIZE = 200_000, 100
rec = b"x" * SIZE
path = tempfile.mktemp()
t = time.perf_counter()
with open(path, "wb") as f:
for _ in range(N): f.write(rec)
f.flush(); os.fsync(f.fileno())
seq_w = time.perf_counter() - t
order = list(range(N)); random.Random(1).shuffle(order)
t = time.perf_counter()
with open(path, "r+b", buffering=0) as f:
for i in order:
f.seek(i * SIZE); f.write(rec)
os.fsync(f.fileno())
rnd_w = time.perf_counter() - t
t = time.perf_counter()
with open(path, "rb", buffering=0) as f:
while f.read(SIZE): pass
seq_r = time.perf_counter() - t
t = time.perf_counter()
with open(path, "rb", buffering=0) as f:
for i in order:
f.seek(i * SIZE); f.read(SIZE)
rnd_r = time.perf_counter() - t
os.remove(path)
print(f"sequential append {seq_w:.3f}s -> {N/seq_w:,.0f} rec/s")
print(f"random overwrite {rnd_w:.3f}s -> {N/rnd_w:,.0f} rec/s ({rnd_w/seq_w:.1f}x slower)")
print(f"sequential read {seq_r:.3f}s, random read {rnd_r:.3f}s ({rnd_r/seq_r:.2f}x)")sequential append 0.295s -> 678,940 rec/s
random overwrite 1.545s -> 129,449 rec/s (5.2x slower)
sequential read 0.748s, random read 1.501s (2.01x)Repeated runs on one ordinary Linux machine gave a random-write penalty between 3.2x and 6.6x and a random-read penalty between 2.0x and 3.3x; the block above is one run, at 5.2x for writes. The spread is the finding: the ratio depends on the machine and on what the page cache is doing. With a single fsync at the end, most of this measures the page cache and the system-call cost of seeking, not a physical disk, and none of it is a Kafka benchmark. What it does show is the direction. Appending to the end of a file is the access pattern storage handles best, and a log never asks for anything else.
05Replication: leader, followers, ISR and acks
One copy of a partition is a single point of failure. Each partition therefore has one leader and zero or more followers on other brokers. Producers and consumers talk to the leader. Followers fetch from it the way an ordinary consumer would, pulling batches of records and appending them to their own log, so a follower's log is the leader's log with the same offsets in the same order, possibly a few records behind.
Kafka does not use a majority vote for the data. It keeps a set called the in-sync replicas, the ISR: replicas that are alive and caught up with the leader. A write counts as committed once every member of the ISR has it. The high watermark is the offset up to which that is true, and consumers can only read below it, so a record that exists only on the leader is invisible until the followers catch up. A follower that stops fetching or falls too far behind is dropped from the ISR, which then shrinks. Only ISR members may be elected leader, which is why a committed record survives a leader crash. The design documentation sizes this by replica count: a topic with f+1 replicas can lose f of them and still keep every committed record.
The producer's acks setting controls when a send counts as done. With acks=0 the producer does not wait. With acks=1 the leader has written it, but if the leader dies before any follower copies the record, the record is gone even though the producer was told it succeeded. With acks=all the acknowledgement comes after every current ISR member has the record. The wording matters: "all" means all in-sync replicas, not all assigned replicas.
Take three replicas, acks=all and the topic setting min.insync.replicas=2. All three up: a write needs three copies, fine. One broker down: the ISR is two, still at the minimum, so writes carry on and nothing acknowledged is lost. Two brokers down: the ISR is one, below the minimum, so the leader rejects writes and the producer gets a NotEnoughReplicas error instead of a silent loss. Now set min.insync.replicas=3 on the same three replicas and a single broker restart blocks every write to the partition. That is the real cost of a stricter setting, and it is why 2 of 3 is the usual choice.
Leave min.insync.replicas at 1 and acks=all stops meaning much. The documentation's own example: with two replicas and one failed, an acks=all write succeeds because the lone leader is the whole ISR, and it is lost if that leader then dies. In that state acks=all behaves like acks=1.
The last knob is unclean.leader.election.enable. If every ISR member is down and a lagging replica comes back, electing it as leader restores availability but makes its shorter log the source of truth, so committed records vanish. By default Kafka (since 0.11) waits for an in-sync replica instead. Turning unclean election on trades durability for uptime, which suits some metrics topics and never a payments topic.
06Delivery guarantees and exactly-once
The delivery vocabulary (at-most-once, at-least-once, exactly-once) is laid out in Message Queues and Delivery Guarantees. For Kafka, the question is what the producer can promise when an acknowledgement goes missing. Before KIP-98 the answer was at-least-once: the producer retries, and the log may now hold the record twice.
KIP-98 added the idempotent producer. The producer gets a producer id (PID) and numbers its records per partition, starting at 0. The broker remembers the last sequence number it accepted for that PID. A lower number is a duplicate and is dropped; a number higher than expected is an out-of-sequence error. The model below loses the first acknowledgement for every record, so the producer sends each one twice.
class Broker:
def __init__(self): self.log, self.last = [], {}
def append(self, pid, seq, v):
exp = self.last.get(pid, -1) + 1
if seq < exp: return "duplicate"
if seq > exp: return "out-of-sequence"
self.last[pid] = seq; self.log.append(v); return "ok"
def send_all(values, use_seq):
b = Broker()
for seq, v in enumerate(values):
for attempt in range(2): # first ack is lost, so the producer retries
if use_seq: b.append("pid-7", seq, v)
else: b.log.append(v) # plain log: no check
return b.log
print("plain retries ->", send_all("abc", False))
print("idempotent ->", send_all("abc", True))plain retries -> ['a', 'a', 'b', 'b', 'c', 'c']
idempotent -> ['a', 'b', 'c']KIP-98 states the limit: the guarantee holds within a single producer session, and a restarted producer gets a new PID. Transactions fill the gap. With a transactional.id, a producer can write to several partitions atomically, and sendOffsetsToTransaction commits the consumed offsets in the same transaction, so consume, transform and produce either all happen or none do. A producer epoch fences out an older "zombie" instance that has the same id. Consumers must set isolation.level=read_committed, because the default is read_uncommitted and they would otherwise see records from aborted transactions.
On defaults: KIP-679 changed the producer default of enable.idempotence to true in Kafka 3.0, but a bug in the config validation (KAFKA-13598) kept the new default from taking effect, so producers that did not set it explicitly were still not idempotent. The fix was merged for 3.0.1, 3.1.1 and 3.2.0; the acks=all default did take effect. Idempotence needs acks=all, retries above 0 and max.in.flight.requests.per.connection of at most 5. Earlier versions, 0.11 through 2.8, default it to false. Whatever the version, set enable.idempotence=true explicitly on producers that must not duplicate.
So "exactly-once" covers records moving from Kafka to Kafka. If the processing step sends an email or updates an external database, a retry repeats that side effect, and the fix is an idempotent write on that side, such as an upsert keyed on a record id.
07Common interview questions
Q: How is Kafka different from a traditional message queue?
A: A queue deletes a message once a consumer acknowledges it and tracks who has received what. Kafka keeps messages in an append-only log until retention expires, and each consumer group tracks its own offset. Many groups can read the same data, and a group can rewind to reprocess.
Q: What ordering does Kafka guarantee, and how do you choose a key?
A: Order is guaranteed within one partition only. Use a key that groups records which must stay in order, such as a user id or an order id, so they hash to one partition. Watch for hot keys, and avoid changing the partition count later, because it remaps keys.
Q: A consumer crashes after processing a batch but before committing. What happens?
A: It restarts from the last committed offset and processes the batch again, so you get duplicates. That is at-least-once. Committing before processing flips the failure into data loss, which is at-most-once. Make processing idempotent and commit after.
Q: How do you avoid losing an acknowledged message, and what is the trap with acks=all?
A: Use replication factor 3, acks=all and min.insync.replicas=2, and leave unclean leader election off. A write is acknowledged only when at least two in-sync replicas hold it. The trap: acks=all means all replicas currently in the ISR, so with min.insync.replicas=1 the ISR can shrink to the leader alone and acks=all behaves like acks=1.
Q: Does Kafka provide exactly-once?
A: Within Kafka, yes, using the idempotent producer plus transactions and read_committed consumers for read-process-write. It does not cover external side effects like emails or database writes, which need their own idempotency.
Q: Why is Kafka fast when it writes to disk?
A: It only appends sequentially, it relies on the operating system page cache, and it can use sendfile to move data to the network without copying it through the application.