Design a Distributed Message Queue (like Kafka) - System Design

Everyone thinks a message queue is a list. “Producers push messages onto the back, consumers pop them off the front, done - it is a queue, the name is right there.” That mental model survives exactly until the interviewer asks the questions that make it a real system: what happens when the consumer crashes halfway through a batch, does the message come back or is it gone? What happens when you have a thousand consumers and one queue, how do they share the load without stepping on each other? What happens when the broker holding your messages dies, are the messages gone with it? And when the network retries a publish, does the message land once or twice, and who pays for the duplicate?

A distributed message queue like Kafka is deceptive because the “list” demo works fine on a laptop with one producer and one consumer, and falls apart the moment you need durability (messages must survive a broker crash), throughput (millions of messages/sec, more than one machine can hold), ordered replay (a new consumer wants to read history from the beginning), and multiple independent consumers (a payments service and an analytics service both want every order event, at their own pace). The whole design comes down to a few decisions that are not obvious: storing messages as an append-only commit log instead of a mutable queue, partitioning that log across machines so it scales past one box, tracking each consumer’s position as an offset it owns rather than deleting messages on read, replicating each partition with a leader-follower protocol so a broker death loses nothing, and being honest that exactly-once is not free - it is at-least-once plus idempotence, and you have to build the second half.

Let me do it properly.

Functional Requirements (FR)

In scope:

• Publish messages to a topic. A producer sends a message (a key, a value, a timestamp) to a named topic. The system durably stores it and acknowledges. This is the write path.
• Subscribe and consume from a topic. A consumer reads messages from a topic, in order within a partition, starting from wherever it left off (or from the beginning, or from the tail). This is the read path.
• Durable retention and replay. Messages are retained for a configured window (time or size based) regardless of whether they have been consumed. A consumer can rewind and re-read history. The queue is a log, not a mailbox that empties on read.
• Consumer groups (load sharing). Multiple consumer instances form a group and split the partitions of a topic between them so the work is shared and parallel. Add an instance, it takes over some partitions; lose one, its partitions move to the survivors.
• Multiple independent subscribers (fan-out). Different consumer groups each get their own full, independent copy of the stream at their own pace - payments reads every order event, analytics reads every order event, neither affects the other.
• Ordering guarantee within a partition. Messages with the same key land on the same partition and are delivered in the order they were produced. Global total order across the whole topic is explicitly not promised (that would kill parallelism).
• Configurable delivery semantics. Support at-least-once (the sane default), at-most-once, and exactly-once (with the extra machinery it requires).

Explicitly out of scope (say it so you own the scope):

• Message transformation / stream processing. Joining, windowing, aggregating streams is a separate layer (Kafka Streams / Flink) that sits on top of the queue. We build the log and the delivery; we do not process the payloads.
• Complex routing / per-message priority. Topic-exchange routing, priority queues, per-message TTL, dead-letter policies as first-class features are the RabbitMQ/SQS style. We build the partitioned-log style (Kafka), which trades rich routing for throughput and replay. We note where a dead-letter topic fits.
• The schema registry. Enforcing that message payloads match a schema is a useful adjacent service; we treat the value as opaque bytes and mention where the registry hooks in.
• Cross-datacenter geo-replication. Mirroring topics between regions (MirrorMaker style) is a real feature but a separate design; we build a single-cluster design and note the extension.
• Exactly-once across arbitrary external sinks. We guarantee exactly-once within the queue’s processing (idempotent producer + transactions). Making a downstream database write exactly-once too requires that sink to cooperate (idempotent writes or its own transaction), which we describe but do not own.

The one decision that drives everything: a durable message queue at scale is not a queue that empties on read - it is a partitioned, replicated, append-only commit log where consumers track their own position. That single reframe - log, not queue; consumer-owned offset, not broker-managed deletion - is what gives you durability, replay, fan-out, and horizontal scale all at once, and everything below is a consequence of it.

Non-Functional Requirements (NFR)

• Scale: a large cluster ingesting on the order of millions of messages/sec and several GB/sec of write bandwidth, across thousands of topics and tens of thousands of partitions, served by a fleet of tens to low hundreds of broker nodes. A single busy topic can be hundreds of thousands of messages/sec on its own. Reads typically exceed writes because every message is usually consumed by more than one consumer group (fan-out multiplies read traffic).
• Latency: produce-ack p99 in the low single-digit to tens of milliseconds for the common durability setting; end-to-end (produce to consume) on the order of milliseconds to tens of milliseconds when consumers are caught up. This is a low-latency pipe, not a batch system, though it batches internally for throughput.
• Availability: 99.9%+ on both produce and consume. A broker failure must not lose accepted messages and must not stall the topic for more than the few seconds it takes to fail over a partition leader. Producers and consumers transparently reconnect to the new leader.
• Consistency: within a partition, a strict total order and a single source of truth (the partition leader). Across partitions, no ordering promise. Reads see only committed messages (replicated to the in-sync set), so a consumer never reads a message that a subsequent leader failover could erase.
• Durability: very high. Once a produce is acknowledged under the durable setting (acks=all), the message survives the loss of any single broker (with replication factor 3 and min.insync.replicas=2, it tolerates one broker loss with zero data loss). Messages are persisted to disk and replicated before the ack; they are not held only in memory.
• Delivery semantics: the default is at-least-once - no accepted message is lost, but a message may be redelivered after a failure. At-most-once and exactly-once are configurable, with exactly-once costing extra coordination (idempotent producer + transactional commits).

Back-of-the-Envelope Estimation (BoE)

Let me ground the design in numbers.

Traffic:

Target ingest:        2,000,000 messages/sec (write) at peak
Average message size: ~1 KB (key + value + headers, packed)
Write bandwidth:      2M/sec * 1 KB = 2 GB/sec inbound

Fan-out on read: each message consumed by ~3 consumer groups on average
Read bandwidth:  2 GB/sec * 3 = 6 GB/sec outbound
Total broker bandwidth: ~8 GB/sec across the fleet (write + read).

The read side is bigger than the write side, and by a factor that depends entirely on fan-out. This is the first non-obvious number: in a queue, reads usually dominate writes because the same message is read once per consumer group, and a popular event stream (say, “user did X”) might feed five or ten independent consumers. Size the network and the page cache for reads, not just writes.

Per-broker capacity and fleet size:

A single broker (NIC + disk) sustainably handles ~1 GB/sec combined
   (modern 25 GbE NIC ~3 GB/sec line rate, disk sequential ~1-2 GB/sec,
    keep headroom -> budget ~1 GB/sec sustained per broker).
Total ~8 GB/sec / ~1 GB/sec per broker -> ~8 brokers just for bandwidth,
   but you need many more for storage, partition count, and failure headroom.
Realistic fleet: tens of brokers (say 30-60) for this load with room to spare.

Storage:

Write volume:        2 GB/sec = 2 * 86,400 sec ≈ 172 TB/day of raw log.
Retention:           keep 7 days of log (replay window).
Primary log storage: 172 TB/day * 7 = ~1.2 PB (one copy).
Replication factor 3: ~1.2 PB * 3 = ~3.6 PB of raw disk across the cluster.

Per broker (60 brokers): 3.6 PB / 60 ≈ 60 TB/broker.
   -> dense storage nodes; this is a disk-bound, sequential-IO workload.

The storage number is large but the access pattern is friendly: it is almost entirely sequential append on write and sequential scan on read, which spinning disks and NVMe both do well and which the OS page cache serves from RAM for consumers that are caught up. The queue is disk-heavy but disk-efficient - the opposite of a random-access database.

Partition count:

Parallelism is bounded by partition count (one partition -> one consumer in a group).
To let a consumer group run, say, 200 parallel consumers on the hottest topic,
   that topic needs >= 200 partitions.
Across thousands of topics: tens of thousands of partitions cluster-wide.
Each partition = an open file/segment + replication streams + metadata,
   so partition count itself is a scaling dimension (see bottlenecks).

Offset / metadata volume:

Offsets are tiny: (group, topic, partition) -> a single 64-bit offset.
Even 10,000 groups * 10,000 partitions = 100M offset records * ~50 bytes
   ≈ 5 GB of offset state, committed periodically (not per message).
Trivial compared to the log; store it in a compacted internal topic.

The headline shape: write bandwidth is a couple GB/sec, read bandwidth is a multiple of that from fan-out, storage is petabytes of sequential log, and the scarce resources are disk throughput, network bandwidth, and partition count - not CPU and not random IOPS. Size for sequential IO and fan-out, and treat partition count as a first-class capacity number.

High-Level Design (HLD)

The system is a cluster of broker nodes that store topic partitions as append-only logs on local disk. Producers write to the leader of a partition; the leader replicates to follower brokers; consumers read from the leader (or a follower) at an offset they track themselves. A metadata/coordination layer tracks which broker leads each partition and coordinates consumer-group membership. There is no central data path - the brokers are the data plane, and clients talk directly to the broker that owns the partition they care about.

   PRODUCERS                                              CONSUMERS (in groups)
  ┌─────────┐                                            ┌──────────────────────┐
  │ producer│  1. ask: who leads topic T partition p?    │ group G1 (payments)  │
  │  (order │◄──────────── metadata ───────────────┐     │  c1  c2  c3          │
  │  events)│                                       │     ├──────────────────────┤
  └────┬────┘                                       │     │ group G2 (analytics) │
       │ 2. produce(T, p, key, value)               │     │  c1  c2              │
       │    to the PARTITION LEADER                 │     └──────────┬───────────┘
       ▼                                            │                │ fetch(T,p, offset)
 ┌────────────────────────── BROKER CLUSTER ────────┼────────────────┼──────────┐
 │                                                  │                │          │
 │   ┌── Broker A (leader T-p0) ──┐   ┌── Broker B ─┴──┐   ┌── Broker C ────────┐│
 │   │  partition T-p0 (LEADER)   │   │ T-p0 follower  │   │ T-p0 follower      ││
 │   │  append-only commit log:   │──►│  replicate     │──►│  replicate         ││
 │   │  [seg0][seg1][seg2..active]│   │ T-p1 (LEADER)  │   │ T-p1 follower      ││
 │   │  T-p1 follower             │   │ T-p2 follower  │   │ T-p2 (LEADER)      ││
 │   └────────────┬───────────────┘   └────────────────┘   └────────────────────┘│
 │                │ 3. leader appends, waits for in-sync replicas (ISR) to ack     │
 │                │    -> "committed" -> ack producer (acks=all)                   │
 │                                                                                 │
 │   ┌────────────────── COORDINATION / METADATA (Raft-based controller) ────────┐│
 │   │  - which broker leads each partition, ISR membership                       ││
 │   │  - broker liveness, leader election on failure                             ││
 │   │  - consumer group membership + partition assignment                        ││
 │   │  - __offsets topic (compacted): (group,T,p) -> committed offset            ││
 │   └────────────────────────────────────────────────────────────────────────────┘│
 └─────────────────────────────────────────────────────────────────────────────────┘

Request flow - producing a message (the write path):

1. The producer wants to publish an order event. It first asks the cluster metadata for who leads topic T, partition p (it caches this). Which partition? If the message has a key, p = hash(key) % numPartitions, so all events for the same order land on the same partition and stay ordered. No key -> round-robin for balance.
2. The producer sends the message to the leader broker for T-p. Under the hood it batches many messages per request for throughput (this batching is why the queue hits millions/sec).
3. The leader appends the message to the tail of that partition’s on-disk commit log and assigns it the next sequential offset. The append is sequential - no in-place update, no index churn on the hot path.
4. The leader replicates the new records to the follower brokers in the partition’s replica set. Once the in-sync replicas (ISR) have the record, it is committed. Under acks=all the leader acks the producer only now - so an acknowledged message already survives the loss of the leader.
5. The producer gets its ack (with the assigned offset). Done. The message sits in the log until retention expires, readable by any consumer any number of times.

Request flow - consuming a message (the read path):

1. A consumer belongs to a consumer group. On join, the group coordinator assigns partitions to the members (partition p0 to consumer c1, p1 to c2, and so on) so the group splits the topic’s partitions and each partition is owned by exactly one consumer in the group.
2. The consumer fetches from the leader of its assigned partition starting at its current offset - “give me messages from offset 4,502 onward.” The broker returns a batch by reading sequentially from the log (served from page cache if hot).
3. The consumer processes the batch, then commits its offset - “I have durably handled through offset 4,600” - by writing to the internal __offsets topic. The offset is the consumer’s bookmark; the broker does not delete the message, it just remembers where this group is.
4. A different consumer group (analytics) fetches the same partitions independently at its own offset - it gets its own full copy of the stream, unaffected by the payments group’s pace. That is fan-out: the log is read many times, deleted by no reader.
5. If a consumer dies, the coordinator rebalances - its partitions are reassigned to surviving members, who resume from the last committed offset. At-least-once falls out naturally: any messages processed-but-not-yet-committed before the crash get redelivered.

The key architectural insight: the broker stores an immutable log and remembers nothing about consumers except, optionally, their committed offset - which the consumers own. A traditional queue couples storage to consumption (message deleted when consumed, broker tracks per-message ack state), which makes fan-out, replay, and high throughput all hard. Kafka’s move is to decouple them: the log is a durable, append-only sequence with retention independent of consumption, and “where a consumer is” is just an integer that consumer advances. That one decision is what simultaneously buys durability (messages persist past consumption), replay (rewind the offset), fan-out (many groups, many offsets, one log), and throughput (append and sequential-scan, no per-message ack bookkeeping on the broker).

Component Deep Dive

Four hard parts: (1) how the messages are actually stored - the commit log; (2) how the log scales past one machine - partitioning; (3) how consumers share work and track position - consumer groups and offsets; and (4) how a partition survives a broker death and what delivery guarantee that gives you - replication and semantics. I will walk each from naive to scalable.

1. Storage: Mutable Queue -> Append-Only Commit Log

We need to store messages durably, serve them to many readers, and sustain GB/sec of writes.

Approach A: A real queue - a mutable list, delete on consume (the naive instinct)

Keep an in-memory (or DB-backed) list per topic. Producers append to the tail. A consumer pops from the head; the broker tracks per-message “delivered / acked” state and deletes a message once acknowledged.

enqueue(msg)  -> list.append(msg)
dequeue()     -> msg = list.popHead(); mark in-flight; on ack -> delete msg

Where it breaks: three ways, all fatal at scale.

• No fan-out. If the payments service pops a message, it is gone - analytics never sees it. To support N independent subscribers you would keep N copies or N cursors plus N-way per-message ack state. The broker now tracks the delivery status of every message for every consumer, which is a mountain of mutable, random-access bookkeeping.
• No replay. Once acked and deleted, history is gone. A new consumer that wants to read from the beginning, or a bug fix that needs to reprocess yesterday, has nothing to read.
• Bad IO pattern. Random deletes and per-message state updates are random-access writes - index churn, fragmentation, locking on the hot path. You cannot sustain millions/sec of that. And keeping it in memory means a crash loses everything unacknowledged.

The mutable-queue model conflates storage with consumption tracking, and both suffer for it.

Approach B: An append-only commit log with consumer-owned offsets

Store each partition as an immutable, append-only log: a sequence of records, each assigned a monotonically increasing offset, written to disk in segment files. Nothing is deleted on consume; the log is retained for a time/size window. A consumer’s position is just an offset it advances and periodically commits. The broker stores the log and (optionally) the committed offsets - nothing else.

partition T-p0 on disk:
  segment files, each a chunk of the log, rolled by size/time:
    00000000000000000000.log   offsets 0        .. 4,999,999
    00000000000005000000.log   offsets 5,000,000.. 9,999,999
    00000000000010000000.log   offsets 10,000,000.. (active, being appended)
  each segment has a sparse .index (offset -> byte position) for fast seek.

append(record):  write to tail of active segment; offset = last + 1  (sequential)
read(offset):    binary-search the sparse index -> byte pos -> sequential read

Why this is the right model:

• Sequential IO is fast. Appends go to the end of the active segment - pure sequential writes, which disks (spinning or NVMe) love and which need no index rewrite per message. Reads are sequential scans from an offset. This is how you get GB/sec on commodity disks.
• Zero-copy reads. The broker serves a fetch by sendfile()-ing bytes straight from the page cache to the socket, never copying through user space or re-serializing. A caught-up consumer is served entirely from RAM (page cache), disk untouched. This is a large part of why one broker can push ~1 GB/sec.
• Fan-out is free. Many consumer groups read the same immutable log at different offsets. The log does not care how many readers there are or how far behind they are; there is no per-consumer copy and no per-message ack state on the broker.
• Replay is free. Rewinding is seek(offset). Reprocess from the start, from a timestamp, or from the tail - it is just choosing where to start reading.
• Retention decoupled from consumption. Segments are deleted when they age out of the retention window (say 7 days) or the topic exceeds its size cap - not when consumed. An old segment is dropped by unlinking the whole file, which is cheap. Optionally, log compaction keeps only the latest record per key (for changelog-style topics like the offsets topic itself).

The commit log is the foundation. Every other nice property - durability, replay, fan-out, throughput - is a consequence of “immutable append-only log with consumer-owned offsets” instead of “mutable queue with broker-managed deletion.”

2. Scaling the Log: One Log -> Partitions

A single log lives on a single broker’s disk. That caps you at one machine’s storage and one machine’s write throughput, and it serializes all reads and writes for the topic through one node.

Approach A: One log per topic (naive)

Each topic is one append-only log on one broker.

Where it breaks: the topic’s total throughput and storage are bounded by one machine. A firehose topic at 500K messages/sec and tens of TB does not fit on one broker’s NIC or disk. Worse, consumption is serial - because there is exactly one log with one ordering, only one consumer can meaningfully read it in order at a time; you cannot parallelize consumption without breaking order. One log means one lane of traffic. You need many lanes.

Approach B: Partition the topic into N independent logs

Split each topic into N partitions, each an independent commit log, spread across brokers. A message is routed to a partition by its key:

partition = hash(key) % numPartitions       # keyed: same key -> same partition
partition = round_robin()                    # no key: spread evenly

topic "orders", 6 partitions, 3 brokers:
   Broker A: orders-p0 (leader), orders-p3 (leader)
   Broker B: orders-p1 (leader), orders-p4 (leader)
   Broker C: orders-p2 (leader), orders-p5 (leader)
   (plus followers of each elsewhere - see replication)

What partitioning buys and what it costs:

• Horizontal scale. N partitions across M brokers give you N times the throughput and storage of one log. Add brokers, spread partitions, and the topic scales past any single machine. Ingest and storage are now cluster-wide, not node-bound.
• Parallel consumption. A consumer group can run up to N consumers, one per partition, reading in parallel. Partition count is the unit and ceiling of parallelism - more partitions, more possible parallel consumers (and this is why partition count is a capacity decision, per BoE).
• Ordering, honestly scoped. Order is guaranteed within a partition, not across the topic. Because same-key messages hash to the same partition, all events for one order (or one user, or one account) are ordered relative to each other - which is the ordering you almost always actually need. Global total order across the whole topic is not promised, because promising it would force everything through one partition and destroy the parallelism. State this trade explicitly: you choose the partition key to be the entity whose per-entity order matters.
• The cost - choosing the key. The partition key is the most consequential schema decision. Key by order_id and all events for an order are ordered but one giant order cannot be split. Key by user_id and a celebrity user becomes a hot partition (the hot-key problem, addressed in bottlenecks). Change the partition count later and hash(key) % N remaps keys to different partitions, breaking the “same key, same partition” invariant for existing data - so partition count is hard to change after the fact. Pick it deliberately.

Partitioning turns “one log on one machine” into “N logs across the cluster,” which is what makes the queue horizontally scalable and consumable in parallel. The price is that ordering is per-partition and the partition key is a decision you largely cannot take back.

3. Consumer Groups and Offsets: Sharing Work Without Losing Your Place

Multiple consumer instances want to share the load of a topic, each must know where it is, and a crash must not lose progress or double-assign a partition.

Approach A: Every consumer reads everything; track position in the consumer’s memory (naive)

Each consumer instance reads all partitions and just remembers in memory how far it has read.

Where it breaks: two problems. First, no load sharing - if every instance reads every partition, you have not parallelized anything; ten consumers each do the full work of one. Second, progress is not durable - if the position lives only in the consumer’s memory, a crash or restart loses it, and the consumer either reprocesses from the beginning (huge duplication) or from the tail (skips everything it had not yet committed - silent data loss). And if you try to have several instances share partitions by ad-hoc agreement, two of them can grab the same partition and process it twice, or a partition can be orphaned when its owner dies with nobody noticing. You need a coordinator and durable offsets.

Approach B: Consumer groups with coordinated assignment and committed offsets

Introduce two mechanisms: a group coordinator that assigns partitions to members, and durable committed offsets stored in the cluster.

Consumer groups (the assignment side):

• Consumers declare a group id. The cluster’s group coordinator (a broker role) tracks group membership via heartbeats.
• The coordinator runs a rebalance that assigns each partition of the subscribed topics to exactly one consumer in the group. So a 6-partition topic with 3 consumers gives each consumer 2 partitions; each partition has one owner; the group as a whole covers all partitions once. This is the load sharing.
• Adding a consumer triggers a rebalance that hands it some partitions from others (scale up). Losing a consumer (missed heartbeats) triggers a rebalance that hands its partitions to survivors (fail over). If there are more consumers than partitions, the extras sit idle - partition count caps useful parallelism, which is why you size partitions generously.
• Rebalances are the tricky part operationally: a naive “stop-the-world” rebalance pauses the whole group while reassigning. Modern designs use incremental / cooperative rebalancing (only the partitions that actually move are paused, and sticky assignment keeps most partitions with their current owner) to avoid a full stall every time one consumer blips. Worth naming in an interview.

Fan-out via multiple groups: two different groups (payments and analytics) each get their own independent assignment and their own offsets over the same partitions. Groups are isolated: the log is read once per group, and one group’s speed or failure never touches another’s. This is how “many independent subscribers” and “shared load within a subscriber” coexist - within a group, partitions are split; across groups, everything is duplicated.

Offsets (the position side):

committed offset per (group, topic, partition):
   stored in an internal COMPACTED topic __consumer_offsets
   commit = produce a record: key=(group,T,p), value=offset, periodically
   on (re)assignment: consumer reads the last committed offset and resumes there

• The offset is the consumer’s durable bookmark - “this group has handled T-p0 through offset 4,600.” It lives in the cluster (the compacted __offsets topic), not in the consumer’s memory, so it survives crashes and moves with the partition on rebalance.
• Commit timing is the delivery-semantics knob, and this is the crux:
  • Commit after processing (at-least-once): process the batch, then commit. If you crash after processing but before committing, the next owner re-reads from the last commit and reprocesses - messages are redelivered, never lost. This is the sane default.
  • Commit before processing (at-most-once): commit first, then process. If you crash mid-processing, the offset already moved past those messages, so they are skipped - lost, never redelivered. Only acceptable when losing a message is cheaper than double-handling it (some metrics).
  • The two are mirror images, and which one you get is purely when you commit relative to when you process. There is no third “just right” option here - exactly-once needs more than offset timing (next section).
• Commit granularity is a throughput/duplication trade. Commit every message and you minimize reprocessing on crash but add overhead; commit every few seconds (the common default) and you get cheap commits but may reprocess a few seconds of messages after a crash. At-least-once means “design consumers to tolerate reprocessing,” which leads straight to idempotent consumers.

Consumer groups plus committed offsets give you shared, parallel, fault-tolerant consumption where progress is durable and each partition has exactly one owner per group - and the timing of the offset commit is exactly the dial that selects at-least-once versus at-most-once.

4. Replication and Delivery Semantics: Surviving a Broker Death, Counting Once

A partition on one broker’s disk dies with that broker. And even with the message safe, the network’s at-least-once nature means a message can be produced twice. We need durability across broker failure and a clear story on exactly-once.

Approach A: One copy per partition, trust the disk (naive)

Each partition lives on exactly one broker. Durability = “we wrote it to disk.”

Where it breaks: a broker dies (disk failure, crash, network partition) and every partition it led is unavailable and possibly lost until (or unless) that machine comes back. For a system whose whole point is not losing accepted messages, a single-copy design is a non-starter. Disk durability protects against a process crash, not against the machine going away.

Approach B: Leader-follower replication with an in-sync replica set

Give each partition R replicas (typically 3) on different brokers. One replica is the leader; the rest are followers. All produces and (by default) consumes go through the leader; followers continuously fetch from the leader to stay current.

partition orders-p0, replication factor 3:
   Broker A: LEADER   [.......... offset 0..4600 ..........]
   Broker B: follower [.......... fetching, caught up  ....]   } in-sync
   Broker C: follower [.......... fetching, caught up  ....]   } replicas (ISR)

produce (acks=all):
   leader appends -> waits until all ISR followers have the record
   -> record is COMMITTED -> leader acks producer

• In-Sync Replicas (ISR). The ISR is the set of replicas fully caught up with the leader. A follower that falls too far behind (slow, or partitioned away) is dropped from the ISR; when it catches up it rejoins. Only messages replicated to the whole ISR are committed, and consumers can only read committed messages - so a consumer never sees a message that a later failover could erase.
• The acks dial (durability vs latency), the producer’s half of the story:

With acks=all plus min.insync.replicas=2, an acknowledged message is on at least two brokers, so any single broker loss is survivable with zero data loss. This is the durable default.

• Leader failover. When a broker dies, the controller detects it (heartbeat loss) and elects a new leader for each affected partition from the ISR. Because every ISR member has all committed messages, the new leader has them too - no committed message is lost. Producers and consumers transparently reconnect to the new leader (they refetch metadata). Failover is seconds, not minutes.
• Unclean leader election (a sharp trade). If the entire ISR is lost and only a stale, out-of-sync follower survives, you choose: stay unavailable until an ISR member returns (consistency, no data loss) or elect the stale replica and lose the un-replicated tail (availability, some data loss). This is CAP in one config flag (unclean.leader.election); default off - prefer consistency for a durable queue.

Now delivery semantics, end to end. The queue’s transport is fundamentally at-least-once: a producer whose ack is lost (network drop after the broker persisted) will retry and can create a duplicate; a consumer that crashes post-process pre-commit will reprocess. Getting to at-most-once or exactly-once is about controlling those two duplication points.

• At-most-once: acks=0/1 and/or commit-before-process. Simplest, may lose messages, never double-processes. Rarely what you want.
• At-least-once (default): acks=all and commit-after-process. Never loses an accepted message, may duplicate. The workhorse.
• Exactly-once - built, not free. Two duplication points, two fixes:
  • Producer duplicates -> idempotent producer. The producer is assigned a producer id (PID) and stamps each message with a monotonic sequence number per partition. The broker remembers the last sequence it accepted per (PID, partition) and de-duplicates retries - a retried produce with a sequence it has already seen is dropped, not appended. This makes producer retries safe: exactly one copy lands in the log even though the producer sent it twice.
  • Consume-process-produce duplicates -> transactions. For the common “read from topic A, process, write to topic B and commit offset” loop, wrap the output writes and the offset commit in a single atomic transaction. Either both the downstream writes and the offset advance commit, or neither does. Consumers reading topic B with read_committed isolation only see messages from committed transactions, so a failed-and-retried processing attempt leaves no visible duplicate. Combined with the idempotent producer, this gives exactly-once within the Kafka pipeline.
  • The honest caveat. Exactly-once holds inside the queue’s own read-process-write. If your consumer’s side effect is an external system (charge a card, write to another database), exactly-once requires that sink to cooperate - be idempotent (a dedup key on the write) or participate in a transaction. The queue cannot make a non-idempotent external call happen exactly once by itself. The clean, widely-used pattern is therefore at-least-once delivery + idempotent consumer (dedup by message key or an idempotency id on the write), which is simpler than full transactions and covers most real needs. Say this plainly: “exactly-once” in practice usually means “at-least-once transport plus an idempotent consumer,” and the queue’s transactional exactly-once is the special case where the sink is another Kafka topic.

Replication makes an accepted message survive any single broker death; the acks dial and the offset-commit timing together select the delivery semantics; and exactly-once is at-least-once plus idempotence (producer sequence numbers, transactional commits, or an idempotent consumer) - never a free property of the transport.

API Design & Data Schema

Producer / Admin API (the write and control plane)

POST /topics                      # admin: create a topic
  { "name": "orders",
    "partitions": 12,
    "replication_factor": 3,
    "retention_ms": 604800000,     # 7 days
    "min_insync_replicas": 2,
    "cleanup_policy": "delete" }   # or "compact" for changelog topics
  201 Created

PRODUCE  (to the partition leader; batched under the hood)
  send(topic="orders", key="order_8f2a", value=<bytes>, headers={...})
  routing: partition = key ? hash(key) % partitions : round_robin
  acks: 0 | 1 | all           # durability dial
  -> { partition: 3, offset: 480127, timestamp: 1752019200123 }
     (returned only after the message is committed to the ISR under acks=all)

  # Exactly-once producer:
  init_producer(transactional_id="orders-processor-1")  # gets a stable PID
  begin_txn()
    send(topic="orders_enriched", ...)   # each stamped with (PID, seq)
    send_offsets_to_txn(group="enricher", offsets={orders-3: 480128})
  commit_txn()                            # output writes + offset commit, atomic

Consumer API (the read plane)

SUBSCRIBE
  subscribe(group_id="payments", topics=["orders"])
  -> coordinator assigns this member a subset of partitions (rebalance)

FETCH  (from the leader of an assigned partition, from the current offset)
  poll(max_bytes=1MB, max_wait_ms=100)
  -> returns a batch: [ { partition, offset, key, value, timestamp, headers }, ... ]
     served zero-copy from page cache when the partition is hot.

COMMIT OFFSET  (the durable bookmark; timing selects the semantics)
  commit(offsets={ orders-3: 480200 })    # after processing -> at-least-once
  # or enable auto-commit every N ms for cheaper, coarser commits

SEEK  (replay / reset)
  seek(partition="orders-3", offset=0)            # from the beginning
  seek_to_timestamp(partition="orders-3", ts=...) # from a point in time
  seek_to_end(partition="orders-3")               # tail only, skip history
  isolation_level: read_committed | read_uncommitted   # for transactions

The seek family is the whole point of “log, not queue” surfaced as API - you can rewind and reprocess because nothing was deleted on read.

Data model: the log, the metadata, the offsets

The heart is not a table - it is the on-disk log segment layout, plus small metadata stores for coordination.

On-disk partition log (the source of truth, per partition per broker):

dir: /data/orders-3/
   00000000000000000000.log     # segment: length-prefixed records, appended
   00000000000000000000.index   # sparse offset -> byte-position index
   00000000000000000000.timeindex# sparse timestamp -> offset index (for seek-by-time)
   00000000000000480000.log     # next segment (rolled by size, e.g. 1 GB, or time)
   ...
record: { offset (implicit/sequential), length, crc, timestamp,
          key_len, key, value_len, value, headers,
          producer_id, producer_epoch, sequence }   # last three: idempotent producer

The log is append-only and immutable; retention deletes whole aged-out segments (cheap unlink), and compaction (for compact topics) rewrites segments keeping only the latest record per key.

Cluster metadata (managed by the Raft-based controller quorum):

partition_state:
   (topic, partition) -> { leader_broker, replica_set[], isr[], leader_epoch }
broker_registry:
   broker_id -> { host, port, rack, last_heartbeat, status }
topic_config:
   topic -> { partitions, replication_factor, retention_ms, min_isr, cleanup_policy }

This is small, strongly-consistent, and highly available - a replicated state machine (Raft) is the right tool, not a big database. It answers “who leads orders-3?” and drives leader election. (Older Kafka used ZooKeeper for this; modern Kafka uses a built-in Raft quorum - KRaft - to avoid the external dependency.)

Consumer offsets - internal compacted topic __consumer_offsets:

key:   (group_id, topic, partition)
value: { committed_offset, metadata, commit_timestamp }
policy: log-compacted -> only the latest offset per key is retained.

Storing offsets as a compacted topic (not a side database) is elegant: offsets get the same durability, replication, and partitioning as message data, and compaction keeps only the latest per key so it stays small.

Why this storage split - log segments on local disk, Raft for metadata, a compacted topic for offsets:

SQL vs NoSQL barely applies - the message store is deliberately not a database. It is a purpose-built append-only log because a general database’s random-access, index-per-row, mutable model is exactly what you do not want on a GB/sec sequential-write, sequential-scan path. The only strongly-consistent relational-shaped need (cluster metadata) is small and handled by a Raft state machine, not a SQL server.

Bottlenecks & Scaling

Where it breaks first, in order, and the fix for each:

1. A topic outgrows one broker (throughput and storage). One log on one machine caps at that machine’s NIC and disk. Fix: partitioning - split the topic into N partitions spread across brokers, giving N times the throughput/storage and up to N-way parallel consumption. Partition count is the scaling and parallelism unit; size it generously up front because changing it later remaps keys.

2. Hot partition / hot key. A celebrity user_id or a skewed key hashes a disproportionate share of traffic onto one partition, hotspotting one broker and one consumer. Fix: choose a higher-cardinality partition key (compound key, or user_id + bucket) so load spreads; for a known hot entity, salt the key across K sub-partitions and merge downstream. If ordering per hot entity is not required, round-robin it. Detect skew from per-partition metrics and rebalance leadership off the hot broker.

3. Broker failure loses/stalls a partition. A single-copy partition dies with its broker. Fix: leader-follower replication (RF 3) with an in-sync replica set and acks=all + min.insync.replicas=2. An acked message is on >=2 brokers; on broker death the controller elects a new leader from the ISR in seconds with no committed-message loss. Keep unclean leader election off to prefer consistency.

4. Consumer group stalls or lags. A slow or crashed consumer holds up its partitions; the group falls behind the tail. Fix: more partitions + more consumers (up to one consumer per partition) to raise group throughput; cooperative/incremental rebalancing so one blip does not stop the world; monitor consumer lag (tail offset minus committed offset) as the primary health signal and autoscale consumers on it. Idempotent consumers so you can commit less often and reprocess safely.

5. Duplicate messages (produce retries, consumer reprocessing). At-least-once transport double-delivers on any ack loss or crash-before-commit. Fix: idempotent producer (PID + per-partition sequence numbers) so retried produces are de-duplicated at the broker; idempotent consumers (dedup by message key / idempotency id on the write) so reprocessing is harmless; transactions (read_committed) for exactly-once within the Kafka read-process-write loop.

6. Rebalance storms. Frequent membership changes (autoscaling, flaky consumers) trigger repeated full rebalances that pause consumption. Fix: static membership (stable member ids so a quick restart does not trigger reassignment) + cooperative sticky assignment (only moved partitions pause, most stay put) + tuned session/heartbeat timeouts so a brief GC pause is not read as a death.

7. Metadata/controller as a single point of failure. If the thing that tracks leadership and runs elections dies, the cluster cannot fail over. Fix: run the controller as a replicated Raft quorum (3 or 5 nodes) so it tolerates a minority failure and itself elects a new controller. Keep its state small (leadership, ISR, configs) so it stays fast and available.

8. Read fan-out saturates the leader / network. Many consumer groups reading the same hot partitions concentrate egress on the leader broker. Fix: rely on the page cache + zero-copy so hot reads never hit disk; allow follower fetching (consumers read from a nearby in-sync follower, e.g. same rack/zone) to spread read load and cut cross-zone bandwidth; spread partition leadership evenly across brokers so no one broker leads all the hot partitions.

9. Storage growth (petabytes of log). Retention at GB/sec fills disks fast. Fix: time/size-based retention (drop aged-out segments by unlinking whole files - cheap), log compaction for changelog topics (keep only the latest per key), and tiered storage - keep the recent hot tail on local disk and offload older segments to cheap object storage, still replayable, so retention is decoupled from local disk size.

10. Too many partitions. Every partition is open files, replication streams, and metadata; tens of thousands per broker strain memory, file handles, and failover time (more leaders to re-elect). Fix: cap partitions per broker, consolidate low-traffic topics, and scale the cluster out (more brokers) rather than exploding partition count on a few. Treat partition count as a budgeted resource, not a free knob.

Wrap-Up

The trade-offs that define this design:

• Log, not queue. We store messages as an immutable, append-only commit log with consumer-owned offsets, not a mutable queue that deletes on read. We trade the simple “pop and forget” model for durability, replay, fan-out, and sequential-IO throughput all at once. This single reframe is the whole design.
• Partitioning buys scale, costs global order. Splitting a topic into N partitions gives horizontal throughput and N-way parallel consumption, at the price of ordering being per-partition (per-key), never topic-global - so the partition key becomes the most important and least reversible schema decision.
• Consumers own their position. Progress is a committed offset the consumer advances, stored durably in the cluster; consumer groups split partitions for shared load while separate groups get independent full copies for fan-out. The timing of the offset commit relative to processing is exactly what selects at-least-once versus at-most-once.
• Replication for durability, acks for the dial. Leader-follower replication with an in-sync replica set and acks=all makes an acknowledged message survive any single broker death; failover elects a new leader from the ISR with no committed-message loss. Unclean leader election is CAP as a config flag - we default to consistency.
• Exactly-once is built, not free. The transport is at-least-once; exactly-once is that plus idempotence - producer sequence numbers to de-dupe retries, transactions for the read-process-write loop, and idempotent consumers for external sinks. In practice, at-least-once plus an idempotent consumer is the pattern that covers most real systems.

One-line summary: a cluster of brokers storing each topic as N partitioned, append-only commit logs replicated leader-to-follower with an in-sync replica set, producers writing to partition leaders under a tunable acks durability dial, consumer groups splitting partitions for shared parallel reads while tracking their own durable offsets for replay and fan-out, and delivery semantics - at-most-once, at-least-once, exactly-once - falling out of where you commit the offset plus idempotent producers and transactions layered on top of an inherently at-least-once transport.