Design a distributed message queue (Kafka)

A distributed message queue lets one service hand an event to many others without waiting for them: checkout says “order paid” once, and fraud checks, search, the warehouse and notifications each pick it up when ready. A Kafka-style system keeps those events as a durable, ordered log that every reader consumes at its own pace and can replay, so one week of UPI transaction events feeds a live fraud model and a 2 a.m. reconciliation job. The obvious design — a table of messages, deleted as they are read — stalls at a few thousand messages a second; a well-built log moves millions a second on ordinary disks. Getting there means rethinking storage, replication, consumer positions, what “exactly once” can honestly promise, and who decides which broker leads which partition.

Understand the problem

Say which kind of broker you are building. A classic queue — the model of brokers such as ActiveMQ and RabbitMQ — hands each message to one worker, tracks delivery state per message and deletes it once acknowledged. A log appends messages to an ordered, retained sequence and lets each reader keep its own position. This problem is the log: one mechanism gives you publish/subscribe and shared work, plus replay. The classic model is in message queues.

Functional requirements

1. Producers should be able to publish records to a topic — key, value up to 1 MB, headers, timestamp — and be acknowledged only once the record is durable.
2. Many consumer groups should each read every record at their own pace, in order per key, and rewind to an earlier offset or time (“replay since 20:00”).
3. Consumers in a group should share a topic’s work and, after a crash, deploy or scale-out, carry on where the group left off.
4. Operators should be able to create topics with a partition count, replication factor and retention policy: 7 days, 2 TB, or only the latest value per key.

Non-functional requirements

• Throughput. 3M records/s in at peak — the first hour of a festive sale, when order, payment and delivery events spike together — and 1M/s on average. Records average 1 KB after compression.
• Fan-out. 4 consumer groups per topic on average, so the cluster sends out about four times what it takes in.
• Latency. acks=all produces acknowledged in under 10 ms at p50 and 30 ms at p99; a record reaches a caught-up consumer in under 50 ms at p99.
• Durability. No acknowledged record is lost when one broker, or one availability zone, fails.
• Availability. 99.99% for producing; a broker crash makes its partitions unwritable for seconds, not minutes.
• Retention. 7 days by default; a consumer hours behind must still catch up.
• Ordering. Per key (per partition), not across a topic.
• Scale. About 2,000 topics and 24,000 partitions per cluster, with room to double.

Out of scope

• Stream processing — joins and windows run on top of the log; see Flink and stream processing.
• Schema management — see serialization and schema evolution.
• Cross-cluster mirroring, beyond one follow-up.
• Per-message routing, priorities and delays — a job scheduler or notification system builds these on a log.
• Security, except where it changes performance.

The set-up

Plan the approach

Name the shape early: “brokers that store partition logs, a small controller quorum that owns metadata, and smart clients that batch, choose partitions and track their own offsets.” Satisfy each requirement simply, then spend the time in the deep dives.

Core entities

• Topic. A named stream such as order-events, with partition count, replication factor and retention policy.
• Partition. An ordered, append-only log; the unit of ordering, parallelism and replication. One replica leads, the others follow.
• Record. Optional key, value bytes, headers, timestamp; stored and sent in batches.
• Offset. A 64-bit position in a partition, assigned by the leader. A record’s identity is (topic, partition, offset).
• Segment. A file holding a contiguous range of offsets, plus two small index files.
• Broker. A server that stores replicas and serves reads and writes.
• Consumer group. Consumers sharing a topic, with one committed offset per partition.
• Controller. Owns metadata: live brokers, partition leaders and in-sync replicas.

topic order-events · partition 1 · offset 18,442,019
key    order_9F3K2      every event for this order lands in partition 1
value  {"type":"ORDER_PAID","amount_paise":149900,"ts":"2026-10-27T20:00:03+05:30"}

API

On the wire this is a binary protocol over long-lived TCP connections, shown here as HTTP. Clients are “smart”: they find each partition’s leader and talk to it directly, with no routing tier.

Metadata. Any broker answers: which broker leads each partition, the replicas, and the in-sync set.

GET /v1/metadata?topics=order-events

{ "brokers": [ { "id": 2, "host": "b2.az-a.internal:9092", "rack": "az-a" } ],
  "topics": [ { "name": "order-events", "partitions": [
    { "partition": 1, "leader": 2, "leader_epoch": 17, "replicas": [2, 1, 3], "isr": [2, 1, 3] } ] } ] }

Produce. One request carries a batch per partition led by that broker. The producer ID and sequence number make retries safe (deep dive 4).

POST /v1/produce
Content-Type: application/json

{ "acks": "all", "timeout_ms": 30000,
  "producer_id": 4012, "producer_epoch": 0,
  "batches": [ { "topic": "order-events", "partition": 1, "base_sequence": 0,
                 "compression": "zstd", "record_count": 48, "records": "<compressed bytes>" } ] }

{ "results": [ { "topic": "order-events", "partition": 1, "base_offset": 18442019, "error": null } ],
  "throttle_time_ms": 0 }

Fetch. Consumers and follower replicas pull from an offset. The broker long-polls until data arrives or max_wait_ms passes, and never returns records beyond the high watermark.

GET /v1/fetch?topic=order-events&partition=1&offset=18442019&max_bytes=1048576&max_wait_ms=500

{ "high_watermark": 18442120, "last_stable_offset": 18442120, "records": "<batches, still compressed>" }

Groups and offsets. Members join a group, heartbeat, and commit how far they got.

POST /v1/groups/search-indexer/offsets
Content-Type: application/json

{ "member_id": "indexer-7", "generation": 42,
  "offsets": [ { "topic": "order-events", "partition": 1, "offset": 18442120 } ] }

Admin. Creating a topic goes to the controller.

POST /v1/topics
Content-Type: application/json

{ "name": "order-events", "partitions": 120, "replication_factor": 3,
  "configs": { "min.insync.replicas": "2", "retention.ms": "604800000", "cleanup.policy": "delete" } }

Back-of-envelope estimates

These numbers decide the broker count and which resource runs out first (method: back-of-envelope estimation).

Ingress. 3M records/s × 1 KB = 3 GB/s at peak (24 Gbps); 1M/s × 1 KB = 1 GB/s on average.

Replication and disk. Three replicas: every byte goes to two followers and three disks. Peak replication traffic is 2 × 3 GB/s = 6 GB/s; peak disk writes are 3 × 3 GB/s = 9 GB/s.

Fan-out. 4 groups × 3 GB/s = 12 GB/s to consumers at peak. Add replication and brokers send 12 + 6 = 18 GB/s while receiving 3 + 6 = 9 GB/s. Egress, not ingest, is the big number.

Storage. 1 GB/s × 86,400 s = 86.4 TB a day; × 7 days = 604.8 TB; × 3 replicas = 1,814 TB, about 1.8 PB on disk.

Brokers. Take a broker with 12 × 6 TB hard disks (72 TB raw), 128 GB of RAM and a 25 Gbps NIC, and keep disks about 60% full: about 43 TB each, so 1,814 ÷ 43 ≈ 42 brokers, 14 per zone. Check the other limits per broker at peak. In: 9 GB/s ÷ 42 ≈ 214 MB/s (1.7 Gbps, 7% of the NIC). Out: 18 GB/s ÷ 42 ≈ 429 MB/s (3.4 Gbps, 14%). Disk: 214 MB/s against about 1.6 GB/s of sequential bandwidth from 12 drives (a 2014 LinkedIn benchmark measured 822 MB/s from six). Storage sets the broker count; the network is mostly idle — remember that for tiered storage.

Page cache. About 100 GB of RAM per broker is page cache: 100 GB ÷ 214 MB/s ≈ 470 s, so it holds about the newest 8 minutes. A consumer less than 8 minutes behind reads from memory.

Partitions. 2,000 topics × 12 partitions on average = 24,000 partitions; × 3 = 72,000 replicas; ÷ 42 = about 1,700 replicas per broker, about 570 of them leaders.

Consumer positions. 24,000 partitions × 4 groups = 96,000 offsets; committed every 5 s, that is 19,200 commits/s. Tracking every message instead would be 3M × 4 = 12M acknowledgements/s.

Latency. Up to 5 ms of producer linger, a sub-millisecond produce round trip, one cross-zone follower fetch (1–2 ms) and a page-cache append in microseconds: under 10 ms at p50. Kafka’s design notes put waiting for a commit “on the order of 10 ms”.

High-level design

Build it one requirement at a time; by the end it works in one cluster at moderate load.

Producers should be able to publish records durably

The core is tiny: a broker appends each record to a file and replies with its position. One file on one machine can neither take 3 GB/s nor survive a dead disk, so split each topic into partitions spread across brokers, each its own ordered, append-only log with offsets counting from 0.

The producer picks the partition. With a key, it hashes the key modulo the partition count, so every event for order_9F3K2 — created, paid, packed, delivered — lands in one partition, in order. Without a key, it fills one partition’s batch, then moves on. It asks any broker for metadata once, caches which broker leads each partition, and sends batches straight to the leaders, which append them and assign offsets.

For durability, each partition has three replicas in three zones; followers copy the leader, which acknowledges a batch once the in-sync copies have it. Deep dive 2 unpacks that sentence.

Many consumer groups should read every record at their own pace

Consumers pull: “partition 1, from offset 18,442,019, up to 1 MB”. Reading deletes nothing, so any number of groups read the same partition, each with its own position. The design notes give the reasons for pull: a slow consumer falls behind instead of being flooded, every fetch returns a big batch, and rewinding is just asking for an older offset. Pull’s weakness, busy-polling an idle partition, is fixed by a long poll of up to 500 ms. (See push vs pull architecture.)

Replay is free. If the search indexer shipped a bug at 20:00, fix it and move its position back to the first offset after 20:00 (a time index makes that lookup cheap); it reprocesses the evening and nobody else notices.

Consumers in a group should share work and resume after failures

A group splits a topic’s partitions among its members: each partition has exactly one owner in the group at a time. That keeps records in order, needs no locks between members, and makes a group’s progress in a partition a single number. A broker acting as group coordinator tracks membership by heartbeats and assigns partitions. Members commit their positions, which the coordinator stores in an internal topic, __consumer_offsets. If a member dies, others take its partitions and resume from the last committed offsets.

Committing after processing gives at-least-once delivery: if a member crashes before committing, the next owner reprocesses its last few records.

Operators should be able to create topics and set retention

Topic creation goes to the controller, which places replicas — three brokers in three zones, leaderships spread evenly — and records it in the cluster metadata; brokers learn the change and create the partition directories. Retention runs on each broker: once the newest record in a segment is older than 7 days, the whole file is deleted. A compacted topic keeps the latest record per key instead.

You now have a working cluster. The deep dives are about what breaks at 3M records a second.

Deep dives

How do we store messages so writes and reads are fast?

Bad: a table of messages. Insert each message into a B-tree-indexed table, update a status column on delivery and on acknowledgement, delete the row at the end. Every message costs several random I/Os, and random I/O is what disks do worst: Kafka’s design notes put six 7,200 rpm drives at about 600 MB/s of linear writes against about 100 KB/s of random ones, over 6,000 times apart. A B-tree also slows as it grows, so the broker degrades exactly when consumers lag and the backlog swells. The 2011 Kafka paper saw it in practice: ActiveMQ’s busiest thread spent most of its time maintaining a B-tree, with 144 bytes of overhead per message against Kafka’s 9. At 3M inserts and 12M acknowledgements a second, the table never gets off the ground.

Good: an append-only log per partition, in segments. Each partition is a directory of segment files. The broker appends only to the newest, active segment and rolls a new one at 1 GiB (the default segment.bytes) or after a week. Each file is named after its first offset, so an in-memory sorted list of these base offsets finds the file for any offset. A record’s ID is its offset; deleting old data means deleting a file. You gain sequential I/O, and a message’s cost no longer depends on how much is stored — the 2014 benchmark wrote a terabyte with no drop in throughput. You pay in flexibility: no deleting or acknowledging one message, no lookup except by offset or time, and order only within a partition.

To jump into a 1 GiB segment, keep a sparse index beside it: one entry per ~4 KB of log (index.interval.bytes defaults to 4,096), mapping an offset relative to the segment’s base to a byte position. That is 2^30 ÷ 4,096 = 262,144 entries; at two 4-byte integers each, 2 MiB — small enough to memory-map. A second index maps timestamps to offsets.

Great: batch, compress, lean on the page cache, and send with zero copy. The log removes seeks; four habits remove the remaining waste of small operations and copying.

• Batch everywhere. The producer fills a batch per partition until batch.size (16 KB) or linger.ms (5 ms by default since Kafka 4.0) is reached. In the 2011 paper, batches of 50 instead of 1 took one producer from 50,000 to 400,000 messages a second. The broker appends a batch in one write; followers and consumers fetch in large chunks.
• Compress batches, not messages. Most redundancy is between records — repeated JSON field names and city codes — so zstd or LZ4 over a batch beats compressing each record. Batches stay compressed on disk and on the way to consumers. You save disk, network and cache; producers and consumers pay CPU.
• Let the operating system cache. Write into the page cache and skip per-message fsync. The kernel uses all free memory as cache (28–30 GB on a 32 GB machine, say the design notes), with no copy in the broker’s heap, no garbage-collection pressure, and a cache that survives a broker restart. Durability comes from replicas on other machines, not from fsync, which the notes say can cost two to three orders of magnitude. After a crash, the broker checks record CRCs at the tail of its newest segment, truncates anything torn and re-syncs from the leader.
• Send with zero copy. Producer, broker and consumer share one binary batch format, so the broker never looks inside the bytes it serves and can hand a file range straight to the socket with sendfile.

Four consumer groups reading the same bytes means one trip into the page cache and four out, never through the broker’s heap. One catch: TLS encrypts in user space, so with TLS Kafka cannot use sendfile and pays the copies plus encryption in CPU.

Why consumers track offsets, not the broker. Per-message delivery state would mean 12M updates a second at peak and an answer to “sent, never acknowledged”. With one owner per partition per group, progress is one integer per partition — 96,000 for the cluster. State is tiny, replay is moving the integer back, and a new consumer group costs no storage. The price: the broker cannot delete “once everyone has read it”, so it deletes by time or size, and a consumer more than 7 days behind must reset to the oldest or newest offset.

How do we replicate without losing acknowledged messages?

Take one partition with three replicas in three zones. When may the leader say “got it”?

Bad: asynchronous primary–backup. The leader acknowledges once the batch is in its own log; followers copy later. Fast — in the 2014 benchmark one producer wrote 786,980 records/s with asynchronous 3× replication against 421,823 when it waited. But anything acknowledged and not yet copied dies with the leader. Each broker leads about 71 MB/s at peak, so with followers 200 ms behind a crash silently discards about 14 MB — around 14,000 acknowledged orders and payments. Fine for click logs; not for UPI.

Good: wait for every replica. Nothing acknowledged is lost, but every write waits for the slowest replica, and a dead replica blocks writes until someone removes it. Removing it automatically is the next idea.

Great: a leader, followers and an in-sync replica set (ISR).

1. Followers pull with ordinary fetches, so replication batches for free; each fetch tells the leader the follower’s log end offset.
2. The leader keeps the ISR: itself plus followers that are caught up. A follower not caught up for replica.lag.time.max.ms (30 s) is dropped; a broker silent to the controller for broker.session.timeout.ms (9 s) is fenced and dropped. It rejoins only once fully caught up.
3. The high watermark is the smallest log end offset in the ISR. Records below it are committed, and consumers see only those, so they never read something a failover could erase.
4. acks=all makes the leader reply once the high watermark passes the batch.
5. min.insync.replicas=2 makes the leader refuse acks=all writes (NotEnoughReplicas) when the ISR drops below two, rather than accept writes living on one disk.
6. Leaders come only from the ISR. Each leadership change bumps a leader epoch, stamped on every batch, so a returning replica can find and truncate exactly the tail that was never committed.

A committed record now survives the loss of any one broker or zone (two, if all three replicas were in sync), and writes continue with one broker down. You pay twice. acks=all latency follows the slowest in-sync replica until it is dropped: a dead follower can stall acknowledgements for up to 9 s, a slow one for up to 30 s. And cross-zone replication is 6 GB/s at peak, which clouds usually bill; at least let consumers fetch from a follower in their own zone.

When every in-sync replica is gone: unclean election. Broker 2 (leader) and broker 1 (the other ISR member) die; only broker 3, which left the ISR at offset 6, is alive. Wait for an in-sync replica, and nothing acknowledged is lost but the partition is down — forever if those disks are gone. Promote broker 3, and the partition is back at once but acknowledged offsets 6–9 vanish and get reused. Kafka has waited by default since 0.11 (unclean.leader.election.enable=false); enable it per topic only where uptime beats completeness, such as metrics. It is the CAP trade-off as one flag.

Why an ISR rather than a majority quorum? Raft, Paxos and Zab commit once a majority has a write and elect from a majority, so 2f + 1 replicas tolerate f failures. The ISR commits once every current in-sync replica has it and elects only from the ISR, so f + 1 replicas tolerate f. The design notes do the maths: surviving two failures takes five copies by majority — five times the disk, a fifth of the throughput — but three with an ISR. For our week, that is 5 × 605 TB ≈ 3.0 PB against 1.8 PB, and 12 GB/s of replication against 6 GB/s. The quorum’s advantage is latency: it waits only for the fastest majority. The ISR also needs a strongly consistent place to record who is in sync — the controller of deep dive 5, which does use majority voting because its data is small. More in data replication and quorums and leaderless replication.

How do consumers share work and keep their place?

Bad: a copy of every message per subscriber. Four groups means four copies of 605 TB plus delivery state for each, and a fifth subscriber adds another 605 TB — in traditional messaging, as the 2014 benchmark post notes, adding a consumer doubles your data.

Also bad: competing consumers on one shared queue. Any free worker takes the next message. One copy, and it scales, but order is gone: ORDER_PAID and ORDER_CANCELLED for one order go to two workers and the cancel can finish first. The broker is back to per-message locks, acknowledgements and redelivery timers.

Good: consumer groups that own partitions. Each partition has one owner per group; groups are independent; members process their partitions in order with no locks. Partitions cap parallelism: order-events has 120, so a 121st member idles. That is why the original paper over-partitions — within limits (deep dive 5). Members heartbeat to the group coordinator; one silent for the session timeout (45 s) or not calling poll() for 5 minutes is removed and its partitions reassigned.

Where positions live. The 2011 design kept offsets in ZooKeeper. Modern Kafka writes them as records to __consumer_offsets (50 partitions, three replicas), keyed by group, topic and partition and compacted. A group’s coordinator is the leader of the offsets partition its group ID hashes to. Commits are just log appends — cheap at 19,200 a second — and can join a transaction. Auto-commit every 5 s (the default) means up to 5 s of records reprocessed after a crash; commit after processing for at-least-once.

The cost is rebalancing. Every membership change reassigns partitions. In the classic eager protocol every member gives up all its partitions and waits for a new assignment — KIP-429 explains this is how it guaranteed no partition ever had two owners — so the whole group stops. A rolling deploy of 60 pods is a leave and a join each, 120 rebalances; at 3 seconds each, 6 minutes of zero progress per deploy.

Three fixes, best combined:

• Cooperative rebalancing (KIP-429, Kafka 2.4): members keep partitions that are not moving; one round revokes only those that must move, a second assigns them. It needs a sticky assignor.
• Static membership: a stable group.instance.id per pod, so a pod restarting within the session timeout gets its partitions back with no rebalance.
• A broker-driven protocol (KIP-848, generally available in Kafka 4.0): the coordinator computes assignments and reconciles members incrementally, with no group-wide barrier.

Lag is the health metric: log end offset − committed offset, alerted on as time. 1.2M records behind at 20,000 a second is 60 seconds. Past about 8 minutes a consumer leaves the page cache and reads disk; near 7 days it is about to lose data.

How do we get ordering and exactly-once delivery?

Ordering. Order exists only within a partition, so put everything that must stay in order under one key: order ID for order events, payment reference for UPI events, driver ID for ride locations. Two traps: adding partitions changes hash(key) mod N, splitting a key’s history across partitions, so size partition counts up front; and with several requests in flight, a failed-then-retried batch can land after its successor — unless the producer is idempotent. (See message ordering.)

Three levels of delivery. At most once: commit, then process; crashes lose records. At least once: producers retry and consumers commit after processing; nothing is lost, but a lost acknowledgement or a crash before commit means duplicates. Exactly once is two problems: no duplicates in the log, and an atomic “consume, process, produce, advance the offset”. See delivery guarantees.

Bad: every consumer dedupes everything. Every team stores and checks message IDs forever, and the log still holds duplicates for each new consumer to trip over.

Good: an idempotent producer. KIP-98 gives each producer an ID (PID) and has it number records per partition from 0. The leader remembers the last sequence per producer and partition; the next batch must start exactly one higher. Lower is a duplicate — acknowledge, don’t append. Higher is a gap and fails with a fatal OutOfOrderSequence. The broker tracks the last five batches per producer, which is why idempotence allows at most five in-flight requests and why order survives retries. It costs a few bytes per producer per partition and is on by default in current clients. Its limit: a restarted producer gets a new PID.

Great: transactions for read-process-write. A settlement service reads payment-events, writes ledger-entries, and must advance its input position exactly when its outputs become visible. With a stable transactional.id, a transaction coordinator on a broker tracks its transactions in an internal transaction log. Starting up bumps the producer’s epoch, fencing any zombie with the same ID and finishing whatever it left open. On commit, the coordinator logs “prepare commit”, writes a commit marker into every partition involved — including the consumed offsets — then logs “committed”. Consumers with isolation.level=read_committed read only up to the last stable offset, before which every transaction has finished.

producer.initTransactions();                 // fences older instances with this transactional.id
while (running) {
  var batch = consumer.poll(Duration.ofMillis(100));
  if (batch.isEmpty()) continue;
  producer.beginTransaction();
  for (var payment : batch) {
    producer.send(new ProducerRecord<>("ledger-entries", payment.key(), toLedgerEntry(payment.value())));
  }
  producer.sendOffsetsToTransaction(nextOffsets(batch), consumer.groupMetadata());
  producer.commitTransaction();              // outputs and input position become visible together
}                                            // on error: abortTransaction(), then rewind the consumer

You gain exactly-once within Kafka. You pay three ways: coordinator round trips and a marker per partition per transaction, so commit every ~100 ms, not per record; read_committed readers see output only after commit, adding that 100 ms; and a producer hung mid-transaction holds back the last stable offset for every read_committed reader for up to transaction.timeout.ms (60 s).

Effectively-once at the edges. Transactions stop at Kafka’s boundary. When a consumer writes to PostgreSQL or calls an SMS provider, make the consumer idempotent: store the offset in the same database transaction as the result, upsert by event ID (the UPI transaction reference is a natural key), or pass an idempotency key. The 2011 paper already preferred this to two-phase commit. See effectively-once consumers.

How do we manage cluster metadata and leadership?

Metadata is everything except messages: live brokers, topic configs, replica placement, each partition’s leader and epoch, ISRs, ACLs and quotas. It is only tens of megabytes for 24,000 partitions, but it sits on the critical path of every failover.

Bad: every partition elects its own leader. A dead broker that led 570 partitions triggers 570 separate elections and 570 metadata updates fanned out to every broker and client. The design notes call this the expensive way; one controller electing all new leaders in a single batch is far cheaper.

Good: one controller, metadata in ZooKeeper. For about a decade Kafka kept metadata in ZooKeeper, elected one broker as controller through it, and had the controller push updates to every broker. KIP-500 lists where it hurt:

• Failover reloaded everything. A new controller read the full state from ZooKeeper before electing a single leader — 14 s for 100,000 partitions even after Kafka 1.1 halved it from 28 s — so a hard controller crash cost a ZooKeeper session timeout plus the reload. Hence the old rule of thumb: at most 4,000 partitions per broker, 200,000 per cluster.
• Pushed updates could half-arrive, and the controller’s memory could drift from ZooKeeper’s.
• A broker cut off from the controller kept serving as long as it reached ZooKeeper, so an acks=1 producer could keep writing to a replaced leader.
• Two distributed systems to run, secure and upgrade.

Great: a Raft-replicated metadata log (KRaft). Three (or five) controllers run Raft over a metadata log. The Raft leader is the active controller; every change is a record, and standbys replay the same log, so they hold the full state. Brokers fetch metadata like consumers — from their last offset, deltas only — and keep a copy on disk for fast restarts. Membership rides the same stream: brokers heartbeat every 2 s, and one silent for 9 s is fenced, dropped from every ISR, and has its leaderships moved in one batch. A fenced broker cannot linger as a stale leader. Kafka 4.0 (March 2025) is the first major release without ZooKeeper.

The controller uses exactly the majority quorum deep dive 2 rejected for data, for the same reason in reverse: metadata is megabytes, so extra copies are cheap; messages are petabytes, so the ISR wins.

What a broker failure costs now. A hard crash: about 9 s to fence, one batch of leadership changes, clients refresh metadata — roughly 10 seconds of unwritable partitions for that broker’s share of leaders. A planned restart is gentler: the controller moves leaderships away first, and clients barely notice. (Batching this work is how Kafka 1.1 cut a controlled shutdown with 10,000 partitions from 6.5 minutes to 3 seconds.)

Partitions are not free, even with KRaft. Each costs open files, index memory, replica-fetcher work, metadata, election work and smaller producer batches. At about 1,700 replicas per broker this design sits inside even the old 4,000 guideline; KRaft raises the ceiling but does not make thousands of near-empty partitions sensible.

How do we retain lots of data cheaply?

Time and size retention. The broker deletes whole segments once the newest record in one is older than retention.ms (7 days), or the partition exceeds retention.bytes. Deleting a file is nearly free, but retention is only as fine as a segment, and the active segment is never deleted. An order-events partition at 2.5 MB/s fills 1 GiB in about 7 minutes. A quiet audit topic at 1 KB/s needs 2^30 ÷ 1,000 ≈ 1.07M seconds — 12.4 days — so it waits for the time-based roll (segment.ms, 7 days) and its data can live up to 14 days. If you promise deletion within 7 days, lower segment.ms on such topics.

Compaction for keyed state. Some topics are a changelog of current state — merchant settings, a driver’s vehicle, a customer’s KYC status — where you want the latest value per key forever. With cleanup.policy=compact, a background cleaner rewrites older segments, keeping only the newest record per key. A null value is a tombstone that deletes the key and is itself dropped after delete.retention.ms (1 day). Offsets and order never change, so a consumer reading from offset 0 rebuilds the current state. __consumer_offsets works this way, and so do logs used for event sourcing and change data capture.

Compaction costs throttled I/O and memory: the cleaner’s key map takes 24 bytes per entry, so 8 GB covers about 366 GB of 1 KB records per pass, say the design notes. Up to half the log may be stale before cleaning starts (min.cleanable.dirty.ratio 0.5), so a topic of 50M merchants at 1 KB holds 50–100 GB however often they change. Trap: a consumer more than a day behind can miss tombstones.

Tiered storage. Consumers mostly read the last few minutes from the page cache; older data serves only replays and backfills. Yet that cold data sets the broker count — 42 brokers bought for disks, NICs at 14%. KIP-405, production-ready since Kafka 3.9, adds a second tier: when a segment rolls, the leader copies it and its indexes to object storage. Local copies follow a short local retention, remote ones the topic’s full retention, and brokers serve old offsets from the remote tier.

Resize with 12 hours local: 1 GB/s × 43,200 s × 3 replicas ≈ 130 TB local, plus 604.8 TB remote in one copy. Now the network sets the count: 18 GB/s of peak egress at 40% of a 25 Gbps NIC (1.25 GB/s each) needs about 15 brokers, so run 18 — six per zone. Each holds 130 ÷ 18 ≈ 7.2 TB and writes 9 GB/s ÷ 18 = 500 MB/s at peak: NVMe, not hard disks. Replacing a dead broker drops from copying 43 TB (a day at a throttled 500 MB/s) to 7.2 TB (4 hours).

You pay elsewhere: a slower first byte and request charges for old reads, more metadata to keep consistent, no tiering for compacted topics, and a page cache that now holds only 100 GB ÷ 500 MB/s ≈ 200 s, so consumers more than about 3 minutes behind read local disk.

How do we choose partitions and keys, and stop one client hurting everyone?

Partition count. Take the larger of throughput ÷ what one partition takes and throughput ÷ what one consumer processes; the second usually wins. order-events peaks at 300,000 records/s and each consumer writes to a database at 2,500/s, so 300,000 ÷ 2,500 = 120 partitions (2.5 MB/s each, trivial for a broker). Plan for growth, since adding partitions remaps keys; 120 also divides evenly among 2, 3, 4, 5, 6, 8, 10, 12, 15 or 20 consumers.

Hot keys. Key by seller ID instead, and one seller’s flash deal bringing 30% of orders pours 90,000 records/s into one partition. The broker shrugs at 90 MB/s, but the partition’s single consumer manages 2,500/s and falls 87,500 behind every second — 5.25M records a minute. Key by what needs ordering — the order, not the seller. If per-seller order truly matters, split only known-hot keys (seller_77#0 to seller_77#7) and merge downstream. Key ride locations by driver, never by city, or Mumbai becomes one hot partition. See hot keys.

Batches shrink as partitions grow. A pod sending 600 records/s keyed over 120 partitions sends 5 per partition per second, so even a 50 ms linger rarely puts two in one batch: many tiny requests, poor compression. Use fewer, busier producers (an ingest gateway), keyless topics where order does not matter, or fewer partitions.

Quotas and backpressure. A backfill replaying a week at full speed can saturate broker NICs for everyone. Brokers enforce quotas per user and client ID: a byte rate per broker and a share of request-handler threads. Over quota, the broker computes the delay needed, returns it as throttle_time_ms and mutes the connection that long, measured over short windows (say 30 of 1 second). A slow cluster fills the producer’s 32 MB buffer and send() blocks, pushing back on the application; pull consumers just accumulate lag. See backpressure and load shedding and noisy neighbours.

The final design

The data path is producers, leaders, followers and consumers; the controller quorum and object storage sit beside it, never in front.

Follow one ORDER_PAID event through the system:

1. Produce. The order service batches per partition (up to 5 ms linger), compresses, tags each batch with producer ID and sequence number, and sends it to the leader found in cached metadata. Clients over quota are slowed, not rejected.
2. Replicate and acknowledge. Followers in the other two zones pull the batch; once every in-sync replica (at least two) has it, the high watermark passes it and the leader acknowledges, typically within 10 ms.
3. Consume. Group members long-poll their partitions from their own offsets; the leader, or a same-zone follower, sends the still-compressed batch from the page cache with sendfile.
4. Commit. Members commit offsets to their coordinator’s __consumer_offsets, or inside a transaction with their outputs; consumers writing to databases dedupe by offset or event ID.
5. Tier and retain. Rolled segments and indexes are copied to object storage; local copies go after 12 hours, remote ones after the topic’s retention; compacted topics keep the newest record per key.
6. Control. The KRaft quorum holds the metadata log that brokers fetch as deltas. A broker silent for 9 s is fenced and its leaderships move to in-sync replicas in one batch; clients refresh metadata and carry on.

Against the requirements: 18 GB/s out over 18 brokers is about 1 GB/s each, under 40% of a 25 Gbps NIC. Acknowledgements take about 10 ms, bounded by one cross-zone fetch, and caught-up consumers read from memory. Every acknowledged record is on at least two brokers in different zones, with unclean election off. One broker or zone can fail without stopping writes; a hard crash costs about 10 seconds on a slice of partitions, and controller failover is an election, not a reload. Seven days or more sit in object storage, replayable through the same API.

What interviewers look for by level

Follow-up questions

What if we need a classic work queue with per-message acknowledgements?

Jobs such as sending OTPs want any free worker to take the next message, per-message acknowledgement, redelivery after a timeout, and more workers than partitions. Kafka 4.0 added share groups for this (early access at release): members share partitions, each record is locked to one consumer for 30 seconds by default and acknowledged, released or rejected individually, with delivery attempts counted. Otherwise use retry and dead-letter topics — or, if routing and priorities dominate, a classic broker; see message queues.

What if a whole region goes down?

Run a cluster per region and mirror asynchronously between them. You lose whatever had not been mirrored, typically seconds, and consumers fail over with translated offsets, since offsets differ between clusters. Stretching one cluster synchronously across regions avoids the loss but adds a cross-region round trip, tens of milliseconds, to every acks=all write. Most teams mirror and keep consumers idempotent. See multi-region and active-active and disaster recovery.

What if one message can never be processed?

A poison record — malformed JSON for one UPI transaction — blocks its partition, because the consumer cannot commit past it. After a few attempts, write it and its error to a dead-letter topic, commit past it and alert. Send transient failures, such as a bank API timing out, to retry topics with growing delays so they do not block healthy records. See delivery guarantees and DLQs.

What if messages are 20 MB video files?

Keep them out of the log. The default maximum batch is about 1 MB (max.message.bytes), and huge records wreck batching, the page cache and replication latency for everyone. Use a claim check: upload to blob storage and publish a small event with the URL, size and checksum.

What if producers must not lose events when the cluster is unreachable?

The producer’s 32 MB buffer lasts 16 seconds at 2 MB/s, then send() blocks and fails. For events born in a database transaction — an order row and its ORDER_PAID event — use a transactional outbox: write the event to an outbox table in the same transaction and let a relay publish it later. It may be published twice but is never lost, so consumers stay idempotent.

How to say it in the interview

“I’ll design a Kafka-style log: about 3 million 1 KB records a second at peak, each read by around four consumer groups, kept for 7 days, and no acknowledged record ever lost.

A topic is split into partitions, and each partition is an append-only log of segment files with a sparse offset index. Producers batch and compress per partition, hash the key so each order’s events stay in order, and send straight to the partition leader. Writes and reads are sequential, the page cache does the caching, and consumers are served with zero-copy sendfile.

Each partition has three replicas in three zones. Followers pull from the leader; the ISR is the set that is caught up, and the high watermark is the lowest log end offset among them. Producers use acks=all with min.insync.replicas of 2, so any one broker or zone can fail without losing an acknowledged record, and unclean leader election stays off. I chose the ISR over a majority quorum because it needs f + 1 copies instead of 2f + 1 for the same fault tolerance.

Consumers pull from an offset in groups; each partition has one owner per group, and positions are committed to an internal compacted topic, so replay is just moving an offset. I’d use cooperative rebalancing and static membership so deploys don’t stop the world. Producers are idempotent, transactions cover read-process-write inside Kafka, and anything writing to a database dedupes by offset or event ID.

Metadata lives in a Raft quorum of three controllers, so failover is an election, not a reload. Sizing: 3 GB/s in becomes 18 GB/s out and 1.8 PB on disk across about 42 brokers; with tiered storage keeping 12 hours locally and the rest in object storage, about 18 brokers on NVMe are enough, and the network becomes the constraint. Finally, quotas and good partition keys stop one client or one hot seller from hurting everyone.”

Concepts used in this design

• Kafka — the system this design reconstructs from first principles.
• Message queues — the classic per-message-acknowledgement model the log replaces.
• Publish/subscribe — many independent consumer groups reading one stream.
• Write-ahead logging — the append-only, sequential structure each partition is.
• Data replication — leaders, followers and the in-sync replica set.
• Quorums and leaderless replication — the majority-vote alternative to the ISR.
• Consensus and leader election — Raft in the controller quorum.
• ZooKeeper and etcd — the external coordinator Kafka used to depend on.
• Delivery guarantees — at-most, at-least and exactly once, plus dead-letter topics.
• Message ordering — keys, partitions and per-key order.
• Effectively-once consumers — idempotent processing at the system’s edges.
• Transactional outbox — publishing reliably from a database transaction.
• Event sourcing and CQRS — compacted topics as a source of current state.
• Backpressure and load shedding — quotas, blocking producers and pull consumers.
• Blob storage — the remote tier and claim checks for large payloads.
• CAP theorem and PACELC — unclean leader election as an availability-versus-consistency switch.

References and further reading

1. Kafka: a Distributed Messaging System for Log Processing (Kreps, Narkhede and Rao, NetDB 2011)notes.stephenholiday.comThe original design: segment files addressed by offset, page cache and sendfile, consumer-held positions, consumer groups over ZooKeeper, and the batching benchmarks against ActiveMQ and RabbitMQ.
2. Apache Kafka documentation: Designkafka.apache.orgWhy the filesystem and page cache, sendfile’s four copies versus one, batch compression, pull consumers, delivery semantics, ISR versus majority quorums, unclean election, log compaction and quotas.
3. Apache Kafka documentation: Configuration referencekafka.apache.orgDefaults used here: 1 GiB segments, a 4 KB index interval, 7-day retention, min.insync.replicas, replica lag and broker session timeouts, producer batching and idempotence, consumer commit and poll settings.
4. KIP-500: Replace ZooKeeper with a Self-Managed Metadata Quorumcwiki.apache.orgProblems with pushed metadata and controller reloads, the Raft controller quorum, brokers fetching metadata deltas, and broker fencing.
5. KIP-98: Exactly Once Delivery and Transactional Messagingcwiki.apache.orgProducer IDs and sequence numbers, transactional IDs and epochs that fence zombies, the transaction coordinator, commit markers, read_committed and the last stable offset.
6. KIP-405: Kafka Tiered Storagecwiki.apache.orgLocal and remote tiers with separate retention, copying rolled segments and indexes to object storage, and why recovery time shrinks with less local data.
7. KIP-429: Kafka Consumer Incremental Rebalance Protocolcwiki.apache.orgWhy the eager protocol revokes every partition, and how cooperative rebalancing moves only the partitions that change owner.
8. Benchmarking Apache Kafka: 2 Million Writes Per Second (On Three Cheap Machines)engineering.linkedin.comProducer and consumer throughput with async and sync replication on commodity disks, flat throughput as data grows, and 2 ms median end-to-end latency.
9. Apache Kafka Supports 200K Partitions Per Clusterconfluent.ioController failover and controlled-shutdown times in the ZooKeeper era, and the 4,000-per-broker and 200,000-per-cluster rule of thumb.
10. Apache Kafka 4.0.0 release announcementkafka.apache.orgThe first major release without ZooKeeper, the new consumer rebalance protocol reaching general availability, and early access to share groups.