Design a distributed message queue

I would build the queue as partitioned, append-only logs. A topic has about 200 partitions, each a sequence of segment files with a sparse offset index. Producers hash the message key to a partition, so a key stays ordered, and send compressed batches to that partition's leader. The leader appends them sequentially and assigns offsets. Writes are fast because they are sequential and recent data is served from the memory cache.

Each partition has three replicas in different zones. Followers fetch from the leader, and an acknowledgement at level all with a minimum of two in-sync replicas means an acknowledged message is on at least two machines. Consumers pull from the leader, and consumer groups share partitions, one consumer per partition. Each group commits its offsets to the broker, which gives replay and independent groups. Committing after processing gives at-least-once, and idempotent producers plus idempotent consumers make it effectively once.

A replicated controller watches heartbeats and, when a leader fails, elects a new one from the in-sync set and raises the epoch, which fences the old leader. The sizing is 1 GB/s of ingest, three copies, about 30 brokers. The main trade-off is that order holds only within a partition, in exchange for horizontal scale.

These numbers are assumptions for the exercise, not facts about a real system:

• 1 million messages a second, 1 KB each, and a replication factor of 3 (one leader and two followers).
• Messages are kept for 7 days. One broker sustains about 100 MB/s of disk writes.
• Each partition is kept to about 10 MB/s so it stays easy to move and recover.
• Consumers mostly read recent data, which is still in the operating system's memory cache.

The arithmetic:

Ingest rate

1,000,000 messages/s × 1 KB

1 GB/s from producers

Disk writes with replication

1 GB/s × 3 copies = 3 GB/s; ÷ 100 MB/s per broker

30 brokers at least

Storage for 7 days

1 GB/s × 86,400 s = 86.4 TB a day; × 7 = 605 TB; × 3 copies; ÷ 30 brokers

≈ 1.8 PB in all, about 60 TB per broker

Partitions

1 GB/s ÷ 10 MB/s per partition = 100; double for headroom

200 partitions, about 7 leaders per broker

Largest consumer group

One consumer per partition at most

up to 200 consumers read in parallel

A partition is an append-only log. New messages are only ever added at the end, and each gets the next offset, a number that grows by one. A message is identified by (topic, partition, offset), and its order is its offset.

The log is cut into segment files, for example one per gigabyte or per hour. Only the newest, the active segment, is written. Older segments never change. Each segment is named by its first offset and has a sparse index file that maps some offsets, say one every 4 KB of data, to a byte position. To read offset 41,200, find the segment by name with a binary search, look up the nearest index entry before it, then scan forward a few kilobytes. It needs no search of the whole log.

Why it is fast. Sequential disk writes are far faster than random ones, because the disk head or the flash controller never jumps around. The operating system keeps the recent part of the log in its page cache, so a consumer that keeps up reads from memory, not from disk. The broker can hand file bytes straight to the network without copying them through the application (the sendfile call). Producers send batches and often compress them. The broker stores the whole batch unchanged, and the consumer unpacks it, so the broker spends little CPU per message.

Retention is cheap. To delete data older than seven days, delete whole old segment files. That is one fast operation per file, not a delete per message. A second mode, compaction, keeps only the latest message per key, which suits a stream of updates to the current state.

Positions are kept per group. The broker does not track or delete messages as they are read. A consumer reads from an offset and commits the offset it has finished. The group's coordinator broker stores it for that group and partition in an internal compacted topic, replicated like any other log. This keeps the broker simple and allows many independent groups at different speeds. It also allows replay: reset the offset to last Tuesday and read again after a bug fix. The cost is that consumers must commit offsets carefully, which is the next topic.

Choosing the partition. Hash the key and take the remainder by the partition count. The same key always lands in the same partition and so stays ordered. Changing the count later reshuffles keys, so pick a count with headroom.

A consumer group is a set of consumers with one group id. Each partition is assigned to exactly one consumer in the group, so messages in a partition are handled in order by one worker. A group can use at most as many active consumers as there are partitions. A second group with a different id receives every message again, independently.

A broker acting as the group's coordinator tracks members through heartbeats. When a member joins, leaves or stops sending heartbeats, the coordinator starts a rebalance and reassigns partitions. During a plain rebalance everyone pauses. Newer incremental rebalancing moves only the partitions that must move. Static membership, where a restarting consumer keeps its identity for a short grace period, avoids a rebalance for a quick restart.

Delivery guarantees depend on when you commit the offset.

• At most once. Commit first, then process. A crash after the commit loses the message.
• At least once. Process first, then commit. A crash before the commit means the batch is read again, so the consumer sees duplicates. This is the usual default.
• Effectively once. Accept at-least-once delivery, and make duplicates harmless.

To make duplicates harmless, work at both ends. On the producer side, an idempotent producer adds a producer id and a sequence number to each batch for a partition. The broker discards a batch it has already stored, so a retry after a lost acknowledgement does not duplicate it. On the consumer side, either make the processing idempotent, for example by an upsert keyed by message id, or save the result and the new offset together in one transaction in the same database.

Be careful with the phrase "exactly once". It holds inside the queue's own read-process-write cycle. A side effect outside it, such as an email, can still repeat unless it is idempotent.

Bad messages. A message that always fails would block its partition if retried forever. After a few attempts, move it to a dead-letter topic with the error, and carry on. Watch lag, which is the log end offset minus the committed offset, and add consumers, up to the partition count, when it grows.

Each partition has a replication factor of 3: one leader and two followers, placed in different zones. All reads and writes go to the leader, which keeps ordering simple. Followers fetch from the leader as a consumer would and append to their own log.

The in-sync replicas (ISR) are the leader plus the followers that are caught up, within a lag limit such as 10 seconds. The high-water mark is the highest offset that every in-sync replica holds. Consumers can read only up to it, so they never see a message that a failover could erase.

Ack levels let the producer choose its safety:

• Ack 0: do not wait. Fastest, and a message can vanish with no sign.
• Ack 1: the leader has written it. If the leader dies before a follower copies it, the message is lost.
• Ack all: every in-sync replica has it. Combined with a minimum in-sync count of 2, the write is refused whenever fewer than two replicas are in sync. An acknowledged message is then on at least two machines.

The minimum in-sync setting is the durability and availability dial. A higher value means fewer losses and more refused writes during trouble.

When a leader fails, the controller notices missing heartbeats after the session timeout, about 6 seconds here. It picks the new leader from the in-sync set, so the chosen replica holds every acknowledged message. It raises the leader epoch, a number stamped on requests. Brokers reject requests carrying an old epoch, which fences off a former leader that wakes up and still believes it is in charge. When that old broker rejoins, it truncates any messages the new leader never had and then follows.

Electing a replica that is not in sync, an unclean election, restores availability sooner and loses acknowledged messages. Leave it off for data that matters.

The controller must not be a single point of failure, so it is a small group of 3 or 5 nodes agreeing through a consensus protocol. With a 6-second timeout and a short election, a partition accepts writes again well inside the 10-second target.