Design a Distributed Message Queue

Your company runs hundreds of services that need to hand events to each other: orders placed, payments settled, profile changes, click logs. Teams keep wiring point-to-point calls and losing data whenever a downstream service is slow or down.

Design a shared, durable message log. Producers append messages to named topics; many independent consumer teams read those topics at their own pace, and some need to replay the last week of data after a bug fix. Peak load is around two million messages per second across the company, and losing an acknowledged message is treated as an incident.

In 45 minutes, settle the guarantees (ordering, delivery semantics, retention), size the cluster, then go deep on how partitions are replicated, how consumer groups share work, and what happens when a broker dies mid-write.

Sketch on the HLD board, write your answer in the Design doc, and size it with the Estimator. Start the mock interview from the timer for a 45-minute run.

Design Path · Data & Infrastructure · problem 12 of 18

Ask before you design. Try answering each yourself, then compare.

1. Is this a work queue where a message disappears once handled, or a log that many consumers read independently?
   Typical answer

   A log. Messages are retained for a time window and each consumer group tracks its own position, so new consumers can replay history.

2. What ordering do consumers need?
   Typical answer

   Order per key (for example per order ID) is required. Global order across the whole topic is not.

3. What delivery guarantee is expected?
   Typical answer

   At-least-once by default, with no acknowledged message ever lost. Effectively-once processing is a stretch goal using idempotent producers and transactional offset commits.

4. What scale and message size?
   Typical answer

   About 1M messages per second on average, 2M at peak, averaging 1 KB, with 7 days of retention by default.

5. What latency do producers and consumers expect?
   Typical answer

   Produce acknowledgement under 10 ms at p99 within a region; end-to-end delivery under 100 ms when consumers keep up.

6. Single region or multi-region?
   Typical answer

   One region with three availability zones. Cross-region mirroring is a follow-up.

7. Do we need per-message routing rules, priorities or delayed delivery?
   Typical answer

   No. Keep the core a simple append-only log; those features can be layered on later.

Requirements

Functional

• Create topics with a chosen number of partitions, a replication factor and a retention period.
• Producers append messages (optionally keyed) and get an acknowledgement with partition and offset.
• Consumer groups read topics, with partitions spread across group members and rebalanced on join or leave.
• Consumers commit offsets and can rewind to an earlier offset or timestamp to replay data.

Non-functional

• 2M messages/s peak (about 2 GB/s in), 7-day default retention.
• No loss of acknowledged messages if any single broker or a whole availability zone fails.
• Produce p99 under 10 ms; end-to-end p99 under 100 ms for caught-up consumers.
• 99.99% availability for produce on any topic with replication factor 3.

Out of scope

• Per-message priorities, delayed delivery and complex routing rules.
• Cross-region replication (discuss as a follow-up).
• Schema registry and stream processing frameworks built on top.

Back-of-the-envelope estimates

• Ingest: 1M msgs/s average × 1 KB ≈ 1 GB/s, 2 GB/s at peak. That is about 86 B messages and 86 TB per day.
• Replication: factor 3 means every byte is written to three disks: 3 GB/s average, 6 GB/s peak of disk writes across the cluster, plus 2 GB/s of broker-to-broker traffic.
• Reads: three consumer groups per topic on average ≈ 3 GB/s out, 6 GB/s at peak. Caught-up consumers are served from the page cache, not disk.
• Storage: 86 TB/day × 7 days × 3 copies ≈ 1.8 PB. With 24 TB of usable disk per broker that is ~75 brokers before headroom; at a sustained 150 MB/s of writes per broker, throughput alone needs ~40.
• Partitions: if one consumer handles ~5 MB/s, the busiest topics need a few hundred partitions to keep up.

Retention storage, not CPU, sets the broker count. That is the main argument for tiering old segments to object storage, which shrinks local disks to a day or two of data.

API

CreateTopic(name, partitions, replication=3, retention=7d, min_insync=2)
Produce(topic, [ {key?, value, headers} ], acks=all, producerId, epoch, seq)
      -> [ {partition, offset} ]
Fetch(topic, partition, offset, maxBytes, maxWaitMs) -> records[], highWatermark
JoinGroup(group, member, topics) -> generation, assignment
CommitOffsets(group, generation, {partition: offset})
ListOffsets(topic, partition, timestamp) -> offset   // for replay

Fetch is a long poll by offset, so the broker keeps no per-message delivery state: a consumer's progress is just one integer per partition, which makes replay and many independent readers nearly free.

Data model

Partition log: each partition is an ordered sequence of segment files (for example 1 GB each) named by their first offset, plus a sparse offset index and a time index per segment. Records are appended in producer batches and never modified; retention deletes whole old segments. Segments older than a day or two move to object storage.

Cluster metadata: topics, partition-to-broker assignment, the current leader, the leader epoch and the in-sync replica (ISR) set for each partition. It is small but must be strongly consistent, so it lives in a Raft-replicated controller quorum.

Consumer offsets: (group, topic, partition) -> offset, metadata, stored in an internal compacted topic so only the latest commit per key survives. Partitioned by group, so one coordinator broker owns each group.

High-level design

Write path. A producer first asks any broker for metadata and learns which broker leads each partition. It hashes the message key to pick a partition (or round-robins if there is no key), batches messages for a few milliseconds, compresses the batch and sends it straight to that partition's leader. The leader appends the batch to the active segment file, relying on the page cache rather than an fsync per write. Follower brokers in other zones continuously pull from the leader. Once every member of the in-sync replica set has the batch, the leader advances the high watermark and acknowledges the producer with the offsets.

Read path. Consumers in a group join through a group coordinator, which hands each member a set of partitions. Each member long-polls the partition leader with “give me records from offset N” and receives data only up to the high watermark, so it never sees a write that could still be lost. Recent data is served from memory with zero-copy transfer; older data is read from local segments or fetched from tiered storage. After processing, the consumer commits its next offset to the compacted offsets topic.

Control plane. The controller quorum detects dead brokers, picks new leaders from the ISR and bumps the leader epoch. A lag monitor compares log end offsets with committed offsets and alerts when a group falls behind.

Open the HLD board and press Reference to see the architecture diagram.

Deep dives

Replication, ISR and when to acknowledge

Each partition has one leader and, with factor 3, two followers in other zones. Followers pull from the leader, and the leader keeps an in-sync set of followers that are within a few seconds of its log end. A write with acks=all is acknowledged once every ISR member has it; with min.insync=2, the leader rejects writes when it is the only copy left, choosing unavailability over silent loss. Batches are not fsynced individually: durability comes from copies in independent zones rather than from a single disk. The high watermark marks the committed prefix, and consumers never read past it. Shrinking the ISR when a follower lags keeps one slow disk from stalling every producer.

Leader failover without losing or duplicating data

When a broker dies, the controller picks a new leader for each of its partitions, only from the ISR, so the new leader is guaranteed to hold every acknowledged record. It increments the leader epoch and records the offset where that epoch begins. A returning old leader may hold a tail that was never committed; it asks the new leader where its own last epoch ended and truncates to that point before replicating again. Allowing an out-of-sync replica to lead (unclean election) restores availability faster but can drop acknowledged writes, so it should be an explicit per-topic choice.

Consumer groups and rebalancing

Within a group, each partition is assigned to exactly one member, so per-key order is preserved and parallelism is capped by the partition count. Members heartbeat to a coordinator; if one goes silent or a new one joins, the coordinator starts a new generation and reassigns partitions. Stopping the whole group on every change causes long pauses for big groups, so use incremental (cooperative) rebalancing that only moves the partitions that must move, plus sticky assignment so members keep their warm state. Commits carry the generation number, letting the coordinator reject offsets from a zombie member that lost its partitions.

From at-least-once to effectively-once

Retries make duplicates inevitable: a producer times out, resends, and the first attempt had in fact landed. Give each producer an ID and a per-partition sequence number; the leader drops any batch whose sequence it has already written, which makes retries idempotent. For read-process-write pipelines, a transaction can write output records and the consumer's offset commit atomically, and downstream readers skip aborted data. Side effects outside the log, such as charging a card or sending an email, still need an idempotency key at the receiver, which the design should state plainly.

Failure modes

• If a broker dies, the controller moves leadership of its partitions to in-sync followers within seconds; producers refresh metadata on the error and retry, and idempotent sequence numbers stop the retries from duplicating data.
• If a whole availability zone fails, each partition still has two of three copies; with min.insync=2 writes continue, and the lost replicas are rebuilt elsewhere at a throttled rate so recovery traffic does not starve clients.
• If a consumer is slow or stuck, its lag grows but nothing else is affected because the log is retained independently; alert on lag in time, not just in messages, and warn before retention deletes unread data.
• If the controller quorum loses its leader, a new one is elected in a second or two; data reads and writes keep flowing to existing leaders, only topology changes pause.
• If a follower's disk slows down, it drops out of the ISR instead of stalling producers, and is re-added once it catches up.

Core algorithm in code

import hashlib


def partition_for(key: str, n: int) -> int:
    """Same key -> same partition, so per-key order is kept."""
    return int(hashlib.md5(key.encode()).hexdigest(), 16) % n


class Partition:
    """A leader log plus follower copies. Consumers only see up to the high watermark."""

    def __init__(self, replicas=3, min_isr=2):
        self.log = []                                   # leader's log: (key, value)
        self.follower_end = {f"r{i}": 0 for i in range(1, replicas)}
        self.isr = set(self.follower_end)               # in-sync followers
        self.min_isr = min_isr

    def append(self, key, value):
        if 1 + len(self.isr) < self.min_isr:
            raise RuntimeError("NOT_ENOUGH_REPLICAS: refusing write")
        self.log.append((key, value))
        return len(self.log) - 1                        # offset

    def follower_fetch(self, follower, max_records=10):
        # followers pull; their fetch position tells the leader how far they got
        end = min(len(self.log), self.follower_end[follower] + max_records)
        self.follower_end[follower] = end

    def mark_lagging(self, follower):
        self.isr.discard(follower)

    def high_watermark(self):
        ends = [len(self.log)] + [self.follower_end[f] for f in self.isr]
        return min(ends)                                # committed = on every in-sync copy

    def read(self, offset, max_records=10):
        hw = self.high_watermark()
        return [(o, *self.log[o]) for o in range(offset, min(hw, offset + max_records))]


class ConsumerGroup:
    def __init__(self, partitions):
        self.partitions = partitions
        self.committed = {p: 0 for p in range(len(partitions))}

    def poll_and_process(self, pid, handler):
        batch = self.partitions[pid].read(self.committed[pid])
        for off, key, value in batch:
            handler(pid, off, key, value)               # side effects first...
        if batch:
            self.committed[pid] = batch[-1][0] + 1      # ...then commit: at-least-once
        return len(batch)


parts = [Partition() for _ in range(3)]
for i, (user, event) in enumerate([("alice", "login"), ("bob", "click"), ("alice", "buy"),
                                    ("carol", "login"), ("bob", "logout"), ("alice", "logout")]):
    p = partition_for(user, len(parts))
    off = parts[p].append(user, event)
    print(f"produce {user:<5} {event:<6} -> partition {p} offset {off}")

print("before replication, high watermarks:", [p.high_watermark() for p in parts])
for p in parts:
    p.follower_fetch("r1")
print("r1 caught up, r2 not yet:            ", [p.high_watermark() for p in parts])
for p in parts:
    p.mark_lagging("r2")                                # r2 too slow: dropped from ISR
print("r2 removed from ISR:                  ", [p.high_watermark() for p in parts])

group = ConsumerGroup(parts)
seen = []
for pid in range(len(parts)):
    group.poll_and_process(pid, lambda pid, off, k, v: seen.append(f"p{pid}@{off}:{k}/{v}"))
print("consumed:", seen)
print("committed offsets:", group.committed)

parts[0].mark_lagging("r1")
try:
    parts[0].append("dave", "login")
except RuntimeError as e:
    print("with only the leader left in sync:", e)

Trade-offs

• acks=all with min.insync=2 gives durability through a zone loss but costs a cross-zone round trip on every produce; acks=1 is faster but loses the tail when a leader dies.
• More partitions give more consumer parallelism but cost more open files, slower failover and longer rebalances, and you cannot reduce them later without breaking key order.
• Relying on replication instead of per-write fsync gives high throughput but means a simultaneous power loss in all zones could lose recent writes.
• Pull-based consumers make backpressure and replay simple but add a little latency compared with pushing; long polling closes most of that gap.
• Tiered storage makes long retention cheap but makes old-data reads slower and adds a dependency on object storage.

What a strong answer shows

• Models the system as a partitioned, append-only log and explains why sequential I/O and the page cache make it fast.
• Defines exactly when a write is acknowledged and what the high watermark protects.
• Explains per-key ordering through key-based partitioning and one consumer per partition per group.
• Handles leader failover from the in-sync set and truncation of uncommitted tails.
• Is honest about delivery semantics: at-least-once by default, idempotent producers and idempotent consumers for effectively-once.
• Sizes the cluster from retention storage and replication bandwidth, not just message rate.

Mistakes to avoid

• Promising global ordering across a whole topic, which forces a single partition and kills throughput.
• Acknowledging a write once the leader has it, then losing it when the leader dies before followers copy it.
• Tracking per-message acknowledgements on the broker, which turns a cheap log into an expensive database.
• Claiming exactly-once delivery end to end without explaining idempotency at the consumer's side effects.
• Ignoring hot keys: one huge customer hashed to one partition caps that customer's throughput.

Follow-ups to expect

• How would you mirror topics to a second region? (Asynchronous replication per partition, offsets differ, so translate consumer offsets or use timestamps on failover.)
• How would you add delayed or scheduled messages? (Separate delay topics per time bucket, or a timer service that re-produces messages when due.)
• How would you handle a single hot key that exceeds one partition? (Salt the key into sub-keys and merge downstream, accepting weaker ordering.)
• How do you support key-value style topics that keep only the latest value per key? (Log compaction in the background.)
• How would you enforce quotas between noisy tenants? (Per-client byte-rate quotas on brokers, throttling by delaying responses.)

Practice next