Design a distributed message queue: a full worked answer

Assumptions: an internal platform used by many teams. Producers publish small messages (around 1 KB) to named topics. Several consumer groups read each topic independently, for example a search indexer, an analytics loader and a notifier all reading "orders". Messages are kept for seven days so a consumer can replay after a bug. Peak around a million messages a second across all topics.

The question that changes the storage design: is this a log or a work queue? In a work queue, each message goes to one consumer and is deleted once acknowledged, so the broker tracks every message's state. In a log, messages stay for the retention period and each consumer group just remembers a position (an offset). The log makes several independent consumers and replay nearly free, and it is what lets throughput reach the numbers above. I will design a log, and offer work-queue behaviour on top through consumer groups.

Second question: what ordering do producers need? A total order across a whole topic means one writer and one reader at a time, which caps throughput at one machine. What applications almost always need is order per key: all events for order 123 in sequence. I will promise ordering within a partition, and route messages by key, so per-key order holds.

Third: delivery guarantee. I will default to at-least-once (a message is never lost once confirmed, but a consumer may see it twice after a failure) and say plainly that consumers must be idempotent. Exactly-once processing inside the log is possible with more machinery; exactly-once effects on outside systems are the consumer's job.

Scope I propose: publish, storage and replication, consumer groups and offsets, retention, and failure handling. Out of scope: cross-region replication, schema management, delayed or priority messages, and a management UI.

Actors: producers, the broker that leads each partition, follower brokers holding replicas, consumer group members, and a small controller that decides which broker leads which partition. The controller is the actor that acts on its own: when a broker dies, it picks new leaders without anyone asking.

The chain: produced by a client → appended to the partition leader's log → copied by the in-sync followers → committed (every in-sync replica has it) → confirmed to the producer → fetched by a consumer → processed → offset committed by the consumer group → deleted when its segment ages past retention.

Failure branches. Producer times out after the leader wrote the message: it retries, and without protection the message is stored twice. The fix is an idempotent producer: each producer has an id and numbers its messages per partition, and the leader drops a sequence number it has already appended. Leader dies after confirming but before followers copied: if the producer asked for leader-only confirmation, that message is gone, which is why important topics confirm only after the in-sync replicas have it. Consumer crashes after processing but before committing its offset: the next owner of that partition starts from the last committed offset and processes the message again. That is at-least-once, and why consumers must be idempotent. Committing the offset before processing flips it to at-most-once: a crash loses the message instead.

The branch the system owns: a consumer group that falls behind further than the retention period. Retention deletes the oldest segments on schedule, whether or not anyone read them, so that group silently skips data. The system must alert on lag measured against retention, long before it gets there. Another branch: a poison message that crashes the consumer every time. With a log it blocks its whole partition for that group, so after a few attempts it goes to a dead-letter topic and the offset moves on.

Peak 1 million messages a second at about 1 KB is about 1 GB/s coming in. With three replicas, the cluster writes 3 GB/s to disk and moves 2 GB/s between brokers for replication. If three consumer groups read everything, that is another 3 GB/s going out. Network, not CPU, is the first thing to size.

Storage uses the average, not the peak. Say the average is a third of peak, about 330 MB/s. A day is roughly 100,000 seconds, so about 33 TB a day, 230 TB for seven days, and about 700 TB with three replicas. At around 20 TB of usable disk per broker that is 35 or more brokers for storage alone, while throughput alone might need a dozen. So retention decides the cluster size.

Partitions set the parallelism. Within a consumer group, each partition is read by one member at a time, so a group can never use more consumers than there are partitions. If one consumer can process 5,000 messages a second and the busiest topic peaks at 500,000 a second, that topic needs at least 100 partitions, and more for headroom, because adding partitions later moves which partition each key maps to, which breaks per-key order for a while.

Writes are sequential appends and reads are mostly sequential scans of recent data that is still in the operating system's page cache. A single disk that manages a few hundred random operations a second can sustain hundreds of MB/s sequentially. The design depends on keeping both paths sequential.

So the hard part here is replication and retention at disk and network speed, while keeping per-key order. Moving bytes is the workload; the design is about never touching them more than necessary.

Topics split into partitions, each an ordered, append-only log. A partition is the unit of ordering, replication and consumer parallelism, which is exactly what Stage 1 and Stage 3 asked for. On disk, a partition is a series of segment files (say 1 GB each) plus a small index from offset to file position. Appending is a write to the end of the active segment; retention deletes whole old segments, which is cheap.

Producers use a client library that batches messages per partition for a few milliseconds, compresses each batch, and picks the partition by hashing the key (or spreads keyless messages evenly). Batching is what makes a million small messages a second affordable: the broker handles thousands of batches, not a million requests.

Brokers each hold many partition replicas. For each partition one replica is the leader, which takes all writes and, by default, all reads. Followers fetch from the leader like consumers do and append the same bytes. Leaders are spread across brokers so the load is too. Fetches are served largely from the page cache, and the bytes can go from file to socket without being copied through the application.

A controller backed by a small consensus group (three or five nodes running Raft) stores the metadata: topics, partitions, which broker leads each one, and which replicas are in sync. It is the only place that decides a new leader. Message traffic never goes through it; it handles only metadata, which is what consensus is good at.

Consumer groups. A coordinator assigns each partition of a topic to exactly one member of the group, and reassigns when members join, leave or stop sending heartbeats. Each group commits its offsets per partition into an internal, compacted topic, so positions are themselves replicated and durable without a separate database.

Retention and compaction. Time- or size-based retention deletes old segments. For topics that represent the latest state per key (a user's current address), log compaction keeps only the newest message per key instead, which is also how the offsets topic stays small.

Hard part one: replication and leader failover. The leader keeps a list of in-sync replicas (ISR): followers that have caught up within a time limit. A message is committed when every replica in the ISR has it, and the leader tracks this as a high watermark; consumers only see messages below it. A producer that asks for confirmation from all in-sync replicas, on a topic configured to require at least two of them, gets a confirmation that survives the loss of any one broker. When the leader dies, the controller picks a new leader from the ISR, so no committed message is lost, and bumps a leader epoch so the old leader, if it comes back, is fenced and truncates anything it wrote that was never committed. Cost: every publish waits for the slowest in-sync follower, and if the ISR shrinks below the minimum, the partition refuses writes rather than accept them unsafely.

The setting interviewers probe here: what if every in-sync replica is gone? Waiting for one to return keeps every committed message and leaves the partition unavailable until then. Electing an out-of-sync replica restores service at once and silently discards whatever it was missing. That is the CAP trade-off as a single configuration flag, and the right answer differs for a click stream and a payments topic.

Hard part two: delivery semantics. At-most-once: commit the offset, then process; a crash loses the message. At-least-once: process, then commit; a crash reprocesses it. Exactly-once within the log: the idempotent producer removes duplicate writes from retries, and transactions let a consumer write its output messages and its input offset atomically, so a read-process-write pipeline inside the queue neither loses nor repeats. Once a consumer touches the outside world (charges a card, sends an email), the queue cannot make that exactly-once. The consumer has to: store an idempotency key with the side effect, or record the processed offset in the same database transaction as its result. Cost: transactions add latency and coordination, so they are opt-in per producer.

Rebalancing and hot partitions. When a member joins or leaves, partitions move between members. A simple protocol pauses the whole group while it reassigns, which shows up as a lag spike on every deploy; an incremental protocol moves only the partitions that need to move. A key that is far hotter than others (one huge customer) overloads its partition and its consumer, and adding partitions does not help, because one key always maps to one partition. Options are to split that key with a suffix where per-key order is not needed, or give that customer its own topic. Cost: either breaks strict ordering for that key, or adds operational special cases.

What to watch. Consumer lag per group and partition, and lag compared with retention; ISR shrink events and partitions below minimum in-sync replicas; publish latency at p99; disk usage against retention; and controller leader changes. Lag is the number the product teams feel.