Engineering Atlas › System Design Interviews

Design a distributed message queue

20 min read · last reviewed 2026-10-17

Open in the interactive app: practice, quiz, mock interview →

In one minute

Requirements

Functional

Non-functional

Out of scope: per-message delays and priorities (see Design Job Scheduler), routing with exchanges (see RabbitMQ), the schema registry, and stream processing on top of the log (see Flink).

Back-of-envelope

These are round, assumed numbers for a company-wide platform.

API

The wire protocol is binary over long-lived TCP connections. The shapes below are shown as pseudo-RPCs.

Produce(topic, partition, acks="all",
        producer_id, producer_epoch, base_sequence,
        batch: [ { key, value, headers, timestamp } ... ])
  -> { base_offset, log_append_time } | error NOT_LEADER | OUT_OF_ORDER_SEQUENCE

Fetch(replica_id or -1, [ { topic, partition, fetch_offset, leader_epoch } ],
      max_wait_ms=500, min_bytes=1, max_bytes=50MB)
  -> [ { partition, high_watermark, log_start_offset, records } ]

Metadata(topics) -> { brokers, [ { topic, partition, leader, replicas, isr } ] }

JoinGroup(group, member_id, session_timeout_ms) -> { generation, leader, members }
SyncGroup(group, generation, member_id, assignments?) -> { my_partitions }
Heartbeat(group, generation, member_id) -> ok | REBALANCE_IN_PROGRESS | UNKNOWN_MEMBER
CommitOffset(group, generation, member_id, [ { topic, partition, offset } ])
  -> ok | ILLEGAL_GENERATION
ListOffsets(topic, partition, timestamp) -> { offset }     (rewind by time)

Producers send whole batches, not single records, because batching is what makes the throughput numbers above possible. Fetch serves both consumers and follower replicas. Follower fetches are how replication happens, and the offset a follower asks for tells the leader how far that follower has caught up. max_wait_ms and min_bytes turn polling into long polling, so an idle consumer costs almost nothing.

Data model

Partition logs live on broker disks, one directory per replica (orders-7/). Segments are named by their first offset: 00000000000004200000.log holds record batches, .index maps offsets to byte positions every 4 KB (log.index.interval.bytes defaults to 4096 (Kafka broker configs)), and .timeindex maps timestamps to offsets. Only the active segment takes appends. Retention deletes whole segments: one unlink instead of millions of row deletes.

Cluster metadata is itself a log, kept by a 3- or 5-node Raft controller quorum (KRaft). Kafka 4.0 (March 2025) was the first release with no ZooKeeper mode at all, so older designs that put the controller in ZooKeeper are now out of date (Kafka 4.0 announcement). Records such as PartitionRecord (replicas, leader, ISR, leader epoch) are replayed by every broker, so each has a local view of who leads what.

Committed offsets live in an internal compacted topic keyed by (group, topic, partition). A group maps to partition hash(group) mod 50 of that topic, whose leader is the group's coordinator. Compaction keeps only the latest offset per key.

Producer state is producer_id -> (epoch, last 5 batch sequences) per partition, snapshotted with segments, so a leader recognises retried batches.

User data is partitioned by murmur2(key) mod partitions. Keying orders by order_id gives per-order ordering; keying by customer_id gives per-customer ordering but turns a whale customer into a hot partition.

High-level design

flowchart LR
  P["Producers"] -->|"Produce batch"| L["Partition leader"]
  L -->|"fetch"| F1["Follower (zone B)"]
  L -->|"fetch"| F2["Follower (zone C)"]
  C["Controller quorum (Raft)"] -->|"metadata log"| L
  C --> F1
  C --> F2
  G["Group coordinator"] <-->|"join, heartbeat, commit"| CG["Consumer group"]
  CG -->|"Fetch from offset"| L
  L -->|"old segments"| T[("Object storage")]

Write path.

  1. The producer hashes the key to a partition and adds the record to that partition's batch, sent at 64 KB or when linger.ms expires. Kafka changed the linger.ms default from 0 to 5 ms in 4.0, because larger batches usually make latency no worse (Kafka producer configs).
  2. The compressed batch goes to the partition leader with the producer ID, epoch and base sequence.
  3. The leader checks the sequence, assigns offsets and appends to the active segment in the page cache.
  4. Followers in the in-sync replica set (ISR) are parked in long-poll fetches, so they pull the batch within a millisecond or two.
  5. Once every ISR member has fetched past it, the leader advances the high watermark and acknowledges. If the ISR is below min.insync.replicas (2 of 3), the leader rejects the write instead.

Read path.

  1. A consumer joins its group; the coordinator bumps the generation and assigns partitions.
  2. The consumer fetches each owned partition from its committed offset.
  3. The leader returns batches only up to the high watermark, so consumers never see data a failover could erase. Tail reads come from the page cache via sendfile.
  4. The consumer processes, then commits the next offset. A crash before committing means the next owner reprocesses: at-least-once delivery.

Deep dives

1. Replication: ISR, quorums and leader epochs

Options: a majority quorum per partition (Raft) needs 2f+1 replicas to survive f failures but ignores one slow node. Kafka's ISR model waits for every in-sync replica and drops one that lags more than replica.lag.time.max.ms (30 s by default (Kafka broker configs)), so f+1 replicas survive f failures, at the price of a strongly consistent controller tracking the ISR. I choose ISR with replication factor 3 and min.insync.replicas=2, because copies are the expensive part at petabyte scale.

The subtle part is failover. Say the old leader appended offsets 100 to 105, but only 100 to 102 reached the follower that becomes the new leader, which then writes new data at 103. When the old leader returns, it must discard its own 103 to 105. Leader epochs make this safe: each term has an epoch and a start offset, and a returning replica asks the leader where its epoch ended and truncates to that point before fetching. Unclean leader election (promoting a replica outside the ISR) trades durability for availability; it is off by default, and I allow it only for telemetry topics. Both knobs exist because Jepsen showed in 2013 that a leader alone in its ISR could acknowledge writes that a later election then threw away.

2. Consumer groups, ordering and rebalancing

Ordering holds only within a partition, so partitions cap parallelism: 12 consumers on 8 partitions leave 4 idle. I size topics for the parallelism expected in two years, because adding partitions later changes hash(key) mod n and breaks per-key ordering during the change.

In the classic eager protocol, every member gives up all partitions on each rebalance, so a 200-member group stalls for seconds whenever a pod restarts. Cooperative incremental rebalancing moves only partitions that change owner, static membership lets a quick restart skip the rebalance, and the server-side protocol from KIP-848 removes the global barrier altogether. It went GA in Kafka 4.0, but clients still opt in with group.protocol=consumer; the client default is still classic (Kafka consumer configs).

Generation fencing keeps rebalancing safe. A consumer stuck in a 40 s garbage-collection pause loses its session and its partitions. When it wakes and commits with generation 7 while the group is at 8, the coordinator rejects the commit. The Practice task builds exactly this.

For work queues, partitions are too coarse: one slow message blocks its whole partition. Either process by key in parallel inside the consumer, or offer per-message acknowledgement with a lock timeout. Kafka's share groups (KIP-932) do exactly this: more consumers than partitions, records acknowledged one by one, and they were declared production-ready in Kafka 4.2 in February 2026 (Kafka 4.2 announcement). SQS has the same per-message model with a visibility timeout of 30 s by default, extendable up to 12 hours, and warns that a message can still be delivered twice within it (SQS docs). I would offer both topic types.

3. Delivery semantics end to end

Producer retries are the first source of duplicates. With the idempotent producer, each batch carries a producer ID and a per-partition sequence number; the leader appends n+1, returns the original offset for a sequence it already has, and rejects a gap with OUT_OF_ORDER_SEQUENCE. Remembering the last 5 sequences allows 5 requests in flight while staying ordered, which is why idempotence requires max.in.flight.requests.per.connection of 5 or less (Kafka producer configs). The producer ID, per-partition sequences, epochs that fence zombie producers, and transactions all come from one design, KIP-98 (KIP-98).

Consumers are the second source: a crash between processing and committing means reprocessing. For log-to-log pipelines, transactions write output records and the offset commit atomically, and readers with isolation.level=read_committed skip aborted data. For a database or payment API, the consumer must make its effect idempotent, for example an upsert keyed by (topic, partition, offset) or a business ID. See Delivery Semantics and the Outbox Pattern.

Down the stack

Layer 1The contract

  • Acknowledged means replicated. With acks=all, once Produce returns base_offset = n, every ISR member holds the batch, and there are at least min.insync.replicas of them. If the ISR is too small, the write is rejected and not silently accepted.
  • Offsets are stable. An offset below the high watermark always refers to the same record on every replica, across any number of leader changes.
  • Per-partition order. Records from one idempotent producer to one partition are stored in send order, without duplicates, even when the producer retries.
  • Consumers see committed data only. Fetches return records below the high watermark, and with read_committed, also below the last stable offset (the first offset of the oldest transaction that is still open).
  • Commits are fenced. An offset commit with a stale generation fails with ILLEGAL_GENERATION.

Layer 2The mechanism

The on-disk unit is the record batch (format v2). Its header holds baseOffset (int64), batchLength, partitionLeaderEpoch, magic (=2), a CRC-32C, attributes (codec, transactional and control flags), lastOffsetDelta, baseTimestamp, maxTimestamp, producerId (int64), producerEpoch (int16) and baseSequence. Records inside store varint deltas from the batch base, so per-record overhead is a few bytes. Brokers store batches exactly as producers compressed them and never recompress on the hot path.

Each replica has a log end offset (LEO); the leader sets the high watermark to the minimum LEO across the ISR and tells followers in fetch responses. Offset lookup is a binary search over segments by base offset, then over the memory-mapped sparse .index, then a short scan of .log. The group coordinator runs a state machine (Empty, PreparingRebalance, CompletingRebalance, Stable), and each completed rebalance increments the generation.

Layer 3The machine

  • Appends are sequential writes. A write(2) into the page cache takes microseconds, with no fsync per batch; the kernel flushes dirty pages after about 30 s. NVMe absorbs 2 to 3 GB/s sequentially, so 250 MB/s of appends uses a tenth of a drive.
  • Tail reads are zero copy. On plaintext connections sendfile(2) moves page-cache pages to the socket without a user-space copy. TLS forces a read and encrypt in user space, roughly doubling CPU per byte.
  • Lagging consumers evict the cache. A consumer reading 3-day-old data pulls cold pages from disk and pushes the hot tail out of the page cache, so tail readers hit disk too. Tiered storage sends cold reads to object storage (20 to 100 ms per request) instead.
  • Replication costs a zone round trip. Cross-zone RTT is 0.5 to 2 ms, so acks=all adds about 2 to 5 ms including the follower's long-poll wake-up.
  • Batches amortise syscalls. A 64 KB batch of 1 KB records costs one request, one hardware CRC-32C and one append instead of 64.

Why it's built this way

  1. Disks and SSDs are much faster at sequential I/O than random I/O, so the store is an append-only log.
  2. An append-only log cannot delete individual messages cheaply, so consumers keep a cursor, and retention removes whole segments.
  3. A cursor per consumer makes replay and fan-out free, but ordering a log across many machines is expensive, so the log is split into partitions with ordering only inside each one.
  4. Calling fsync per write would cap throughput at the device's flush latency (Kafka's design notes put the cost at two to three orders of magnitude), so durability comes from copies on independent machines in different zones, and a replica that lost unflushed data must fully re-sync before rejoining the ISR (Kafka design).
  5. Copies can diverge after a failover, so leader epochs and a high watermark define which prefix is committed and which tail must be truncated.

Failure modes

In the wild

Trade-offs

OptionStrengthsWeaknessesChoose when
Partitioned log (Kafka, Pulsar)Replay, many consumer groups, very high throughputOrdering only per partition, head-of-line blocking, partition count hard to changeEvent streams, CDC, analytics pipelines
Per-message queue (SQS, RabbitMQ)Any worker takes any message, per-message retry and delayNo replay, broker tracks state per message, lower peak throughputTask queues, uneven job durations
ISR replicationf+1 copies tolerate f failuresNeeds a strongly consistent controller, latency follows the slowest ISR memberStorage cost matters
Majority quorum per partitionIgnores one slow replica, no external controller2f+1 copiesLatency-sensitive, smaller data
Local disks onlySimple, predictable latencyFleet sized by retentionShort retention
Tiered storageCheap long retention, faster broker replacementCold reads take tens of ms, more moving parts; production-ready only since Kafka 3.9 (Kafka 3.9 announcement)Replay over weeks or months
Idempotent consumerWorks with any sinkEvery team must implement itSide effects outside the log
Log transactionsExactly once inside the logExtra latency, only covers log-to-logStream processing jobs

Leadership lens

Check yourself

  1. The old leader appended offsets 100 to 105, but only 100 to 102 reached the follower that becomes the new leader, which then writes new data at 103. What happens when the old leader returns?

    Show answer

    It asks the new leader where its leader epoch ended and truncates its log to that point, discarding its own 103 to 105, before fetching again. Because consumers only read below the high watermark, which was never past 102 on the ISR, nobody ever saw the discarded records.

  2. Why does the design choose ISR replication over a majority quorum per partition, and what does it give up?

    Show answer

    ISR tolerates f failures with f+1 copies instead of 2f+1, which matters because copies are the expensive part at petabyte scale. The price is a strongly consistent controller to track the ISR and latency that follows the slowest in-sync replica until it is dropped.

  3. Using the lesson's numbers, why does disk rather than network set the broker count, and what lever shrinks it?

    Show answer

    Peak network is about 4.5 GB/s, which needs only about 4 brokers on 25 Gbit/s NICs kept below 40%. Seven days of retention at 65 TB a day across three replicas is about 455 TB, which at 30 TB per broker needs about 16, so the fleet is 18; tiered storage moves old segments to object storage and shrinks it.

  4. A consumer stuck in a 40-second GC pause loses its partitions and then wakes up and tries to commit offsets. What prevents it from overwriting the new owner's progress?

    Show answer

    Every rebalance increments the group's generation, and commits carry the generation. The woken consumer commits with generation 7 while the group is at 8, so the coordinator rejects it with ILLEGAL_GENERATION.

Sources and further study

Primary

Systems docs

Books

Other curricula

Watch

Free resources

Open in the interactive app: practice, quiz, mock interview →