Design a Distributed Message Queue

Dozens of services publish events: orders placed, payments taken, pages viewed. Dozens more read them at their own pace, and some want to reread last Tuesday's events after fixing a bug. You are asked to build the broker in the middle: a cluster that accepts a hundred thousand messages a second, keeps them for a week, and does not lose an acknowledged message when a broker dies. It is the kind of product people mean when they say Kafka.

Interviewers ask this to see what you know under the abstraction. How to use a queue (queue versus log, consumer groups, backpressure) is in Message Queues and Streams. This lesson builds the log itself: files on disk, replicas, offsets, and the exact meaning of "acknowledged". The steps follow A Framework for Any Design.

#Clarify the problem

• Functional: produce messages to a topic, with an optional key; consume in groups, where each group gets every message and its members share the work; commit and rewind positions; delete data after a retention period, or keep only the latest value per key.
• Non-functional: no acknowledged message is lost when one broker dies; order is kept per key; produce p99 under about 10 ms; 300,000 messages a second at peak; 7 days of retention; consumers can be hours behind without slowing producers.
• Out of scope: delayed and priority messages, per-message acknowledgement, replication between regions.

#Estimate the scale

• Ingest: 100,000 × 1 KB = 100 MB/s on average, 300 MB/s at peak.
• Per day: 100 MB/s × 86,400 s ≈ 8.6 TB.
• Reads: three groups read everything, so 300 MB/s out on average and 900 MB/s at peak, plus replica traffic.

Check the disks of 20 brokers at peak: 300 MB/s × 3 copies ÷ 20 = 45 MB/s of writes each, and 900 MB/s ÷ 20 = 45 MB/s of consumer reads. Comfortable for sequential I/O.

#Define the API

Metadata(topics)
  -> per partition: leader broker, replicas, in-sync replicas
 
Produce(topic, partition, batch, acks = 0 | 1 | all,
        producer_id, sequence)
  OK base_offset=5000
  NOT_LEADER                 leadership moved; refresh metadata, retry
  NOT_ENOUGH_REPLICAS        fewer in-sync copies than required
  DUPLICATE_SEQUENCE         already written; returns the first offset
 
Fetch(topic, partition, offset, max_bytes, max_wait_ms)
  records from offset up to the high watermark
  OFFSET_OUT_OF_RANGE        that data was deleted by retention
 
JoinGroup(group, member) / Heartbeat(group, member, generation)
  -> assigned partitions, generation=12
 
CommitOffset(group, generation, partition, offset)
  OK
  ILLEGAL_GENERATION         you were removed from the group; rejoin

The decision worth arguing is pull versus push. Consumers pull: each one asks for the next bytes from its own offset, at its own pace. A slow consumer then falls behind without slowing anyone, a consumer can batch large reads, and rewinding is just asking for an older offset. To avoid busy polling, fetch waits up to max_wait_ms for data before answering empty.

#Model the data

orders-7/                             partition 7 of topic "orders"
  00000000000004000000.log            closed segment, about 1 GB
  00000000000004000000.index          offset -> byte position, every 4 KB
  00000000000005000000.log            active segment: appends go here
  00000000000005000000.index
 
record batch (as sent, as stored, as served)
  base_offset, producer_id, sequence, crc
  records: offset_delta, timestamp, key, value, headers
 
cluster metadata                      held by the controller
  topic -> partitions
  partition -> replicas [1, 2, 3], leader 1, leader_epoch 4,
               in_sync [1, 2, 3]
 
consumer offsets                      an internal topic, compacted
  (group, topic, partition) -> next offset to read

There is no database here. The broker's own files are the store, and the only access patterns are "append at the end" and "read forward from offset X". Consumer positions are themselves messages, stored in an internal topic that keeps only the latest value per key. The small, critical metadata lives in a controller replicated with consensus.

#High-level design

Every partition has one leader and two followers on different brokers. Each broker leads about a third of its partitions. The picture shows one partition.

Step through one message with acks=all, a consumer reading it, then the leader dying in the middle of a retry.

#Deep dives

#The log on disk

A random 1 KB write on a spinning disk costs a seek of about 10 ms, so 100 writes a second, or 100 KB/s. Appending to the end of a file, the same disk streams about 200 MB/s, 2,000 times more. SSDs narrow the gap, but large sequential writes still mean fewer, bigger operations. So the broker only appends: no updates in place, no per-message index.

• Batching. Producers group messages for a few milliseconds, say 16 KB at a time. At 100,000 messages a second that is about 6,250 requests instead of 100,000, and the batch is compressed as a unit.
• The page cache. The broker writes to the operating system's file cache and does not force each write to disk. Consumers that keep up read data that is still in memory, so most reads never touch the disk.
• Zero copy. Because stored bytes equal sent bytes, the broker can ask the kernel to send a file range straight to the socket without copying it through the program.

Segments and the index. A partition is split into segment files of about 1 GB, each named after its first offset. To serve "read from offset 4,712,345", the broker picks the segment by name, then uses a sparse index with one entry per 4 KB of log: a 1 GB segment has about 250,000 entries, 2 MB, kept in memory. Binary search finds the nearest entry in about 18 steps, then the broker scans forward less than 4 KB.

Retention and compaction. Deleting old data means deleting whole closed segments: a file delete, not a rewrite. A segment goes once its newest record is older than 7 days. For topics where only the latest value per key matters, such as "current address of user 42", use compaction instead: a background job rewrites old segments keeping only the last record for each key. A record with an empty value, a tombstone, deletes the key, and is itself removed after a grace period so slow readers still see the delete. The consumer offsets topic is compacted this way.

#Replication and what "acknowledged" means

Followers fetch from the leader like consumers. The leader tracks which followers are caught up, within say 10 seconds: the in-sync replicas (ISR). The high watermark is the highest offset every ISR member has, and only records below it are visible to consumers. The producer chooses how long to wait with acks:

Be exact about the guarantee. With 3 replicas, acks=all and a minimum ISR of 2, an acknowledged record is in the log of at least 2 brokers. "In the log" means the page cache, not necessarily the disk: durability comes from copies failing independently, so put the 3 replicas in different racks or zones. The minimum matters: without it, the ISR could shrink to the leader alone, and acks=all would quietly mean acks=1.

The controller. A quorum of 3 controller nodes keeps the metadata in a consensus-replicated log. It notices dead brokers by missed heartbeats, picks a new leader from the ISR, and records every ISR change. Each new leader gets a higher leader epoch, and brokers reject requests carrying an older one. This closes the zombie-leader race: an old leader cut off from the controller cannot remove followers from the ISR, so it cannot acknowledge an acks=all write alone. With acks=1 it can, and those writes are cut off when it rejoins.

If every ISR member dies, either wait for one to return (unavailable, no loss) or let an out-of-sync replica lead (available, acknowledged records lost). Wait, by default.

#Ordering and producing without duplicates

Order is per partition, and the partition is chosen by hashing the key, so all of order-881's events stay in order. Two retries can break even that: batch 42 fails, batch 43 succeeds, then 42 is retried and lands after 43. A retry after a lost acknowledgement, as in the tracer, writes a duplicate.

The idempotent producer fixes both. Each producer gets an id, and numbers its batches per partition. The broker remembers the last few sequence numbers per producer: a repeat is answered with the original offset, and a gap (43 before 42) is rejected so the producer resends in order. This covers retries within one producer session: a restarted producer gets a new id, so its resends slip through unless it keeps a stable id. The cost is a little state per producer on each broker (the general idea is in Idempotency and Retries).

#Consumer groups, offsets and delivery

One broker acts as the group coordinator. Members send heartbeats; when one joins, leaves or misses heartbeats, the group rebalances and partitions are reassigned. Each assignment has a generation number, and offset commits from an old generation are rejected, so a consumer that paused for a minute cannot overwrite the position of the one that took its partitions.

The simplest rebalance stops every member and reassigns everything, pausing all 32 partitions to move one. A cooperative rebalance pauses only the partitions that change owner. Prefer it.

Delivery depends on when the offset is committed. Commit after processing for at-least-once: if offsets are committed every 5 seconds and a consumer dies, up to 5 × 2,500 = 12,500 of its messages are processed again. For exactly-once within the log, a consumer that reads, transforms and writes to another topic can commit its output and its offsets in one transaction; readers that only read committed data never see a half-done result. That promise ends at the log: an email or a database write outside it still needs idempotent processing.

#Bottlenecks and failures

At 10x, 1 GB/s in and 1.8 PB for a week, disks run out first: 1.8 PB ÷ 12 TB = 150 brokers, about 200 with headroom. Move closed segments older than a day to cheaper object storage and keep only recent ones on local disk. Partition count grows too, and a failed broker leading thousands of partitions takes longer to fail over. Hot keys stay hot partitions; more brokers do not split them. A consumer replaying last week reads from disk and evicts recent data from the page cache, slowing readers that were keeping up: cap replay bandwidth per client.

#What a strong answer sounds like

#Follow-up questions

#Why not use a database table as the queue?

Every consume becomes an update or delete, indexes grow and shrink, and many consumers fight over the same rows. A log only appends, and each group's position is one number per partition.

#How do you add partitions to a busy topic?

Adding partitions changes hash(key) mod n, so a key's new events land in a different partition from its old ones, and order breaks for that moment. Pick the count for future load, or create a new topic and migrate consumers.

#How would you keep data for a year?

Tier it: recent segments on broker disks, older ones in object storage, fetched on demand. Old reads are slower; brokers stay small.

#How do you support many small topics?

Each partition costs open files, index memory and metadata. Share partitions across tenants with a key prefix, or cap partitions per broker.

#How would you run it in two regions?

Keep one cluster per region and copy chosen topics asynchronously. Offsets differ between clusters, so map them by timestamp when a consumer fails over, and expect some duplicates.