Design a Distributed Message Queue

A distributed message queue lets producers send messages that are durably stored and later consumed by one or more consumers, decoupling the two in time and rate. The design problem is achieving high throughput, durability, and ordering on a cluster of machines that fail. This article designs a log-based broker, the model used by Kafka, and notes where a traditional queue model differs. For the concepts, see message queues vs pub/sub and Kafka internals.

Requirements

Functional

• Producers publish messages to a named topic.
• Consumers subscribe to topics and read messages. Multiple independent consumer groups can each read all messages, and within a group each message is processed by one consumer.
• Messages are retained for a configurable period and can be replayed from a chosen position.
• Ordering: messages with the same key are delivered in the order they were produced.

Non-functional

• High throughput: millions of messages per second across the cluster.
• Low latency: end-to-end in single or low double-digit milliseconds under normal load.
• Durability: acknowledged messages survive broker failures.
• Availability: producers and consumers continue when a broker fails.
• Horizontal scalability by adding brokers.
• Configurable delivery semantics: at-most-once, at-least-once, and effectively-once.

Out of scope: complex routing, message transformation, per-message priorities, delayed delivery (mentioned in follow-ups).

Estimation

Assumptions: 1 million messages/s peak, 1 KB average message, retention of 7 days, replication factor 3.

• Ingress: 1M x 1 KB = 1 GB/s.

• Storage per day: 1 GB/s x 86,400 s is about 86 TB/day. With 7-day retention, about 600 TB. With replication factor 3, about 1.8 PB.

• Replication traffic: each message is copied to 2 followers, so 2 GB/s of inter-broker traffic.

• Egress: with 3 consumer groups reading everything, about 3 GB/s (reads of recent data come from the page cache).

• Brokers: assume a broker sustains 100 to 200 MB/s of ingress. Leader ingress of 1 GB/s then needs about 5 to 10 brokers, and follower fetch traffic adds load on the same brokers, so more are needed. Storage sets the count: 1.8 PB over 30 brokers is 60 TB each, and fewer, larger disks per broker are possible only at the cost of longer recovery after a failure.

• Partitions: a partition is consumed by one consumer per group, so partition count bounds consumer parallelism. If a single consumer processes 10 MB/s, then 1 GB/s needs at least 100 partitions, and more for headroom.

• Per-broker load: with 30 brokers and 1,000 partitions x 3 replicas = 3,000 replicas, each broker hosts about 100 replicas and leads about 33 partitions, so each leader handles about 1 GB/s / 30 = 33 MB/s of ingress plus 67 MB/s of follower fetch and about 100 MB/s of consumer fetch. That is comfortable, and shows the cluster is storage-bound, not throughput-bound.

• Page cache: consumers keeping up read data written seconds ago. A broker with 64 GB of RAM for cache holds about 64 GB / 33 MB/s = 32 minutes of its leader data, so a consumer lagging by more than about 30 minutes reads from disk and competes with writes.

• Recovery: re-replicating 60 TB at a throttled 200 MB/s takes about 3.5 days, which is the exposure window after a disk or broker loss; this is what argues against very dense brokers.

What the numbers imply: sizing follows disk capacity and recovery time, not CPU. Sequential disk writes and the operating system page cache make these rates feasible on commodity hardware, which is central to the design. Partition count is set by consumer parallelism (at least 100, choose about 1,000 for headroom, or about 33 per broker), and the inter-broker traffic of 2 GB/s means the network needs 25 Gbit or faster NICs on brokers.

API design

Producer:

produce(topic, key?, value, headers?, acks)    -> { partition, offset }
produce_batch(topic, [messages], acks)         -> [ { partition, offset } ]

Consumer:

subscribe(topic, groupId)
poll(maxMessages, timeout)                     -> [ { topic, partition, offset, key, value } ]
commit(offsets)                                -> ok
seek(topic, partition, offset)

Admin:

create_topic(name, partitions, replicationFactor, retention)

acks selects the durability level: 0 (do not wait), 1 (leader wrote it), all (all in-sync replicas wrote it). The offset returned by produce identifies the message position within a partition.

Data model

The core structure is the partition log: an ordered, append-only sequence of messages. Each message has a monotonically increasing offset within its partition.

graph LR
    subgraph Topic["Topic: orders (3 partitions)"]
    P0["Partition 0: offsets 0..n"]
    P1["Partition 1: offsets 0..m"]
    P2["Partition 2: offsets 0..k"]
    end

On disk each partition is a directory of segment files (for example 1 GB each). Each segment has:

• a log file with message batches appended sequentially, named by the first offset it contains,
• a sparse offset index mapping offsets to file positions,
• a sparse time index mapping timestamps to offsets.

Only the newest segment is written. Retention deletes whole old segments, which is a cheap file delete.

Cluster metadata: topic configuration, partition-to-broker assignment, leader and in-sync replica (ISR) sets, and consumer group offsets. Metadata is small and needs strong consistency, so it is kept in a consensus-based store (ZooKeeper, or a built-in Raft quorum, see consensus, Paxos and Raft). Committed consumer offsets are stored in an internal, replicated, compacted topic.

High-level design

graph TD
    PR["Producers"] --> B1
    PR --> B2
    PR --> B3
    subgraph Cluster["Broker cluster"]
    B1["Broker 1: leader P0, follower P1"]
    B2["Broker 2: leader P1, follower P2"]
    B3["Broker 3: leader P2, follower P0"]
    end
    B1 <--> B2
    B2 <--> B3
    B3 <--> B1
    CTRL["Controller + metadata quorum"] --> B1
    CTRL --> B2
    CTRL --> B3
    B1 --> CG1["Consumer group A"]
    B2 --> CG1
    B3 --> CG1
    B1 --> CG2["Consumer group B"]

• Brokers. Store partition replicas and serve produce and fetch requests. Each partition has one leader that handles all reads and writes for it, and follower replicas that copy from the leader. Leaders are spread across brokers to balance load.
• Controller. One broker (or a quorum) that watches broker liveness, elects new partition leaders on failure, and publishes metadata.
• Clients. Fetch metadata to learn which broker leads each partition and connect directly to that broker. No separate routing tier is needed.

Produce and consume flow

sequenceDiagram
    participant P as Producer
    participant L as Leader broker
    participant F as Follower brokers
    participant C as Consumer
    P->>L: produce batch (key, value), acks=all
    L->>L: append to partition log
    F->>L: fetch new data
    L-->>F: messages
    F->>F: append to replica log
    L->>L: advance high watermark when ISR has it
    L-->>P: ack (partition, offset)
    C->>L: fetch from offset X
    L-->>C: messages up to high watermark
    C->>C: process
    C->>L: commit offset

Deep dive

1. Why the log design is fast

• Sequential I/O. Appending to the end of a file and reading sequentially are efficient on both HDDs and SSDs. Random access is avoided.
• Page cache. The broker relies on the operating system page cache instead of its own cache. Recent writes are read by consumers from memory.
• Batching. Producers batch messages per partition, and batches are compressed once and stored and sent in the same format, so the broker does not decompress and recompress them. Consumers fetch batches.
• Zero copy. Sending log data from the page cache to a socket can use the sendfile system call to avoid copying through user space.
• Consumer-tracked position. The broker does not track per-message acknowledgment state. A consumer group stores one offset per partition, so cost per message does not grow with the number of consumers.

Trade-off: batching increases throughput but adds latency (a linger time before sending).

2. Replication and durability

Each partition has a replication factor N (commonly 3). The leader appends the message and followers fetch it. The in-sync replica set (ISR) contains the leader and the followers that are caught up within a configured lag limit. A message is committed when all ISR members have it, and consumers can only read up to that point (the high watermark).

Producer acknowledgment levels:

A minimum ISR size setting (for example 2) makes writes fail when too few replicas are in sync, trading availability for durability, so acks=all cannot silently degrade to a leader-only write.

Leader failure: the controller picks a new leader from the ISR, which has all committed messages. Electing a replica outside the ISR (unclean election) restores availability sooner but can lose committed messages. This is the CAP trade-off expressed as a setting.

Typical settings for a durable topic: replication factor 3, min.insync.replicas=2, acks=all, unclean election disabled. This tolerates one broker failure with no loss and no write outage; two simultaneous failures make the partition reject writes (durability preferred over availability). A follower is dropped from the ISR when it has not caught up within about 10 to 30 seconds (replica.lag.time.max.ms), so one slow follower delays acks=all writes for at most that long before being removed. Spread replicas across racks or zones so a zone loss removes at most one replica. Cost of acks=all in one zone: about one extra intra-datacenter round trip (about 1 ms); across zones a few ms. Batching amortizes it: with linger.ms of 5 to 10 and 64 to 256 KB batches, throughput per producer rises severalfold for a latency cost of that linger.

Alternative: quorum replication (a message is committed when a majority acknowledges), used by Raft-based logs. It tolerates a slow replica without removing it from a set, but needs 2f + 1 replicas to tolerate f failures, while an ISR scheme with f + 1 replicas tolerates f failures at the cost of a separate mechanism to decide membership (see quorums and replication protocols).

3. Ordering and partitioning

Order is guaranteed only within a partition. The producer picks the partition:

• With a key: hash(key) mod partition_count, so all messages of a key go to the same partition and stay ordered.
• Without a key: round robin or sticky batching for even spread.

Consequences:

• No global order across partitions. Global order requires one partition, which limits throughput to one broker and one consumer per group.
• Changing the partition count changes the hash mod mapping, so keys move to different partitions and per-key order can break during the change. Plan partition counts in advance, or use consistent hashing for key placement.
• Hot keys overload one partition. Mitigate by salting keys when per-key ordering can be relaxed.

Choosing the partition count: partitions = max(target throughput / per-partition throughput, target consumer parallelism), rounded up with headroom for 2 to 3 years, because raising it later breaks key mapping. With about 10 MB/s per partition (a conservative practical figure) and 1 GB/s, that is 100; choose 200 to 500. Too many partitions inflate metadata and leader election time (thousands per broker is where controller failover and recovery become slow). Hot-partition handling: detect via per-partition bytes-in skew (alert when one exceeds about 2x the mean), then salt the key (key#0..k) for known heavy keys, giving up strict per-key ordering across the salts, or use a custom partitioner that isolates a heavy key on its own partition. Alternative to hash mod n: consistent hashing or a lookup table, which limits key movement when partitions are added but is not what most brokers provide by default.

Retries can reorder messages when several requests are in flight. Producers use sequence numbers per partition so the broker rejects out-of-order or duplicate batches, which preserves order with retries.

4. Consumer groups and delivery semantics

A consumer group shares the work of a topic: each partition is assigned to exactly one consumer in the group. Different groups are independent, giving publish-subscribe across groups and queue behavior within a group. When consumers join or leave, or fail (detected by missed heartbeats), the group rebalances and partitions are reassigned. During the rebalance, consumption of affected partitions pauses, so the design keeps rebalances rare and short.

Delivery semantics come from when the offset is committed relative to processing:

• At-most-once. Commit the offset, then process. A crash after commit loses the message.
• At-least-once. Process, then commit. A crash between them causes reprocessing, so processing must be idempotent (see delivery semantics and idempotency).
• Effectively-once. An idempotent producer (deduplicated by producer id and sequence number) removes duplicates caused by producer retries. Transactions write output messages and the consumed offsets atomically, so a consume-process-produce pipeline within the same cluster commits exactly once. Effects on external systems still need idempotent handling.

5. Retention, compaction, and consumer lag

Retention removes segments older than a time limit or beyond a size limit. Log compaction is an alternative for keyed data: the broker keeps only the latest message per key, which makes a topic a changelog of current state.

Consumer lag is the difference between the latest offset in a partition and the group's committed offset. It is the key health metric. Growing lag means consumers are slower than producers. Because the log is a buffer, slow consumers do not block producers, until retention deletes data the consumer has not read.

6. Comparison with a traditional queue model

Choose the log model for high-throughput streams, replay, and multiple consumers. Choose a per-message model for work queues with individual retry, priorities, or delayed delivery (see RabbitMQ, SQS, and broker comparison).

Failure modes

At 10x scale (10 GB/s, about 300 brokers): broker failures become routine (with 300 brokers, an annual disk failure rate of a few percent means a failure every few days), so recovery must be automatic and throttled; single clusters approach controller and metadata limits, so split into multiple clusters by workload (with a routing layer or by team); and cross-zone replication traffic becomes a real cost, which motivates follower fetching from the nearest replica and tiered storage to keep local disks small and recovery fast.

Bottlenecks and scaling

• Throughput. Add partitions and brokers. Increase batch size and use compression. Partition count also bounds consumer parallelism.
• Too many partitions. Each partition costs file handles, memory, and replication work and lengthens leader election after a failure, so partition counts are chosen with those costs in mind.
• Hot partitions. Skewed keys need key redesign or salting.
• Broker failure and recovery. Followers catch up by fetching from the leader. Rebuilding a broker's data takes time proportional to its size and can compete with production traffic, so replication throttles are applied.
• Storage. Tiered storage moves old segments to object storage (see object and blob storage) and keeps recent data on local disks.
• Rebalances. Use static membership and incremental (cooperative) assignment to avoid stopping all consumers.
• Backpressure. Producers have bounded buffers and block or fail when brokers are slow. Quotas limit per-client throughput (see backpressure, retries, and dead-letter queues).
• Poison messages. Because the log has no per-message retry, consumers route unprocessable messages to a dead-letter topic.

Trade-offs and follow-up questions

• Durability versus latency: acks=all and a minimum ISR add a replication round trip to each write.
• Availability versus consistency on failover: clean versus unclean leader election.
• Partition count: parallelism versus overhead, and the difficulty of changing it later.
• What if a consumer is slower than producers for hours? Lag grows; the log buffers it, but if lag exceeds retention (7 days here) the data is deleted unread. Alert on lag time (lag divided by the consume rate) rather than raw offsets, and scale consumers up to the partition count; beyond that, more partitions are required.
• What if you need exactly-once end to end? Idempotent producer plus transactions cover the broker-to-broker pipeline; for a database sink, write the result and the consumed offset in one database transaction, or upsert with a natural key.
• What if a producer crashes mid-batch? Unacknowledged batches are retried by a new session; with the idempotent producer, duplicates from the same producer session are rejected via sequence numbers; a restarted producer with a new id needs application-level dedupe keys.
• What if you need a single global order? One partition: about 10 to 100 MB/s ceiling and one active consumer per group. Alternative: order per key, or a sequencer that stamps messages with a monotonic id and consumers reorder in a bounded window.
• What if a rebalance storm occurs? Use static membership, a longer session timeout, and cooperative assignment so only moved partitions pause.
• How would delayed or scheduled messages be supported? Store them in time-bucketed internal topics or a separate scheduler, and publish to the target topic when due.
• How would a multi-region deployment work? Replicate topics asynchronously to another cluster. Active-active needs conflict-free handling of the same key produced in two regions.
• How to guarantee that a database change and a message publish both happen? Write them through a transactional outbox, not two independent writes (see transactional outbox and CDC).
• How would per-message priorities be added? Separate topics per priority, with consumers favoring the higher-priority topics.