Design a distributed message queue

Design a distributed message queue like Kafka. Start with one broker and one log, grow it with partitions by key, consumer groups with committed offsets and three replicas with an in-sync set, then stress it with what acknowledged means, exactly once, disk speed, rebalances and tiered storage, with code.

Senior28 min core · 46 min full depth

Read first: Queues and streams Replication
Scaling patterns: Idempotency and exactly-once effects Hot keys and hot partitions Backpressure and load shedding Multi-region and active-active
Building blocks: Queues and streams Partitioning Replication Consensus Consistency models Blob storage Transactions and sagas
Related: Pub-sub system Notification system Ad click aggregator

“Design a distributed message queue, something like Kafka. Many teams publish events into it, and many services read them.”

It sounds like plumbing. It’s a storage system with a consensus problem inside. A message queue in this sense is a durable, replicated, partitioned log. Producers append to the end. Consumers read forward from a position they remember, and nothing is deleted on read. The hard questions are what that log promises: when a write is safe, what order readers see, and what “exactly once” can mean.

So this page starts with one broker and one log, and lets the numbers grow it. Then it tests the result against those promises.

The answer in one minute

If you only have a minute, this is the design. The rest of the page explains each part.

An append-only log, read by position. Each message gets an offset, its position in the log. Consumers remember their own offset, so many teams can read and replay the same log.
Split topics into partitions, by key. Each partition is its own log on one broker. Order holds per partition, so all events for one order go to one partition.
Consumer groups share the partitions. Each partition is read by one member of a group, which keeps order without locks. Processing, then committing the offset, gives at least once.
Three replicas in three zones, acknowledged by the in-sync set. A write is acknowledged once every replica that’s keeping up has it. At least two must have it. Consumers only see messages that can’t vanish.
Exactly once, stated precisely. Producers don’t duplicate on retry, and transactions make read-process-write atomic inside the log. Side effects outside it still need an idempotent consumer.

Scope it

The first question changes the whole storage model: is it a queue or a log?

Interviewerminute 0

Design a distributed message queue, something like Kafka.

What they are testing

Whether you ask “queue or log?” first, then pin down throughput, ordering and the delivery guarantee. About three minutes. Drawing brokers before knowing whether messages are deleted on read is the red flag.

These four questions matter most:

A queue (delete on acknowledge) or a log (replay)? A log. Messages are kept 7 days, and any reader can replay.
How much traffic? 1 million messages a second, 3 million at peak, 1 KB each.
What order do readers need? Per key (all events of one order, one user), not global.
How durable is “acknowledged”? It survives losing any one broker, disk or data center zone.

From these answers we make five promises:

Throughput. 1 GB/s produced on average, 3 GB/s at peak, read about three times over.
Fast. A produce in under 20 ms at p99, fully replicated. Producer to consumer in under 100 ms.
Durable. An acknowledged message survives losing any one broker, disk or zone.
Always up. 99.99% per partition. A new leader in seconds, not minutes.
Ordered. A total order per partition, kept even when producers retry.

We leave out three things. Per-message acknowledgements, delays and priorities, because a log tracks positions, not messages. Schema registries and stream processing, which are clients of this system. And writes in two regions at once, because two orders for one partition can’t be merged.

Go deeperMore questions, and what each changes1 min

Question	We assume	If the answer were different
What delivery guarantee?	At least once; exactly once for read-process-write	Outside side effects need idempotent consumers whatever we do
How many readers per message?	About three consumer groups per topic	Hundreds of readers make fetch traffic the thing to size for
Message size?	1 KB on average, 1 MB at most	Large payloads go to blob storage behind a pointer
How many topics and teams?	About 2,000 topics, 30,000 partitions, 50 teams	Quotas and isolation matter, as here
Global order?	No, per key	One partition: one broker’s 300 MB/s and one consumer, forever
Survive a region synchronously?	No	Adds 30 to 80 ms to every produce

The numbers that decide it

Now we work out the numbers. Count bytes, not messages, and multiply by copies and readers.

Assumptions

Messages produced a second, average: 1 million
Bytes per message, compressed: 1 KB
Peak over average: 3×
Replication factor: 3×
Consumer groups reading each message: 3
Days of retention: 7
Usable disk per broker: 14 TB
Messages per consumer a second: 50,000
Largest topic’s share of all traffic: 20%

Produce1 GB/s3 GB/s at peak

1 million messages a second at 1 KB. One broker appends about 300 MB/s.

So we every busy topic is split into partitions spread over many brokers.

Broker traffic24 GB/s40 brokers

Each message is written on 3 replicas, fetched twice by followers and once by each of 3 consumer groups: 3 GB/s × 8. A broker moves about 1 GB/s, and we keep each near 60%, so it can take over a dead neighbor’s partitions.

So we size for reads, not appends.

Storage1.8 PB130 brokers

1 GB/s × 7 days × 3 replicas. At about 14 TB per broker, that’s three times the brokers traffic needs.

So we keep only recent data on brokers, and move the rest to object storage.

Partitions32largest topic

The largest topic peaks at 600,000 messages a second. At 50,000 per consumer that’s 12 consumers; 24 partitions with headroom, rounded to 32.

So we size partitions by consumer parallelism, and over-provision once, because adding them later moves keys.

Storage, not traffic, sizes this cluster. Try the calculator: cut retention to 3 days, and that gap nearly closes. Make consumers 100 times slower, and the largest topic needs 2,400 partitions, which is a reason to fix the consumer, not to add partitions.

Change an assumption

Messages produced a second, average1 millionBytes per message, compressed1 KBPeak over average3×Replication factor3×Consumer groups reading each message3Days of retention7Usable disk per broker14 TBMessages per consumer a second50,000Largest topic’s share of all traffic20%

Produce traffic, average: 1 GB/s
Produce traffic, peak: 3 GB/s; Past the roughly 300 MB/s one broker appends: split topics into partitions spread over brokers.
Broker traffic at peak, in plus out: 24 GB/s; Appends on every replica, follower fetches, and one fetch per consumer group.
Brokers for throughput, 60% busy: 40; At about 1 GB/s per broker, leaving room to take over a failed broker’s partitions.
Retained on broker disks: 1.8 PB
Brokers for disk: 130
Disk brokers per throughput broker: 3.3; Disks, not traffic, size the cluster: keep hours locally and move older segments to object storage.
Partitions for the largest topic, 2× headroom: 24; Set by how many consumers must share the topic, not by broker throughput.
Cross-AZ bytes a month: 6.9 PB; Replication, plus producers reaching leaders in other AZs. At cloud prices this rivals the broker bill.

Build the design

We start with the simplest design that is correct. Then we let the numbers break it, one at a time. Each step fixes only what broke, adds only the boxes that fix needs, and says what the fix costs. Throughout, follow one scenario: a producer sends an order event, and the billing service reads it.

Step 1. One broker, one log

A broker appends each message to the end of the topic’s log file and gives it the next offset. A consumer asks for “everything from offset N”, processes it, and remembers N plus the count for next time. Nothing is deleted on read, so three teams can read the same log at their own pace, and any of them can rewind and replay.

Distributed message queue · architecture

Stage 1: One broker, one log

Color keyServiceDatabaseQueue or streamBackground workerNew in this stageAsync or occasional

Messages produced per second: 3,000,000

Stage 1 of 6

One broker, one log

Why this stage

Nothing yet.

This is the smallest design that accepts messages, keeps them in order and lets readers come back for them.

What we add

A broker that appends each message to a log file per topic and hands it a sequential offset.

Consumers pull from an offset they remember.

What it costs

One machine caps throughput at about 300 MB/s, one disk holds everything, and one failure loses data.

At 3M messages produced per second, Brokers is 1000% busy. Stage 2 is the fix.

Pinch or tap the zoom buttons to read small labels.Drag any box to move it. Reset puts every box back.

The numbers behind the busy bars

What the traffic is

Messages are 1 KB after compression.
Three consumer groups each read every message.
From stage 4, every message is written 3 times, and followers fetch it twice more.

What one machine can do

One broker appends about 300,000 messages a second (300 MB/s) and serves about 900,000 fetched messages from page cache.
In a cluster, one broker moves about 1 GB a second in total: appends, replica fetches and consumer fetches.
One consumer process handles about 50,000 messages a second of light processing.

Headroom

Instance counts keep each box near 60% busy. That leaves room to take over a failed broker's partitions.

This is correct, and it is how every queue in this family works. But the numbers say 3 GB/s at peak, far more than one machine can write.

Step 2. Partitions, by key

One log can’t be split across machines without someone deciding the order of every write. So the topic is split instead, into partitions: independent logs, each with its own offsets, each on one broker. A controller assigns partitions to brokers and publishes the map. Clients cache it and send each batch straight to the broker that owns the partition.

The producer picks the partition, usually as a hash of the message’s key, so one key always lands in one partition. That choice is the ordering contract. Choose the key from what must stay in order, not from what spreads load: order_id for order events. A hot key (one huge customer) makes one partition hot. The fix is a finer key that keeps the needed order, such as customer plus order.

Distributed message queue · architecture

Stage 2: Partitions across brokers

Color keyServiceDatabaseQueue or streamBackground workerNew in this stageAsync or occasional

Stage

Messages produced per second: 3,000,000

Stage 2 of 6

Partitions across brokers

Why this stage

At the 3 million messages/s peak (3 GB/s), one broker that appends about 300,000 a second is 1,000% busy.

One log cannot be split across machines, so the topic has to be.

What we add

Each topic split into partitions, each partition an independent log on one broker, with the producer choosing the partition by a hash of the key.

A controller owns the map from partition to broker.

What it costs

Order now holds only within a partition, and a message’s key decides which order it is part of.

At 3M messages produced per second, Consumer groups is 6000% busy. Stage 3 is the fix.

Pinch or tap the zoom buttons to read small labels.Drag any box to move it. Reset puts every box back.

The numbers behind the busy bars

What the traffic is

Messages are 1 KB after compression.
Three consumer groups each read every message.
From stage 4, every message is written 3 times, and followers fetch it twice more.

What one machine can do

One broker appends about 300,000 messages a second (300 MB/s) and serves about 900,000 fetched messages from page cache.
In a cluster, one broker moves about 1 GB a second in total: appends, replica fetches and consumer fetches.
One consumer process handles about 50,000 messages a second of light processing.

Headroom

Instance counts keep each box near 60% busy. That leaves room to take over a failed broker's partitions.

Go deeperHow many partitions a topic needs1 min

A topic needs enough partitions for its slowest side. Say one consumer handles 50,000 messages a second, and the topic peaks at 600,000. The group needs at least 12 consumers, so at least 12 partitions. Teams usually add room for growth. Adding partitions later changes which partition a key hashes to, and that breaks order for the keys that move.

Too many partitions cost something too. Each one is a set of open files on its broker. It’s a leader to elect when a broker dies, and an entry in the controller’s metadata. A failover that must move 100,000 leaders takes longer than one that moves 1,000. So a common rule is a few thousand partitions per broker at most. And the partition count per topic is chosen once, with growth in mind.

Step 3. Consumer groups and offsets

Now writes spread out. But reads don’t: one consumer handles about 50,000 messages a second, and each team must read 3 million. So each team runs a consumer group: each partition is read by exactly one member. That keeps order without locking, and caps parallelism at the partition count. A group coordinator on one broker tracks members by heartbeat, assigns partitions, and stores each group’s committed offsets.

The commit loop decides the delivery guarantee. Process, then commit, and a crash replays: at least once. Commit, then process, and a crash skips: at most once. This loop commits every 5 seconds, and right away when partitions are taken from it.

CommitLoop.javacommit_loop.pyCommitLoop.cscommit_loop.gocommit-loop.ts

// At-least-once consumption: process first, commit after. A crash between the two replays the
// messages since the last commit, so the handler must be idempotent. Committing before
// processing would turn the same crash into lost messages instead: at most once.
record TopicPartition(String topic, int partition) {}

record ConsumerRecord(TopicPartition partition, long offset, byte[] value) {}

interface Consumer {
    // Poll is also the liveness check: go longer than max.poll.interval.ms (5 minutes by
    // default) between polls and the group evicts this member and rebalances.
    List<ConsumerRecord> poll(Duration timeout);
    void commitAsync(Map<TopicPartition, Long> offsets); // returns at once
    void commitSync(Map<TopicPartition, Long> offsets);  // waits for the coordinator
}

final class CommitLoop {
    // Bounds the replay after a crash to about 5 seconds of messages. Committing after every
    // message would add a replicated write to the offsets topic per message.
    private static final Duration COMMIT_EVERY = Duration.ofSeconds(5);

    private final Consumer consumer;
    private final java.util.function.Consumer<ConsumerRecord> handler; // must be idempotent
    private final Clock clock;
    // Per partition, the next offset to read: last processed + 1. That is what a commit
    // stores, so a restart begins at the first message not yet processed.
    private final Map<TopicPartition, Long> processed = new HashMap<>();
    private Instant lastCommit;
    private volatile boolean running = true;

    CommitLoop(Consumer consumer, java.util.function.Consumer<ConsumerRecord> handler, Clock clock) {
        this.consumer = consumer;
        this.handler = handler;
        this.clock = clock;
        this.lastCommit = clock.instant();
    }

    void stop() { running = false; }

    void run() {
        try {
            while (running) {
                for (ConsumerRecord r : consumer.poll(Duration.ofMillis(500))) {
                    handler.accept(r);                              // if this throws, r is not
                    processed.put(r.partition(), r.offset() + 1);   // marked, so it is redone
                }
                Instant now = clock.instant();
                if (!processed.isEmpty() && Duration.between(lastCommit, now).compareTo(COMMIT_EVERY) >= 0) {
                    // Async commits are never retried: a retry could land after a newer commit
                    // and move the offset backwards. The next periodic commit covers a failure.
                    consumer.commitAsync(Map.copyOf(processed));
                    lastCommit = now;
                }
            }
        } finally {
            // Leaving cleanly: wait for this one, so the next owner starts exactly here.
            consumer.commitSync(processed);
        }
    }

    // The client calls this inside poll() when a rebalance takes partitions away. Commit them
    // synchronously now, or the next owner starts from an older offset and redoes our work.
    void onPartitionsRevoked(Collection<TopicPartition> revoked) {
        Map<TopicPartition, Long> mine = new HashMap<>();
        for (TopicPartition tp : revoked) {
            Long next = processed.remove(tp);
            if (next != null) mine.put(tp, next);
        }
        consumer.commitSync(mine);
    }
}

Distributed message queue · architecture

Stage 3: Consumer groups and offsets

Color keyServiceDatabaseQueue or streamBackground workerNew in this stageAsync or occasional

Stage

Messages produced per second: 3,000,000

Stage 3 of 6

Consumer groups and offsets

Why this stage

Three applications each read all 3 million messages/s.

One consumer process handles about 50,000 a second, so each application needs dozens of processes sharing the partitions, and a crashed one must resume where it stopped.

What we add

Consumer groups: each partition goes to exactly one member of a group.

A group coordinator on a broker hands out partitions and stores committed offsets in a compacted internal topic.

What it costs

Rebalances pause consumption when members come and go, and a crash between processing and committing replays messages.

At 3M messages produced per second, every part has headroom. Raise the traffic until something turns red.

Pinch or tap the zoom buttons to read small labels.Drag any box to move it. Reset puts every box back.

The numbers behind the busy bars

What the traffic is

Messages are 1 KB after compression.
Three consumer groups each read every message.
From stage 4, every message is written 3 times, and followers fetch it twice more.

What one machine can do

One broker appends about 300,000 messages a second (300 MB/s) and serves about 900,000 fetched messages from page cache.
In a cluster, one broker moves about 1 GB a second in total: appends, replica fetches and consumer fetches.
One consumer process handles about 50,000 messages a second of light processing.

Headroom

Instance counts keep each box near 60% busy. That leaves room to take over a failed broker's partitions.

Go deeperWhy consumers pull instead of the broker pushing1 min

In a push design, the broker sends messages to consumers as they arrive. In a pull design, each consumer asks for the next batch when it is ready. This log uses pull, for three reasons.

A slow consumer can’t be flooded: it simply asks less often, and its lag grows on the broker’s disk, where it costs nothing. Batches grow by themselves under load, because a consumer that falls behind gets many messages per request. And a consumer can rewind just by asking for an older offset.

The cost is a consumer that asks when nothing is new. So a fetch can wait on the broker for up to a set time, such as 500 ms, until enough data arrives. This “long poll” keeps idle consumers cheap.

Step 4. Three replicas, one leader

Every message now has one home. That is the next thing to break: with 40 brokers, losing one every few weeks is normal. So every partition gets three replicas in three zones. A leader takes all writes, and two followers copy from it with the same fetch request consumers use.

The leader keeps an in-sync replica set (ISR): the replicas that have caught up within the last 30 seconds. With acks=all, a write is acknowledged once every ISR member has it. And min.insync.replicas = 2 refuses writes when only the leader is left, so an acknowledged message is always in at least two zones.

Who leads and who’s in sync must never be ambiguous. So the controller is a three-node Raft quorum: three machines that agree on cluster metadata as a shared log. When a broker dies, the controller picks new leaders from each ISR, and everyone learns it from that log. The controller is never on the path of a message.

Distributed message queue · architecture

Stage 4: Replicate and elect leaders

Color keyServiceDatabaseQueue or streamBackground workerNew in this stageAsync or occasional

Stage

Messages produced per second: 3,000,000

Stage 4 of 6

Replicate and elect leaders

Why this stage

With 20 to 40 brokers, a broker is lost every few weeks and a disk more often.

Without copies each failure loses acknowledged messages and takes its partitions offline.

A failure domain, not load, forces this stage.

What we add

Three replicas of every partition in three AZs: a leader takes writes, followers fetch from it, and only in-sync replicas count for acks=all.

The controller becomes a three-node Raft quorum that elects partition leaders.

What it costs

Three times the disk writes, twice the fetch traffic on leaders, and a produce now waits for the slowest in-sync follower.

At 3M messages produced per second, every part has headroom. Raise the traffic until something turns red.

Pinch or tap the zoom buttons to read small labels.Drag any box to move it. Reset puts every box back.

The numbers behind the busy bars

What the traffic is

Messages are 1 KB after compression.
Three consumer groups each read every message.
From stage 4, every message is written 3 times, and followers fetch it twice more.

What one machine can do

One broker appends about 300,000 messages a second (300 MB/s) and serves about 900,000 fetched messages from page cache.
In a cluster, one broker moves about 1 GB a second in total: appends, replica fetches and consumer fetches.
One consumer process handles about 50,000 messages a second of light processing.

Headroom

Instance counts keep each box near 60% busy. That leaves room to take over a failed broker's partitions.

Follow one produce and one consume:

Distributed message queue · architecture

Trace a request: Produce with acks=all

Color keyServiceDatabaseQueue or streamBackground workerNew in this stageAsync or occasionalHop in motionHops doneResponse

Produce with acks=all: step 1 of 4

Producers to Brokers

The client hashes the key to partition 7 and adds the record to that partition’s batch. The batch leaves when it reaches 64 KB or after 5 ms of linger, compressed with zstd, straight to partition 7’s leader, found in cached metadata. 5 ms of linger plus about 1 ms of network.

6 ms of a 20 ms budget

Producers → Brokers6 ms
Brokers → Log segments0.5 ms
Brokers → Follower replicas3 ms
Response

Pinch or tap the zoom buttons to read small labels.Drag any box to move it. Reset puts every box back.

This is enough for a mid-level answer. A log with offsets, partitions by key, consumer groups, and three replicas with a leader. The stress tests are what make it a senior answer.

Go deeperThe API2 min

Produce(acks = all, timeout = 30 s, transactional_id?)
  topic "orders", partition 7:
    RecordBatch(producer_id 4211, epoch 3, base_sequence 880, 412 records, zstd)
→ partition 7: error NONE, base_offset 81,920,000
  errors: NOT_LEADER (refresh metadata, retry), NOT_ENOUGH_REPLICAS (in-sync set too small),
          OUT_OF_ORDER_SEQUENCE, PRODUCER_FENCED, THROTTLED (quota, with delay ms)

Fetch(max_wait = 500 ms, min_bytes = 1, isolation = read_committed)
  topic "orders", partition 7: fetch_offset 81,920,431, max 1 MB
→ partition 7: high_watermark 81,920,912, records...

OffsetCommit(group "billing", generation 41, orders-7 → 81,920,912)

Clients use a binary protocol on long-lived connections. Every request is a batch, and a producer can have up to 5 in flight per broker. A fetch is a long poll: with nothing new, it waits up to max_wait, so a caught-up consumer gets data within milliseconds. The generation number rejects commits from a member already removed from the group.

Consumers pull rather than being pushed to. That builds in backpressure: a slow consumer simply asks less often, and the log holds the rest. The cost is a thick client library in every language.

Go deeperOn disk: segments and batches1 min

A partition is a directory of segments. Each is a log file of record batches, plus two small indexes, named by the first offset it holds:

orders-7/
  00000000000081000000.log         record batches, appended in order, about 1 GB per segment
  00000000000081000000.index       sparse: offset → byte position, one entry per 4 KB of log
  00000000000081000000.timeindex   sparse: timestamp → offset
  00000000000081920000.log         the active segment: the only file ever written
  leader-epoch-checkpoint          epoch → first offset written in that epoch

A record batch is the unit of everything: compression, replication, the duplicate check and the fetch response. Its header carries the base offset, the leader’s epoch and a checksum. It also carries the producer ID, epoch and sequence number. The records inside are compressed together.

Offsets are messages too: a commit is a record in a special internal topic that keeps only the latest value per key. Cluster metadata is also a log, kept by the controller quorum.

How partitions are chosen	Order kept for	Use when
Hash of a message key	Every message with that key	Events about one thing must stay in order
No key, sticky batches	Nothing	Logs and metrics
One partition	Everything	Low-volume control topics only

Adding partitions later changes hash(key) mod count, so most keys move. New messages for a key land in a new partition while older ones may be unread in the old one. So over-provision at creation.

Stress it

The design is complete. Now we test the promises the log makes. What “acknowledged” survives. What “exactly once” can mean. Why disk is fast enough. Deploys that stall consumers. And ten times the size. Only the last one adds boxes.

What “acknowledged” means

Interviewerminute 20

The leader of a partition dies right after acknowledging a write. Which replica becomes leader, and is the write still there?

What they are testing

Whether you know that only an in-sync replica may become leader, and what the high watermark promises.

acks=all means all in-sync replicas, not all replicas. If both followers fall out of the ISR, “all” is just the leader, and an acknowledged write is on one machine. That’s why min.insync.replicas = 2 exists: below two in sync, the leader refuses the write rather than acknowledge a single copy.

Each replica’s log ends somewhere. The high watermark is the smallest of those ends across the ISR. Every message below it is on every in-sync replica. Consumers only see messages below the high watermark. Without that rule, a consumer could read a message, act on it, and then watch it vanish in a failover.

Leader failover: the new leader comes from the ISR, the old one truncates

Scroll or pinch to zoom, drag to pan

Color key
Service
Queue or stream

The new leader comes from the ISR, so it has every acknowledged message. The old leader, when it returns, asks where the previous leader epoch ended in the new leader’s log, and cuts off anything after it. Those messages were never acknowledged, so losing them is allowed, and every replica ends with the same log.

The epoch also stops a zombie leader, one cut off from the controller that still thinks it leads. It can’t shrink the ISR on its own, and the followers have stopped fetching from it. So its acks=all writes time out instead of succeeding, and it steps down once it hears of the new epoch.

When no in-sync replica survives, there are two ways to fail. Wait for an ISR member to come back, and the partition is down until then: right for orders and payments. Or allow unclean leader election: an out-of-sync replica leads, its missing messages are gone, and writes resume. It’s off by default, and should stay off for anything that matters, because it silently turns acknowledged messages into lost ones.

Go deeperThe acks settings, and why not a majority quorum1 min

acks	The leader acknowledges after	Survives	Use for
0	Nothing: the producer doesn’t wait	Nothing guaranteed	Metrics you can lose
1	Its own append	A follower failure, not the leader’s	Logs, clickstreams
all	Every member of the current ISR	Losing all but one in-sync replica	Anything that matters; the default

Why not a Raft majority for every write, like the controller? It works, and some systems do it. But with all three in sync, the ISR keeps acknowledged data readable after losing two of three. A three-node majority quorum can’t even elect a leader then. The ISR’s membership must still be decided by something consistent, and that’s the controller’s Raft log. So consensus goes where it’s cheap, on metadata.

The controller. Older Kafka kept metadata in ZooKeeper, and a new controller had to read every partition’s metadata before acting, minutes on large clusters. The built-in Raft quorum keeps standbys with the metadata already in memory, so failover takes about a second. Kafka 4.0 removed ZooKeeper entirely.

At least once, and what exactly once really means

Product asks for “exactly once”. Can we give it to them?

Three places can duplicate or lose a message, and each needs its own fix:

Where	What goes wrong	Fix	What it gives
Producer to broker	The acknowledgement is lost; the retry writes it again	Idempotent producer	No duplicates or reordering from retries
Consume, process, produce	A crash after writing output, before committing input	Transactions	Exactly once from input topic to output topic
Consumer to another system	A crash after the side effect, before the commit	Idempotent consumer	Effectively once in that system

The idempotent producer. Each producer gets an ID and an epoch, and numbers its batches per partition. The leader remembers the last 5 batches per producer. A retry gets the original offset back, and a gap is refused, because it would reorder the partition. It’s on by default.

A retried batch is written once

Scroll or pinch to zoom, drag to pan

Color key
Database
Service

The leader’s side of that check is the whole mechanism:

ProducerStateTable.javaproducer_state_table.pyProducerStateTable.csproducer_state_table.goproducer-state-table.ts

// Broker side of the idempotent producer. The broker gives each producer an id (PID) and an
// epoch; the producer numbers its batches per partition 0, 1, 2... The partition leader keeps
// this table, so a retried batch is answered with its original offset instead of being
// written twice, and a batch that skips a number is refused instead of being written out of order.
final class ProducerStateTable {
    // Matches the producer's limit of 5 requests in flight: any batch it can still be retrying
    // is one of the last 5, so 5 is all the history a duplicate check ever needs.
    static final int CACHED_BATCHES = 5;

    record Batch(long producerId, short epoch, int firstSeq, int lastSeq, List<byte[]> records) {}

    interface PartitionLog {
        long append(List<byte[]> records); // returns the offset of the first record
    }

    static final class ProducerFencedException extends RuntimeException {
        ProducerFencedException(String m) { super(m); }
    }

    static final class OutOfOrderSequenceException extends RuntimeException {
        OutOfOrderSequenceException(String m) { super(m); }
    }

    private record Appended(int firstSeq, int lastSeq, long baseOffset) {}

    private static final class ProducerState {
        final short epoch;
        final ArrayDeque<Appended> recent = new ArrayDeque<>(); // oldest first

        ProducerState(short epoch) { this.epoch = epoch; }

        int lastSeq() { return recent.isEmpty() ? -1 : recent.peekLast().lastSeq(); }
    }

    // Followers rebuild this same table as they replicate, because PID, epoch and sequence are
    // in every batch header. A new leader therefore rejects duplicates from before the failover,
    // which is exactly when producers retry the most.
    private final Map<Long, ProducerState> producers = new HashMap<>();
    private final PartitionLog log;

    ProducerStateTable(PartitionLog log) { this.log = log; }

    // One lock per partition: the check and the append must be atomic, or two copies of the
    // same retry could both pass the check.
    synchronized long append(Batch b) {
        ProducerState s = producers.get(b.producerId());
        if (s != null && b.epoch() < s.epoch) {
            // A newer instance with the same transactional id has started. This one is a zombie,
            // perhaps paused by GC, and must never write again.
            throw new ProducerFencedException("epoch " + b.epoch() + " < " + s.epoch);
        }
        if (s == null || b.epoch() > s.epoch) {
            // First batch from this producer, or from its new epoch: numbering restarts at 0.
            if (b.firstSeq() != 0) throw new OutOfOrderSequenceException("a new epoch starts at 0");
            s = new ProducerState(b.epoch());
            producers.put(b.producerId(), s);
        } else {
            for (Appended a : s.recent) {
                // A retry of a batch we already wrote: the ack was lost, not the data.
                if (a.firstSeq() == b.firstSeq() && a.lastSeq() == b.lastSeq()) return a.baseOffset();
            }
            if (b.firstSeq() != s.lastSeq() + 1) {
                // A gap means an earlier batch never landed. Writing this one would reorder the
                // partition, so refuse it; the producer resends in order.
                throw new OutOfOrderSequenceException("expected " + (s.lastSeq() + 1) + ", got " + b.firstSeq());
            }
        }
        long base = log.append(b.records());
        s.recent.addLast(new Appended(b.firstSeq(), b.lastSeq(), base));
        if (s.recent.size() > CACHED_BATCHES) s.recent.removeFirst();
        return base;
    }
}

A producer ID lasts only as long as the process. Transactions add a stable identity: with a transactional.id, a restarted producer keeps its identity and bumps the epoch, which fences the old instance. A transaction coordinator writes commit or abort markers into every partition touched. Consumers reading in read_committed mode see only committed transactions.

Read, process, write as one transaction

Scroll or pinch to zoom, drag to pan

Color key
Service
Queue or stream

A crash before the commit aborts the transaction: the output stays invisible, the input offsets never moved, and the replacement redoes the work once. That’s exactly once from input topic to output topic, and nothing more.

Why a log on disk is fast

Every message is written to disk. How is that fast enough for 3 GB/s?

Disks are fast at sequential access, and this design does nothing else. Five mechanisms carry it:

Mechanism	What it does	Where it stops working
Sequential appends	Each partition writes only at the end of one file; no seeks	Thousands of partitions per disk become random IO
Page cache	Writes land in the operating system’s memory; recent reads come from there	A replay of old data pushes recent data out
Batching	Producers, brokers and consumers move whole compressed batches	Low traffic with no wait sends tiny batches
Zero-copy	The kernel copies file bytes straight to the network	Encryption: the data must pass through the program again
Sparse index	One index entry per 4 KB maps an offset to a byte position	Practically never: a 1 GB segment’s index is about 2 MB

Go deeperRetention and compaction1 min

Retention deletes whole segments, once a segment’s newest message is older than the retention time. No single message is ever deleted. Compaction serves topics that hold state, such as the offsets topic. A background cleaner rewrites old segments, and keeps only the newest record per key. A record with an empty value deletes its key. It’s dropped itself after 24 hours, so slow consumers still see the delete. Offsets never change. The log just gets gaps.

Deploys that stall consumers

Every time the billing consumers are deployed, lag spikes for a minute. Why, and how do we stop it?

A rolling deploy of 60 instances triggers a rebalance each time one leaves or joins. With the original protocol, every member gives up every partition until all rejoin: 120 full stops per deploy.

The fix is two settings. Cooperative rebalancing revokes only the partitions that actually move, and the rest keep consuming. Static membership gives each instance a stable ID. So a restart within the session timeout causes no rebalance at all.

The other cause is slow processing. A member that doesn’t poll for 5 minutes is removed from the group. 500 records at a second each breaks that. The new owner replays the same slow batch, and the group storms. Poll fewer records, or pause the partition while a worker pool catches up.

Lag is the high watermark minus the committed offset. Alert on it in time, not messages: 10,000 behind is nothing at 50,000 a second, and an hour at 3 a second. If one partition’s lag keeps growing while others are fine, it’s almost always a hot key. Adding consumers doesn’t help, because only one member reads a partition.

Go deeperRebalance protocols, and backpressure1 min

Approach	What changes	Cost
Eager rebalance	Everyone gives up everything, then gets a new assignment	The whole group stops each time
Cooperative	Only partitions that move are revoked	Two rounds instead of one for a moved partition
Static membership	A restart within the timeout causes no rebalance	Its partitions wait for it during the restart
Broker-side assignment	The coordinator computes assignments; members sync by heartbeat	A newer protocol clients must support

Backpressure is mostly free in a pull design. A slow consumer falls behind, and the log holds the difference. On the produce side, brokers delay responses to clients over their byte quota. And a producer whose buffer fills blocks in send, which pushes the pressure back into the application.

Tiered storage, and ten times bigger

Seven days of history needs about 130 brokers for disk, but traffic needs only 40. And at ten times, would we still run one cluster?

Live consumers read only the last few seconds. Old data serves replays. So when a segment closes, the leader uploads it to object storage and keeps only the last 12 hours locally: about 3.2 TB per broker. Object storage is durable on its own, so history is kept once, not three times. The broker count follows traffic again, 40 not 130. Rebuilding a broker copies 12 hours, not 7 days. And a replay of old data no longer pushes recent data out of memory, the usual cause of latency spikes.

Distributed message queue · architecture

Stage 5: Tiered storage

Color keyServiceDatabaseQueue or streamBackground workerNew in this stageAsync or occasional

Stage

Messages produced per second: 3,000,000

Stage 5 of 6

Tiered storage

Why this stage

Seven days at 1 GB/s with three replicas is 1.8 PB.

Throughput needs about 40 brokers, but their disks hold only about 14 TB each, so storage would force 130.

Data size, not load, forces this stage.

What we add

Brokers keep about 12 hours of segments locally and upload closed segments to object storage, which serves older reads.

What it costs

Reads of old data take tens of milliseconds instead of one, and the broker now depends on an object store for history.

At 3M messages produced per second, every part has headroom. Raise the traffic until something turns red.

Pinch or tap the zoom buttons to read small labels.Drag any box to move it. Reset puts every box back.

The numbers behind the busy bars

What the traffic is

Messages are 1 KB after compression.
Three consumer groups each read every message.
From stage 4, every message is written 3 times, and followers fetch it twice more.

What one machine can do

One broker appends about 300,000 messages a second (300 MB/s) and serves about 900,000 fetched messages from page cache.
In a cluster, one broker moves about 1 GB a second in total: appends, replica fetches and consumer fetches.
One consumer process handles about 50,000 messages a second of light processing.

Headroom

Instance counts keep each box near 60% busy. That leaves room to take over a failed broker's partitions.

Follow a consumer catching up from old data:

Distributed message queue · architecture

Trace a request: Replay three days back

Color keyServiceDatabaseQueue or streamBackground workerNew in this stageAsync or occasionalHop in motionHops doneResponse

Replay three days back: step 1 of 4

Consumer groups to Brokers

A new group, built to backfill a search index, asks for partition 7 from three days ago. It first maps a timestamp to an offset with the time index, then fetches from that offset.

1 ms of a 100 ms budget

Consumer groups → Brokers1 ms
Brokers → Log segments0.2 ms
Brokers → Object storage40 ms
Response

Pinch or tap the zoom buttons to read small labels.Drag any box to move it. Reset puts every box back.

At ten times, one cluster would be 400 brokers and 300,000 partitions. The data path copes, but operations don’t. A rolling restart takes a day and a half, and every bad config reaches every team. So we run cells. A cell is an independent cluster of at most about 100 brokers, serving a fixed set of teams. A small directory says which cell holds a topic. A mirror copies critical topics to a second region, asynchronously, and records how offsets map between them.

Distributed message queue · architecture

Stage 6: Cells and a second region

Color keyServiceDatabaseQueue or streamBackground workerNew in this stageAsync or occasional

Stage

Messages produced per second: 3,000,000

Stage 6 of 6

Cells and a second region

Why this stage

At 10 times the load one cluster would be 400 brokers and hundreds of thousands of partitions: a rolling restart takes days, and one bad config or controller bug stops every team.

One region is also one failure domain.

What we add

Several independent clusters, each serving a group of teams, and an asynchronous mirror of critical topics to a standby cluster in a second region.

What it costs

Clients must know which cluster holds a topic, and a regional failover loses the last seconds of mirrored data and changes offsets.

At 3M messages produced per second, every part has headroom. Raise the traffic until something turns red.

Pinch or tap the zoom buttons to read small labels.Drag any box to move it. Reset puts every box back.

The numbers behind the busy bars

What the traffic is

Messages are 1 KB after compression.
Three consumer groups each read every message.
From stage 4, every message is written 3 times, and followers fetch it twice more.

What one machine can do

One broker appends about 300,000 messages a second (300 MB/s) and serves about 900,000 fetched messages from page cache.
In a cluster, one broker moves about 1 GB a second in total: appends, replica fetches and consumer fetches.
One consumer process handles about 50,000 messages a second of light processing.

Headroom

Instance counts keep each box near 60% busy. That leaves room to take over a failed broker's partitions.

Go deeperCells, stretch clusters and mirrors compared1 min

Option	What it buys	What it costs
Cells by team	Failures and noisy teams stay in one cell	A directory; a topic shared by two cells must be mirrored
A cluster across regions	Synchronous durability across regions	Every produce pays a cross-region round trip, 30 to 80 ms
Async mirror to a region B	Survives a region loss; produce latency unchanged	Loses the last seconds on failover; offsets differ between them

Reads of old data from object storage take about 40 ms to the first byte. The cluster depends on the object store being up for history. And compacted topics stay fully local.

When things break

Interviewerminute 40

The leader of a busy partition dies at peak. What do producers and consumers see?

What they are testing

Whether a caught-up replica takes over in seconds, and nothing acknowledged is lost.

A partition leader dies. The controller notices within seconds. It picks a new leader from the in-sync replica set, so the new leader has every acknowledged message. Producers get errors for a few seconds, refresh the partition map, and retry. With idempotence on, a retried batch is not written twice. Consumers reconnect to the new leader, and carry on from their committed offset. A follower that had copied past the high watermark cuts its log back to it. So no replica keeps a message that was never acknowledged.

A whole zone goes dark. Every partition has one replica per zone, so losing a zone leaves two of three. About 10,000 partitions need new leaders, chosen from their ISRs within seconds. Producers see a few seconds of retries, and no acknowledged message is lost. The other brokers absorb 50% more traffic, which is why they run near 60%. When the zone returns, its catch-up must be throttled.

Click any box to break it and see what users would notice:

Distributed message queue · architecture

Break it: what fails, who notices, what users see

Color keyServiceDatabaseQueue or streamBackground workerNew in this stageAsync or occasional

Break

Failure: click any box to break it

A broker goes down

What breaks

It leads about 1/40 of all partitions.

Producers and consumers of those partitions get errors until new leaders are elected.

How we notice

Its heartbeats to the active controller stop; the controller fences it after broker.session.timeout.ms, 9 seconds by default.

What users see

Producers retry for a few seconds and lose nothing: idempotence removes duplicates from the retries.

Consumers pause on those partitions.

What we do

The controller picks a new leader from each partition’s in-sync replicas and publishes the change in the metadata log.

Clients refresh metadata and carry on.

The replacement broker catches up from leaders.

Pinch or tap the zoom buttons to read small labels.Drag any box to move it. Reset puts every box back.

Correlated failure is the real risk: a bad config on every broker at once, or disks filling everywhere during a long object storage outage. So changes go out one zone and one cell at a time.

Go deeperEvery failure, and what users see1 min

Part	Failure	What users see	What we do
Broker	Crash	Seconds of produce retries on its partitions; no loss	New leaders from the ISR; retries are idempotent
Follower	Slow or dead	Produce p99 rises for up to 30 s	Dropped from the ISR; two still in sync
Disk	Fails	Partitions on it fail over	Replace; copy 12 hours back, not 7 days
Controller	Crash	Metadata changes pause about a second	A standby with the full metadata log takes over
Consumer	Crash	Its partitions pause; up to 5 s replayed	Rebalance; idempotent processing
Object storage	Outage	Replays older than 12 hours fail; live traffic fine	Segments stay local until uploaded; watch disk space
Region	Outage	Mirrored topics fail over after a decision	Translate offsets; lose seconds of data

Operation	Guarantee	What users can see
Produce with acks=all	On every in-sync replica, at least two zones	A slow follower slows produce for up to 30 s
Order within a partition	Total order	Different keys can be reordered relative to each other
Consumer reads	Only below the high watermark	A message is visible a few ms after its acknowledgement
Delivery to a group	At least once	Up to 5 s of duplicates after a crash or rebalance

Two consumers in one group can be minutes apart. Say the two events used different keys. Then “order 42 shipped” in one partition can be processed before “order 42 created”.

Go deeperRunning it: alerts, deploys and cost1 min

Wake someone up when:

produce success drops below 99.9% in a cell
produce p99 passes 50 ms
partitions sit under two in-sync replicas for over a minute (they’re refusing writes)
any partition has no leader
a critical group’s lag, in seconds, passes its target

Deploys roll one broker at a time, waiting until every partition is fully copied again. A stopping broker hands off its leaderships first, so clients see a leader change, not a timeout. Moving data to new brokers is throttled: add a replica, wait for it to join the ISR, drop the old one.

Cost. The largest line is often traffic between zones: about 6.9 PB a month, over $100,000. That already assumes consumers fetch from a replica in their own zone. Compression (zstd on 64 KB batches cuts JSON three to five times) and tiered storage are the other levers.

Tenancy. Clients authenticate with mutual TLS or SASL. Topics carry per-team access lists. Byte quotas slow a runaway producer down, instead of letting it crash a broker. Protocol changes ship to brokers first and clients after, because old clients must keep working against new brokers.

Wrap up

The whole design

Here is everything we built, in the order we built it. Use the buttons to step through it.

Distributed message queue · architecture

The full design, stage by stage

Color keyServiceDatabaseQueue or streamBackground workerNew in this stageAsync or occasional

Stage

Messages produced per second: 3,000,000

Stage 6 of 6

Cells and a second region

Why this stage

At 10 times the load one cluster would be 400 brokers and hundreds of thousands of partitions: a rolling restart takes days, and one bad config or controller bug stops every team.

One region is also one failure domain.

What we add

Several independent clusters, each serving a group of teams, and an asynchronous mirror of critical topics to a standby cluster in a second region.

What it costs

Clients must know which cluster holds a topic, and a regional failover loses the last seconds of mirrored data and changes offsets.

At 3M messages produced per second, every part has headroom. Raise the traffic until something turns red.

Pinch or tap the zoom buttons to read small labels.Drag any box to move it. Reset puts every box back.The whole design. Step through it: one broker and one log, partitions across brokers, consumer groups and offsets, replicas and leaders, tiered storage, then cells and a second region.

The numbers behind the busy bars

What the traffic is

Messages are 1 KB after compression.
Three consumer groups each read every message.
From stage 4, every message is written 3 times, and followers fetch it twice more.

What one machine can do

One broker appends about 300,000 messages a second (300 MB/s) and serves about 900,000 fetched messages from page cache.
In a cluster, one broker moves about 1 GB a second in total: appends, replica fetches and consumer fetches.
One consumer process handles about 50,000 messages a second of light processing.

Headroom

Instance counts keep each box near 60% busy. That leaves room to take over a failed broker's partitions.

The decisions, and what each one costs

Decision	We chose	We gave up	We’d choose differently if
Storage model	Append-only log, consumers track offsets	Per-message acks, delays, deletes	It were a task queue with per-message retries
Ordering	Per partition, by key	Global order	Volume fit one partition and global order were required
Replication	Leader plus ISR, acks=all, at least two in sync	Writes when only one replica is in sync	Losing data were acceptable for availability
Metadata	A Raft controller quorum	A simpler outside dependency	We already ran ZooKeeper for older versions
Delivery	At least once, transactions for pipelines	Exactly once into other systems	Every consumer wrote back to the log
Storage tiers	12 hours local, history in object storage	Fast reads of old data	Retention were a day or two
Scale out	Cells of about 100 brokers, an async mirror	One global namespace, zero-loss region failover	One team owned all topics

How to spend 45 minutes

0 to 4. Queue or log first. Then throughput in bytes, order per key, and what “acknowledged” must survive.
4 to 8. 3 GB/s at peak, 24 GB/s of broker traffic and 40 brokers, 1.8 PB at 7 days.
8 to 20. Build it: the log, partitions, consumer groups, replication. A mid-level answer can stop here.
20 to 30. The stress tests: what “acknowledged” means, then exactly once.
30 to 40. Disk speed, rebalances, then tiered storage and cells.
40 to 45. A leader failing, then a zone, then your own questions.

Never skip the ISR and high watermark, the commit loop, or a precise answer to “exactly once”.

Follow-up questions

Interviewer asksA consumer must write each message to Postgres exactly once. How?

Store the consumed offset in the same Postgres transaction as the write, and on startup read from the offset stored there. A crash commits both or neither. The alternative is a write keyed by a message ID under a unique constraint.

Interviewer asksProduct wants per-message retries with delays, like a job queue. Where does that go?

On top of the log. A failed message goes to a retry topic per delay (1 minute, 10 minutes, 1 hour), then to a dead letter topic, and the original is committed. Order per key is lost for retried messages. If most traffic needs this, a work queue such as SQS is the better tool.

Go deeperMore follow-up questions1 min

A team wants to send 50 MB video files through the queue. Store the file in blob storage and send its key and checksum. Large messages stall a partition’s consumer, fill buffers and make replication bursty.

Why not just add partitions whenever a topic gets busy? Because hash(key) mod count changes, and most keys move. New messages for a key land in a new partition. Older ones may still be unread in the old one, so they can be processed out of order. Over-provision at creation, or move to a new topic once the old one drains.

A producer pauses for 40 seconds, its replacement starts, then the old one wakes up. What stops duplicate writes? With a transactional ID, the replacement bumps the epoch. Leaders then reject the old producer as fenced. That’s the first check in the idempotence code. Without one, the two have unrelated producer IDs, and the old batches can land.

Design a distributed message queue

The answer in one minute

Scope it

The numbers that decide it

Build the design

Step 1. One broker, one log

Stage 1: One broker, one log

Why this stage

What we add

What it costs

Step 2. Partitions, by key

Stage 2: Partitions across brokers

Why this stage

What we add

What it costs

Step 3. Consumer groups and offsets

Stage 3: Consumer groups and offsets

Why this stage

What we add

What it costs

Step 4. Three replicas, one leader

Stage 4: Replicate and elect leaders

Why this stage

What we add

What it costs

Trace a request: Produce with acks=all

Stress it

What “acknowledged” means

At least once, and what exactly once really means

Why a log on disk is fast

Deploys that stall consumers

Tiered storage, and ten times bigger

Stage 5: Tiered storage

Why this stage

What we add

What it costs

Trace a request: Replay three days back

Stage 6: Cells and a second region

Why this stage

What we add

What it costs

When things break

Break it: what fails, who notices, what users see

What breaks

How we notice

What users see

What we do

Wrap up

The whole design

The full design, stage by stage

Why this stage

What we add

What it costs

The decisions, and what each one costs

How to spend 45 minutes

Follow-up questions

Further reading

Keep reading

The designbeyondcode digest, every Saturday