Design a distributed message queue

Build the thing most other designs on this site lean on: a durable, partitioned log that thousands of producers write to and many independent consumers read from, at millions of messages a second. Think Apache Kafka, Amazon Kinesis or Pulsar. The core idea is one sentence; the interview is about what happens when brokers fail, consumers are slow, and one key gets all the traffic.

1. The prompt

"Design a distributed message queue that services use to publish events and consume them, durably and at high throughput."

What makes it hard:

• "Durable" and "fast" pull against each other: every copy to another machine costs time.
• Consumers come and go, crash mid-batch and fall behind, and the system must neither lose nor endlessly redeliver.
• Ordering is expected, but global ordering doesn't scale.

2. Frame the problem

3. What it must do

1. A producer can publish a message (key, value) to a topic and learn when it's durably stored.
2. A consumer group can read every message of a topic in order per key, sharing the work across its members.
3. Messages are kept for the retention period, so a group can replay from any position.

Not now: per-message routing rules and priorities (a work queue's job), delayed delivery, schema registry, cross-region replication (a follow-up).

4. How well it must do it

5. Numbers that change the design

Illustrative numbers. Skipped: the number of topics (metadata only) and producer count (connections are cheap).

6. The nouns

• Topic, split into partitions: each an append-only, ordered log.
• Message: key, value, headers, timestamp, and its offset (position) in the partition.
• Broker: a server holding partition replicas; one replica of each partition is the leader.
• Consumer group and its committed offset per partition.
• Controller: the metadata authority (which broker leads which partition).

partition      topic, id, leader_broker, replicas[], in_sync_replicas[], leader_epoch
segment files  partition/00000000000004180000.log  + .index (offset → file position) + .timeindex
message        offset, timestamp, key, value, headers             (inside compressed batches)
offsets        (group, topic, partition) → committed_offset       (itself stored in an internal, compacted topic)

7. The contract

Binary protocol over long-lived TCP (not REST: per-message HTTP is too costly)
Produce   {topic, partition (or key), batch[{key, value, headers}], acks: 0 | 1 | all}
  ← {base_offset} | error (NOT_LEADER → refresh metadata and retry)
Fetch     {topic, partition, from_offset, max_bytes, max_wait_ms}    (long poll)
  ← batches
CommitOffset  {group, topic, partition, offset}
JoinGroup / Heartbeat / LeaveGroup                                    (group membership)
Metadata  {topics} ← partition → leader map

Producers send to the partition leader directly; the key's hash picks the partition. A producer id plus a sequence number per partition lets the broker drop duplicate retries (idempotent producer).

Data flow: producer batches by partition → leader appends to its log → followers fetch and append → leader acknowledges per acks → consumers fetch from their committed offset → process → commit the next offset.

8. First working version

Brokers that append to partition logs on disk, a controller that knows the leaders, producers that send to leaders, consumer groups that pull.

8.1 Publishing

An append is a sequential write to the end of the current segment file: the operating system's page cache absorbs it and the disk sees large sequential writes, which is why one broker can take hundreds of MB a second.

8.2 Consuming as a group

Partitions are divided among the group's members; each partition has exactly one reader in the group, so order per partition is kept. The consumer commits the next offset after processing: if it crashes before committing, the next owner re-reads from the last commit (at least once).

8.3 Keeping and deleting

Retention deletes whole old segment files, never single messages: cheap and fast. Consumers can rewind to any retained offset to replay.

9. Make it hold up

9.1 Replication and acknowledgements durability · availability

Try the trade-off in the lab in section 13: crash the leader after message 7 with each setting.

9.2 Who decides who leads availability · consistency

A controller holds the cluster metadata: partitions, replicas, ISR, leaders. It must itself be fault tolerant and agree on one truth: Kafka moved this from ZooKeeper to a built-in Raft quorum of controllers (KRaft). When a broker dies, the controller picks a new leader from the ISR for each of its partitions and bumps the partition's leader epoch; brokers and clients reject requests carrying an old epoch, so a returning old leader can't accept writes (fencing, Module 21). Rule to state: an out-of-sync replica must never become leader unless you explicitly accept data loss ("unclean" election).

9.3 Consumer groups and rebalancing no skipped messages

9.4 A hot key throughput · latency

All messages with one key go to one partition, so a celebrity user or a huge customer can overload one partition while others idle. Options: spread that key over several partitions with a suffix (losing per-key order unless you re-merge), give big tenants their own topics, or rely on the producer's partitioner only for keys that truly need order and use round-robin for the rest. Watch per-partition bytes in, not just the topic total.

9.5 Duplicates and exactly once consistency

9.6 A petabyte of retention cost

Keep recent segments on brokers' local disks for fast reads, and move older ones to object storage (D35); brokers fetch them back when a consumer replays history. Brokers become lighter, so adding or replacing one moves little data. For "latest value per key" topics (account settings, a table's changelog), use log compaction: old values for a key are removed in the background, keeping only the latest.

9.7 The whole design after the deep dives

10. Interviewer follow-ups

11. When things break

12. Wrap-up and the clock

Trade-offs we made:

• A log, not a queue: replay and many readers, at the cost of storage.
• Order per partition only.
• Acknowledgement level chosen per producer: safety or latency.
• At least once by default; exactly once only inside the log.
• Tiered storage: cheap history, slower replays.

What we'd watch: bytes in and out per broker and partition · under-replicated partitions · ISR shrink/expand rate · produce and fetch latency p99 · consumer lag per group · rebalance count and duration.

At 10×: more brokers and partitions, client quotas, tiered storage everywhere; the protocol and design stay the same.

Common mistakes:

• Deleting messages when consumed (that's a work queue, not a log).
• Promising global ordering.
• Acknowledging before replication and calling it durable.
• Committing offsets before processing.
• Ignoring rebalances and hot keys.

13. Try it, check yourself, read more

Acknowledgement lab. A producer sends messages 1 to 10 to a partition with one leader and two followers. Followers copy with a delay of two messages. The leader crashes right after message 7. Which acknowledged messages survive the failover?

Model: the follower that becomes leader is the most up to date one; messages it doesn't have are gone. Outlined = acknowledged to the producer. Illustrative.