How to design a distributed message queue

A distributed message queue is really an append-only log; hand the reader the offset and almost every property (replay, extra consumers, sequential writes) falls out of that one choice.

The one decision

• Consumer owns the read position, not the broker. The bookmark is one integer per group per partition (“finished everything before this offset”).
• Broker’s whole job collapses to “append bytes, serve bytes by offset.”
• Consequence: replay is a parameter, extra consumer groups are near-free, and all failures move to the consumer side.

Terms and interface

• Topic: named message stream. Partition: one append-only file the topic splits across. Offset: position within a partition, from zero, set at append. Broker: machine storing/serving partitions.
• produce(topic, key, value) → (partition, offset); key picks partition via hash(key) % P.
• fetch(topic, partition, offset, max_bytes) → in-order batch from offset. Followers replicate using this same call.
• commit(group, topic, partition, offset): durable read position, stored in the compacted __consumer_offsets topic.
• Ordering is total inside a partition, undefined across partitions.

Sizing (1M msg/s, 1KB, RF 3, 7-day retention, 3 groups)

• Storage: ≈ 1.81 PB cluster-wide. Not the binding constraint.
• Network: each byte crosses a NIC 8x (in, out to 2 followers, in at 2, out to 3 groups) → 8 GB/s. This binds.
• Brokers: 8 GB/s ÷ 75 MB/s (60% of a 1 Gbps NIC) ≈ 107, round to 128. NIC ~50%, disk ~2.3%.
• “A message broker is a network device that happens to persist.” Writes are sequential, so the disk never seeks.

Log on disk

• Partition = directory of segment files, rolled shut at 1 GiB, never modified.
• 24-byte header per record; 1 KB message = 1,024 bytes; 1 GiB segment ≈ 1,048,576 records.
• Sparse .index (offset→byte) and .timeindex (time→offset), one entry per 4 KB, held in RAM. fetch = binary search + short scan.
• Only mutable state in the system: the committed offset.

Log vs table (~320x cheaper)

• Table-as-queue touches ~8 random 4 KB I/Os per 1 KB message (row + 2 indexes + claim read + claim write + ack write + ack index + vacuum) → ~328 devices for 1M msg/s.
• Log appends 1,024 B sequentially → 1 device. ~10x from sequential-vs-random, ~32x from write amplification.
• Reads served from page cache by sendfile; ~100 GB RAM vs ~23 MB/s log ≈ 1.19 h cached. One lagging consumer reading old data evicts everyone’s cache. Consumer lag is the primary health metric.

Partitions: parallelism = ordering knob

• Count can only be raised, never lowered (lowering re-routes keys, breaks per-key order).
• Global ordering forces 1 partition → 1 consumer → ~100 msg/s. Order per key instead.
• Partition count set by consumer speed, not throughput: 1M ÷ 100 msg/s = 10,000 at 100% util; +20% headroom = 12,000; round to 16,384. Max useful consumers = partition count.
• Sticky partitioner batches keyless records (cuts request rate ~16x); keyed records can’t use it.

Delivery semantics

• At-least-once is the default. Duplicates = rate × commit_interval × restarts ≈ 58/s at 5 s interval. Group size cancels.
• Exactly-once across two systems is impossible (needs 2PC or infinite acks; Two Generals). Kafka’s exactly-once is stream processing (all endpoints in the log), not delivery; adds ~50 ms.
• Idempotent consumer (sink is a DB): in the same transaction as the effect, insert a dedup key on a unique column. Key on the producer-minted id, never topic:partition:offset (offsets go backwards after unclean election).
• Dedup table remembers only the redelivery horizon (session timeout + rebalance, e.g. 60 s ≈ 960 MB), not the full retention window.

Replication, ISR, acks

• ISR: replicas caught up within replica.lag.time.max.ms. High watermark = highest offset on every ISR member; consumers can’t read past it, so failover never drops a seen record.

• acks=all ≠ on disk: appends land in page cache, no per-record fsync; two simultaneous losses can still lose acked data.
• Unclean leader election promotes an out-of-ISR replica: truncates ~30 s of data silently and shortens the log (why dedup keys must not use offsets). unclean.leader.election.enable: false for payments, true for clickstream.

Consumer lag and rebalance

lag         = log_end_offset - committed_offset
lag_seconds = lag / consume_rate      (what a human acts on)
d(lag)/dt   = produce_rate - consume_rate   (what predicts)

• Sign of d(lag)/dt is the diagnosis: positive = broken, negative = finite drain.
• Drain time set by headroom: 30 M ÷ (1.2M − 1M) = 150 s at 20%; = 600 s at 5%; ∞ at zero.
• Backpressure lives on the producer: bounded buffer, send() blocks up to max.block.ms then throws. ~33 s of broker downtime before app threads see exceptions.
• Rebalance storm: slow poll() → ejected → rebalance stalls all → backlog grows → bigger batch → slower → ejected again. Keep max.poll.records × sec_per_record < max.poll.interval.ms. Fix by lowering the batch. Cooperative rebalancing + static membership turn a 12,000-consumer deploy from 12,000 rebalances to ~0.

Push vs pull

• Chose pull: backpressure sits where capacity is known, consumer sets its own batch, replay is only expressible as fetch(offset).
• Cost is idle polling; long poll (fetch.max.wait.ms) fixes it: request rate depends on data rate, not poll interval (~5x fewer requests).

Gotchas

• Mixing binary (GiB) and decimal (GB) units silently is a ~7% error.
• Retention expiring unread data is the silent failure: alarm on lag_seconds > 0.5 × retention.
• Poison message blocks its partition: bounded retries then a dead-letter topic.
• New brokers start empty and don’t self-fill: moving 14.18 TB over half a NIC takes ~2.6 days. Plan in “days to rebalance.”
• Managed (SQS) is price-competitive but rejected on capability: no replay, no independent groups, at-least-once only.