Design a Distributed Message Queue (like Kafka)

1Clarifying the requirements#

Before I draw a single box, I want to pin down what kind of queue this is. "Message queue" means five different products depending on who is asking, and getting this wrong costs you the entire design.

Candidate: Are we building a work queue like SQS where a message is consumed once and deleted, or a log like Kafka where messages persist and multiple independent consumers each read the whole stream?

Interviewer: The log model. Think Kafka. Consumers should be able to replay.

Candidate: What ordering guarantee do we owe? Global order across the whole topic, or per-key order?

Interviewer: Per-key is enough. All events for one user should arrive in order. Events across different users can interleave.

Candidate: Delivery semantics? At-least-once is cheap. Exactly-once is expensive and I want to know if we actually need it.

Interviewer: Default to at-least-once. Call out how you would offer exactly-once for the teams that pay for it.

Candidate: Retention? And roughly how big is the fleet?

Interviewer: 7-day default retention, configurable. Assume a large internal platform, thousands of topics.

What those answers just told me:

• "Per-key is enough" is the most freeing answer I could get. Global order would force a single partition and kill parallelism; per-key order is the exact hook that lets me shard.
• "Default at-least-once, offer exactly-once" means duplicates are acceptable and the exactly-once path is an opt-in product, not a baseline.
• "Thousands of topics, 7-day retention" sets the storage and metadata scale before I compute a single number.

Functional requirements

• Producers publish messages to a named topic; each message can carry a partition key.
• A topic is split into partitions; order is guaranteed within a partition, not across.
• Consumers read via consumer groups; each partition is owned by exactly one consumer in a group at a time.
• Consumers commit offsets so they resume where they left off, and can seek backward to replay.
• Messages persist for a retention window regardless of whether they have been read.

Non-functional requirements

• High write throughput, millions of messages/sec at the cluster level.
• Durability: an acknowledged write survives the loss of a single broker, ideally two.
• Low end-to-end latency, p99 under about 50 ms for a produce-to-consume hop.
• Horizontal scale by adding brokers and partitions.
• Availability over strict consistency on the read path; a brief stale read beats an outage.

2Back-of-the-envelope estimation#

I only need one number to shape this design, so let me chase it directly: sustained write bandwidth, and what it turns into on disk.

Write bandwidth.

• Say the platform carries 2 million messages/sec at peak, average message 1 KB including headers.
• That is 2 GB/sec of ingest. Writes dominate here, which is unusual and important.

Read amplification.

• In a fan-out log every consumer group re-reads the whole stream, so reads are a multiple of writes.
• Assume 3x read amplification, so about 6 GB/sec read.
• Served mostly from page cache, because consumers tail the head of the log where the hot bytes already live in RAM.

Storage (the number that bites).

• 2 GB/sec times 86,400 seconds/day is about 173 TB/day of raw ingest.
• With replication factor 3, that is about 520 TB/day on disk.
• Over a 7-day retention window, about 3.6 PB live at any moment.

Fleet size.

• A modern box with NVMe and a 25 Gbps NIC comfortably does about 150 to 200 MB/sec of durable sequential writes once you count replication overhead.
• To absorb 2 GB/sec of primary writes plus the replication traffic on top, I need roughly 40 to 60 brokers for headroom.
• Call it a 60-broker cluster with room to grow and to survive a lost rack.

Partition count.

• If one partition sustains about 10 MB/sec write, I need about 200 partitions for a hot topic to hit 2 GB/sec.
• Partition count is the unit of parallelism, so I over-provision it on purpose.
• I will defend exactly why that over-provisioning is non-negotiable in the deep dive, because it is the operational trap that bites real teams.

3Core APIs#

Thin surface on purpose. The cleverness lives in the storage layer, not the API, and an interviewer who sees you spend ten minutes on REST verbs knows you have buried the lede.

POST /v1/topics/{topic}/produce
Content-Type: application/json

{
  "key": "user-8841",
  "value": "<base64 payload>",
  "acks": "all",
  "idempotency_key": "req-5f3a9c"
}

200 OK
{ "partition": 17, "offset": 884213, "timestamp": 1730600000 }

GET /v1/topics/{topic}/consume?group=billing&max_bytes=1048576
-> returns a batch of records with partition and offset per record

POST /v1/topics/{topic}/commit
{ "group": "billing", "offsets": { "17": 884213, "18": 991002 } }

POST /v1/topics/{topic}/seek
{ "group": "billing", "partition": 17, "offset": 500000 }

Two fields do almost all the hard work here:

• acks: all carries the durability story: the write is not acknowledged until the in-sync replicas have it.
• idempotency_key carries the exactly-once story: a retried produce with the same key dedupes instead of appending twice.
• Everything else is plumbing. I defend both of these below, because if you cannot explain what they actually guarantee, they are decoration.

4High-level design#

The spine is a partitioned, replicated commit log.

• Producers hash the key to pick a partition and send to that partition's leader broker.
• The leader appends to its local log, replicates to followers, then acks.
• Consumers in a group are assigned partitions, pull batches, and track their own offsets.
• Consumers pull, they are not pushed to, because pull lets a slow consumer set its own pace instead of drowning.

I need a control plane separate from the data plane.

• Someone has to decide which broker leads partition 17, detect failures, and trigger leader elections.
• Old Kafka used ZooKeeper; modern Kafka uses a built-in Raft quorum called KRaft.
• I use a small Raft-based controller quorum of 3 or 5 nodes holding cluster metadata: topic configs, partition-to-broker assignments, and the in-sync replica set per partition.
• The key move: keep this quorum off the data path. It touches membership and failure events only, never individual messages, so the firehose never goes near the one strongly-consistent component.

Why it is laid out this way.

• The partitioner lives client-side, so producers talk straight to the right leader with no proxy hop. A proxy would add a whole network round trip, and you cannot afford that under a 50 ms p99.
• Consumers also talk directly to leaders.
• Offsets are stored as just another internal topic: a compacted log keyed by group plus partition. The system eats its own dog food, and offset storage inherits the exact same replication guarantees as user data. One storage engine, not two.

5Data model#

The on-disk log is the real data model, and this is the part I want concrete, because it is what makes a log cheap.

How a partition is laid out.

• Each partition is an append-only sequence of segment files plus a sparse offset index.
• Writes go to the tail of the active segment, sequentially, so the disk head never seeks.
• Reads by offset binary-search the sparse index to find the right segment and byte position.

The record (what lives in the log):

The metadata (what lives in the controller):

SQL or NoSQL? Neither, mostly, and I would push back on the framing.

• The message store is a purpose-built append-only log on local disk, because the access pattern is sequential write and sequential read by offset.
• A B-tree database would be actively wrong here: it buys random-access lookups I never make, and pays for it with write amplification and seeks I cannot afford at 2 GB/sec.
• I want sequential I/O and the OS page cache doing the work for free.
• Cluster metadata is tiny but needs strong consistency and leader election, so that is a replicated state machine via Raft, not a general database.
• Consumer offsets go in a compacted log topic, because I only ever need the latest offset per group per partition, and compaction keeps just the newest value per key, which is exactly that.

6Deep dives#

6.1 Replication, ISR, and the ack trade-off (the signature hard part)#

This is where the interview is won or lost, so I slow down.

The setup.

• A write to partition 17 goes to the leader.
• Followers continuously fetch from the leader to catch up, using the exact same fetch path a consumer uses. A follower is just a special consumer, which is a nice simplification.
• The set of followers caught up within a bound is the in-sync replica set, the ISR.

The whole game is one question: when does the leader ack the producer? That is the acks knob, and each setting is a different durability contract:

• acks=0: fire and forget. Fast, loses data on any blip. Fine for metrics, never for payments.
• acks=1: leader writes locally and acks. If the leader dies before a follower copies that record, the record is gone. Silent data loss, the worst kind, because nobody gets an error.
• acks=all: leader acks only after all ISR members have the record. Survives leader loss as long as one ISR member survives.

The subtle part is min.insync.replicas. This is what separates people who read a blog post from people who have run this.

• Say replication factor is 3 and I set min.insync.replicas=2.
• The leader refuses the write if the ISR shrinks below 2.
• So I can lose one broker and keep accepting writes with full durability.
• If I lose two, the partition goes read-only rather than silently accepting writes it cannot protect.
• That is a deliberate consistency-over-availability call. A producer getting a retriable error for a few seconds is recoverable. A payments event that was acked and then vanished is not.

Two more things that mark a senior answer.

• High watermark. Consumers can only read up to the highest offset replicated to all ISR members, not the leader's raw tip. This is what stops a consumer from reading a record that then vanishes in a failover.
• Leader epoch fencing. When a leader dies and a new one is elected, the controller bumps the epoch. A zombie old leader that comes back from a network partition and tries to accept writes gets rejected because its epoch is stale. Without this you get log divergence, two "leaders" accepting different records at the same offset. That is the classic split-brain corruption bug, and naming it unprompted relaxes the interviewer.

Failover flow.

• The controller notices the leader's heartbeat lapsed.
• It picks a new leader from the ISR only, never from a lagging replica, because a lagging replica would silently drop committed data.
• It bumps the epoch and pushes new metadata to clients.
• Clients refresh and redirect. Downtime is the detection window plus election, typically a few hundred ms.

6.2 Partitioning, consumer groups, and offset management#

Ordering lives and dies by partitioning.

• I hash the message key to a partition: partition = murmur2(key) % numPartitions.
• Same key, same partition, same ordered log.
• No key means round-robin, which gives throughput but zero order guarantee. That is a fine trade for stateless events as long as you said so on purpose.

The operational trap that bites real teams: you cannot cheaply increase partition count after the fact.

• Adding partitions changes the modulo, so user-8841 that was on partition 17 might map to partition 23 going forward, while its old events stay on 17.
• Per-key order breaks across the resize, and it breaks silently, which is the dangerous part.
• So I over-provision partitions up front, that 200 number from the estimate, rather than resharding live.
• If I truly must grow, I create a new topic and migrate; I do not rehash in place.
• This is the single most common production regret with these systems, and naming it up front shows you have felt the pain.

Consumer groups give the parallelism.

• Within a group, each partition is owned by exactly one consumer, so order is preserved per partition.
• With 200 partitions and 50 consumers, each consumer owns about 4 partitions.
• Add consumers up to 200 and each owns one; beyond 200 they sit idle, because there is nothing left to assign.
• Partition count caps consumer parallelism. That is a second independent reason to over-provision: not just write throughput, but how many consumers can ever work in parallel.

Rebalancing is the hard operational edge.

• When a consumer joins, leaves, or dies, the group must reassign partitions.
• The naive stop-the-world rebalance pauses every consumer in the group while reassigning, a latency cliff you feel across the whole pipeline.
• I use incremental cooperative rebalancing: only the partitions that actually move get revoked, the rest keep flowing.
• A group coordinator, one broker elected per group, runs this.

Offset management is where at-least-once vs exactly-once gets real. It comes down to one ordering decision: commit before you process, or after.

• Commit the offset before processing, then crash: that record is skipped. At-most-once, data loss.
• Commit after processing, then crash in the gap between processing and commit: that record is reprocessed on restart. At-least-once, duplicates.
• At-least-once is the sane default. The cost is that downstream must tolerate duplicates, a real tax but a cheap and well-understood one.

6.3 Exactly-once: idempotent producers plus transactions#

"Exactly-once" is really "effectively once," built from two mechanisms, and I want to be precise, because hand-waving here is a red flag an interviewer will poke at immediately.

Producer idempotence kills duplicates from producer retries.

• Every producer gets a producer ID; each record carries a monotonic sequence number per partition.
• If a producer's ack gets lost on the network and it retries, the leader sees a sequence number it already has and dedupes it instead of appending twice.
• That is what the idempotency_key in my API maps to under the hood.
• This solves exactly one class of duplicate, the retry, and it is worth saying it only solves that one.

Transactions solve the consume-process-produce loop, the thing a stream processor does all day.

• The processor reads from topic A, computes, writes to topic B, and commits its input offset, and we want all-or-nothing across those.
• The design: a transaction coordinator, another internal log, tracks the transaction.
• Writes to B and the offset commit to A's offset topic are tagged with the transaction ID and written as uncommitted.
• On commit, the coordinator writes a commit marker to every involved partition.
• Consumers configured as read_committed skip any records inside an aborted or still-open transaction, so they never see a half-finished unit of work.

The honest trade-off I say out loud, because the interviewer is waiting for it:

• Exactly-once roughly doubles coordination overhead and adds latency from the commit markers and the two-phase dance.
• It only works inside the Kafka boundary. The moment your side effect is an external API call or a non-transactional database, "exactly-once" degrades right back to "at-least-once plus an idempotency key on your side."
• So I offer transactions as an opt-in for stream jobs that live entirely inside the log, and tell everyone else: make your consumer idempotent with a dedup key and sleep well. That is the grown-up answer.

7Bottlenecks and trade-offs#

What breaks first at scale: the controller.

• It is the one strongly-consistent component, and if metadata churn gets heavy (think thousands of partitions failing over at once during a rack outage), election storms can stall the whole cluster's write path.
• Mitigation: keep the controller quorum small and fast, cap partitions per broker, and make clients back off and cache metadata so a controller hiccup does not snowball into a thundering herd of refresh requests.

The other early bottleneck: a hot partition.

• One celebrity key, say a single giant tenant, concentrates on one partition and one leader, and you cannot split it without breaking that key's order.
• There is no clean fix inside the model.
• The answer is a key-design conversation with producers, or a salted key scheme that trades order granularity for spread: append a bucket suffix so one tenant fans across four partitions and you re-merge order downstream.
• Say the quiet part: you are trading some ordering to buy throughput.

Single points of failure.

• Per partition, the leader is a transient single point, mitigated by ISR failover.
• The controller quorum is the cluster-level one, mitigated by Raft, which survives minority loss.
• Disks are the sneaky one: a single bad NVMe can corrupt a log segment, which is exactly why every record carries a CRC checked on read and why replication factor 3 exists.
• One of those three defenses catches each failure mode.

The consistency call I made on purpose.

• For durable topics I chose acks=all with min.insync.replicas=2, so a double-broker loss on a partition makes it reject writes rather than accept un-protectable ones.
• I would rather a producer get a retriable error for a few seconds than tell a payments team their event is safe when it is not.
• For lossy telemetry topics I would flip to acks=1 and take the throughput.
• Same cluster, different per-topic durability, and I make that a first-class config, not a cluster-wide compromise.

Cost vs retention.

• Holding 3.6 PB hot is expensive, and that cost scales linearly with every day of retention you add.
• The move at scale is tiered storage: recent segments on local NVMe for low-latency tailing reads, older segments offloaded to object storage (S3-class) and fetched on demand for replay.
• Retention becomes cheap and nearly unbounded, at the cost of slow cold reads for old data.
• For a log where 95% of reads hit the hot tail, that is a trade I take every time.

Let the access pattern pick the storage engine. Sequential-only is why a plain log beats a database here, not habit.

8Summary#

The defensible decisions, each tied back to a constraint:

• Partitioned append-only commit log for sequential I/O and page-cache reads, because 2 GB/sec kills any design with seeks.
• Per-key ordering via key hashing, with partitions over-provisioned up front, because resharding breaks order silently.
• A small Raft controller off the hot path owning metadata and leader election, because the firehose must never touch the strongly-consistent component.
• acks=all plus min.insync.replicas and a high watermark, so acknowledged writes genuinely survive failover.
• Leader-epoch fencing to kill split-brain before two leaders diverge the log.
• At-least-once as the default, with opt-in idempotent producers and transactions for the teams that truly need effectively-once.

If I had more time I would detail the tiered storage segment lifecycle, the cooperative rebalance protocol state machine, quotas to tame noisy producers, and a dead-letter-topic pattern for poison messages that keep crashing a consumer. I would also benchmark the real per-broker ceiling instead of trusting my 150 MB/sec guess, because that single number sets the whole fleet size, and if it is off by 2x my 60-broker cluster is either half-empty or on fire.

If you want to see how a payments team would actually consume this log with idempotent, duplicate-safe processing, pair this with our UPI and wallet system design guide.