02. Distributed Message Queue — Low-Level Design¶

~20 min read · Part 3 of 4 (Overview → HLD → LLD → Q&A)

The HLD named the boxes. This file opens the three that carry the design's weight — the on-disk partition log, the offset-and-consumer-group machinery, and the ISR replication protocol — and pins down the data formats, the algorithms, and the concurrency corners where a queue actually breaks.

Data models¶

A partition is not a table; it is a directory of segment files plus two index files per segment. The log stores records; the indexes make an offset seekable without scanning.

topic "txns", partition 7  →  /kafka-logs/txns-7/
    00000000000000000000.log     # segment: records for offsets 0 .. 5_242_879
    00000000000000000000.index   # offset  → byte position (sparse)
    00000000000000000000.timeindex# timestamp → offset (sparse)
    00000000005242880.log        # next segment starts at its base offset
    00000000005242880.index
    ...

The segment file name is the base offset of its first record, so locating the segment for offset 5_400_000 is a binary search over file names, not a scan. Each .index maps a subset of offsets to exact byte positions (sparse — one entry per few KB), so a seek does a binary search in the index then a short sequential read in the .log. The .timeindex exists so a consumer can say "give me everything since 09:00" and have the broker translate that timestamp to an offset.

A single record on disk carries what ordering and exactly-once need:

Record {
    offset:            int64     # position in the partition, assigned by leader
    timestamp:         int64
    key:               bytes     # routing key; hash(key) chose this partition
    value:             bytes
    headers:           [kv]
    producer_id:       int64     # set by idempotent producer
    producer_epoch:    int16     # fences zombie producers after restart
    sequence_number:   int32     # per (producer_id, partition), for dedup
}

producer_id, producer_epoch, and sequence_number are the three fields that make idempotent production work: the broker remembers the last sequence it accepted per producer per partition and rejects a replay. Consumer offsets live in their own compacted topic rather than beside the data:

__consumer_offsets, key = (group_id, topic, partition)
                    value = (committed_offset, metadata, commit_timestamp)

Log compaction keeps only the latest value per key, so this topic stays small — one row per (group, partition) regardless of how many times a group commits.

Component internals¶

Component 1 — The partition log (append and seek)¶

Responsibility: append records sequentially at high throughput and let any reader jump to an arbitrary offset in roughly constant time.

class PartitionLog:
    def append(records: Batch) -> int64          # returns base offset, appends to active segment
    def read(start_offset: int64, max_bytes) -> Batch
    def _active_segment() -> Segment             # rolls a new segment at size/time limit
    def truncate_to(offset: int64) -> None       # follower uses this to undo un-replicated tail

append writes to the tail of the active segment in page cache and returns immediately; the OS flushes to disk in the background, and durability against a broker restart comes from replication, not from fsync on every write (an fsync-per-message policy would cap throughput far below 1M/s). read resolves start_offset to a byte position through the sparse index, then hands the raw bytes to the socket via a zero-copy sendfile, so the broker never parses the records it serves — a 5 MB/s partition and a 30 MB/s catch-up read cost the broker almost the same CPU.

Component 2 — Consumer group coordination and offset commit¶

Responsibility: divide a topic's partitions among a group's members, track each member's progress, and reassign on membership change without double-processing or gaps.

class GroupCoordinator:
    def join_group(group, member) -> assignment      # triggers rebalance
    def commit_offset(group, topic, partition, offset) -> None
    def fetch_committed(group, topic, partition) -> int64
    def heartbeat(group, member) -> ok | rebalance_in_progress

Each consumer commits the offset of the next event it expects, i.e. last_processed + 1, so on restart it resumes exactly after the last event it finished. The order of process-then-commit is the whole delivery-semantics story: commit after processing gives at-least-once (a crash before commit reprocesses the batch); commit before processing would give at-most-once (a crash after commit skips the batch). There is no "commit exactly with processing" without transactions, which is why exactly-once needs the extra machinery in the edge-cases section.

Component 3 — ISR replication and high-water mark¶

Responsibility: keep RF copies of a partition, acknowledge only what is safely replicated, and never expose an event to consumers before it is durable.

class Leader:
    def on_produce(batch) -> offset               # append + await ISR fetch
    def on_follower_fetch(follower, fetch_offset)  # serve replication + advance HWM
    def high_water_mark() -> int64                # min(LEO across ISR); consumers read up to here
    def maybe_shrink_isr()                        # drop followers past replica.lag.time

The leader tracks each replica's log-end offset (LEO) — how far it has replicated — and defines the high-water mark (HWM) as the minimum LEO across the current ISR. Consumers are only ever served events up to the HWM, so a consumer can never read an event that is not yet on every in-sync replica. That single rule is why a leader failover loses nothing a consumer already saw. When acks=all, the producer's ack is released at the same moment the HWM advances past its batch.

Core algorithm — the day-behind consumer catches up¶

This is the scenario made mechanical. The warehouse loader group returns after a day down, 86.4 billion events behind, while producers keep writing 1M/s. Walk what each of its 200 consumers does.

Resume position. Each consumer calls fetch_committed(group, "txns", p) and gets the offset it committed a day ago — say partition 7 committed at offset 4.32 × 10^9. The partition's current log-end offset is ~4.752 × 10^9 (a day of 5,000/s added 432 million events). The gap, 432 million events on this partition, is this consumer's backlog.
Retention check (the make-or-break step). The broker confirms offset 4.32 × 10^9 still lives on disk. Retention is 7 days and the data is 1 day old, so the segment is present. Had retention been 24 hours, the segment holding 4.32 × 10^9 would have been unlinked and the fetch would return OffsetOutOfRange; the consumer's auto.offset.reset policy would then jump it to the earliest or latest offset — either silently skipping the missing day or replaying from the log's start. This is where "catch up without losing data" is won or lost, and it was decided when retention was configured, not now.
Bulk fetch. Each consumer requests large batches (max.partition.fetch.bytes raised for catch-up) starting at its committed offset. The broker sendfiles contiguous segment bytes — sequential disk reads at the disk's full bandwidth, roughly 30 MB/s per partition here, 6 GB/s across the group.
Process and commit forward. The consumer processes each batch and commits last_processed + 1. It processes at ~30,000 events/s per partition versus the 5,000/s still arriving, so it gains 25,000 events/s per partition, 5,000,000/s across 200 partitions.
Convergence. Backlog 86.4 × 10^9 drained at 5 × 10^6 surplus/s finishes in ~17,280 s ≈ 4.8 hours. As each consumer's committed offset approaches the log-end offset, its fetches shrink from full segments to the live trickle, lag falls to seconds, and the group is caught up.
Ordering preserved throughout. Because each account's events are confined to one partition and one consumer reads that partition, catch-up replays each account's events in the exact original order — speed changed, order did not. That is the payoff of keying by account_id at produce time.

Sequence diagram — an `acks=all` write under a follower drop¶

Producer      Leader(P7)        Follower B (ISR)    Follower C (ISR)
   │  send(key,value)  │              │                  │
   ├──────────────────▶│ append @off=N│                  │
   │                   │  (page cache)│                  │
   │                   │◀── fetch(N) ─┤ replicate        │
   │                   ├── records ──▶│ append @off=N    │
   │                   │              │                  │
   │                   │◀───────────── fetch(N) ─────────┤
   │                   ├───────────── records ──────────▶│ append @off=N
   │                   │                                 │
   │        (C stalls; misses replica.lag.time.max)      │
   │                   │  maybe_shrink_isr(): ISR={Leader,B}
   │                   │  HWM = min(LEO over {Leader,B}) = N
   │◀── ack(off=N) ────┤  (released once HWM ≥ N; no wait on C)
   │                   │
   │                   │  ...later C catches up, rejoins ISR

The write is acknowledged the instant the HWM covers it across the current ISR — dropping the stalled follower C from the ISR is what unblocks the ack rather than letting one slow disk stall every producer. The event was on the leader and follower B before the ack, so the loss of any single broker after this point keeps it.

Concurrency and edge cases¶

Duplicate on producer retry (idempotence). A producer sends batch seq=42, the leader appends it, but the ack is lost to a network blip; the producer retries seq=42. The broker, remembering the last accepted sequence per (producer_id, partition), sees 42 is already applied and returns success without appending a second copy. Without this, every retry under load doubles messages — and load is exactly when retries spike.
Zombie producer fencing. A producer instance hangs, a replacement starts and registers a higher producer_epoch, then the original wakes and tries to write. The broker rejects the stale epoch, so a network-partitioned old instance cannot corrupt the stream after its replacement took over.
Exactly-once across process-and-produce. A stream processor reads from topic A and writes to topic B; a crash between the B-write and the A-offset-commit would, under at-least-once, reprocess and double-write to B. Kafka transactions bind "append to B" and "commit A's offset" into one atomic unit, and consumers of B set isolation.level=read_committed so they never see the messages of an aborted transaction. This is the only way to get true exactly-once, and it cuts effective throughput because of the extra coordination — the reason it is opt-in, not default.
Rebalance and the double-processing window. When a consumer dies, the partition it owned is reassigned to another member, which resumes from the last committed offset — so any events the dead consumer processed but had not yet committed are reprocessed by the new owner. At-least-once again; the defense is idempotent downstream writes, not tighter commits.
Hot partition from key skew. If one account_id produces 100,000 events/s, hashing pins all of them to one partition doing 100k/s while its 199 siblings idle at 5k/s — a self-inflicted bottleneck that adding brokers cannot fix, because a partition is single-leader and single-consumer-in-a-group. The fix is at the key: sub-key the hot account (account_id#shard) to spread it, accepting that its events lose strict cross-shard order — the direct price of parallelism.
Reading beyond the high-water mark. A consumer can never fetch past the HWM even though the leader's log physically extends further, because those tail events are not yet on all ISR members and could vanish in a failover. This is why a just-produced event with acks=1 may be briefly invisible to consumers: it is on the leader but not yet replicated, so it sits above the HWM until the followers catch up.