How to design a distributed message queue

In this lesson, we’ll design a distributed message queue and watch nearly every property of it fall out of a single design choice. A message queue sits between a program that produces work and a program that consumes it: the producer hands a message off and returns immediately, without waiting for the work to finish. By the end you’ll be able to size the fleet from a workload, name the one decision the whole system turns on, and say exactly where these systems fail.

Here is that decision up front, so you can watch it pay off. What we are building is not really a queue but an append-only log: new records are written at the end, and nothing already written is ever modified or deleted in place. Almost every useful property of a system like Kafka follows from that one choice. (“Kafka” and “distributed log” are used interchangeably here; nothing below depends on Kafka specifically.)

Terms

• A topic is a named stream of messages, like clicks or payments.
• A partition is one of the independent append-only files a topic’s data is split across.
• An offset is the position of a record within one partition, counting from zero, assigned at the moment of the append.
• A broker is a machine that stores partitions and serves them.

The interface

The whole system is two calls plus a bookmark:

• A producer calls produce(topic, key, value) and gets back a partition number and an offset.
• A consumer calls fetch(topic, partition, offset) and gets the records stored from that position onward, in order.
• A consumer records how far it has read with commit(group, topic, partition, offset).

One detail here carries everything downstream: the reader supplies the offset. The system does not remember where each reader got to; the consumer does.

The one decision: who owns the read position

That detail is the entire design, so let’s make it explicit. A message queue’s central choice is who owns the read position, the bookmark recording how far a reader has got. The two answers build completely different systems.

• If the broker owns it, it must track per-message state for every reader. Deleting data becomes a distributed problem: the broker can only drop a record once it knows every reader is done with it. And a message that has been read is gone.
• If the consumer owns it, the bookmark is a single integer the consumer durably records (“I have finished everything before this position”). The broker’s job collapses to “append bytes, serve bytes by offset.”

We take the second answer, and everything downstream is a consequence of it.

That choice moves where breakage can happen. What breaks in these systems is never throughput; the sizing below shows disk running at about 2% of capacity. The failures all sit on the consumer side:

• a consumer group (a set of consumers that divide a topic’s partitions between them and share one set of bookmarks) that cannot keep up;
• a rebalance (reassigning partitions when group membership changes) that stalls every member at once;
• a retention window (the age at which the broker deletes old records regardless of who has read them) that expires data a slow consumer had not reached yet.

The back half of the lesson is those three failures, priced. First we pin down the requirements they have to hold up.

Requirements

Functional

• produce(topic, key, value) → (partition, offset). The key selects the partition, so all records for one key land together. Offsets increase only within a partition; offsets from different partitions cannot be compared.
• fetch(topic, partition, offset, max_bytes) → a batch starting at offset. Because the consumer names the offset, replay (deliberately re-reading messages you already processed) is just a parameter.
• commit(group, topic, partition, offset): the durable read position for one consumer group.
• Retention by time or size, independent of whether anyone read the data.
• Multiple independent consumer groups per topic, each with its own offsets, so a second reader never disturbs the first.

Non-functional

A few terms these use: RF 3 (replication factor 3) means every record lives on three machines, one leader that accepts writes and two followers that copy it. min.insync.replicas=2 means a write is not acknowledged until at least two copies hold it. p99 is the latency 99 out of 100 requests come in under.

These targets set the sizing, and the sizing tells us which of them binds first.

Back-of-envelope sizing

We size here less to count machines than to find which resource runs out first, because that resource is the one the rest of the design has to respect. Assumptions: 1,000,000 msg/s at 1 KB, RF 3, 7-day retention, 3 consumer groups.

Storage. 1 M/s × 1 KB × 86,400 s/day × 7 days × 3 copies ≈ 1.81 PB across the cluster. Big, but not the binding constraint, as the next two numbers show.

Network. Each produced byte crosses a network interface eight times (in once, out to two followers, in at two followers, out to three consumer groups), so 1 GB/s of ingest becomes 8 GB/s cluster-wide. That factor of eight is why the network, not the disk, is the thing to watch.

Brokers. Taking a conservative 1 Gbps NIC (125 MB/s) and budgeting 60% for the steady path (the rest absorbs a lagging consumer catching up and rebalance bursts): 8 GB/s ÷ 75 MB/s ≈ 107 brokers. Round up to 128 (a power of two makes partition-to-broker assignment even).

Which resource binds. On that fleet each broker’s NIC runs at about 50% while its sequential disk write sits at about 2.3% of a drive’s bandwidth. Line those two up and the answer is stark: the network runs out first, by a wide margin. A message broker is a network device that happens to persist.

The word sequential is doing real work in that last sentence. The log only ever appends to the end of a file, so the drive never seeks. A modern NVMe SSD is roughly 10x faster at sequential than at random I/O, and this design only ever does sequential, which is exactly what the log-versus-table comparison below turns on.

This is a hardware answer, not a workload one: a 10 Gbps NIC would drop the fleet from 107 to 11 brokers for the same workload. In an interview, state which end of the range you took; the rest of this lesson is sized on the 1 Gbps, 128-broker fleet. Now that we know the log’s shape is what saves us, let’s lay out that shape on disk.

Data model: the log

Segments. A partition is not one enormous file but a directory of segment files, each rolled shut at a fixed size (1 GiB here) and never touched again. Every record carries a 24-byte header (offset, length, a CRC checksum, timestamp) before its payload, so a 1 KB message occupies 1,024 bytes on disk and a 1 GiB segment holds about 1,048,576 of them. (Config sizes like segments are binary units, 1 GiB is 1,073,741,824 bytes, while capacity estimates use decimal GB; mixing the two silently is a ~7% error.)

Indexes. Alongside each segment sit two small lookup files: an offset index (.index, mapping offset → byte position) and a time index (.timeindex, mapping timestamp → offset, which makes “start me at 9 a.m.” expressible). Both are sparse (one entry per 4 KB of log) so they stay in memory. A fetch(offset) is therefore a binary search over the sparse index plus a short forward scan: a logarithmic number of comparisons and one seek, not a scan of the whole partition.

One mutable integer. Nothing is ever updated in place: no per-message “read” flag, no tombstones. The only mutable state in the entire system is one integer per consumer group per partition, the committed offset. Even that lives in an ordinary compacted topic, __consumer_offsets, where the broker keeps only the newest record per key, so a group’s bookmark takes constant space no matter how often it commits. Once you accept that the only mutable state is one integer per group per partition, the rest of the machine is just wiring, so let’s follow a single message through it.

Architecture

flowchart TB
    P1(["producer"]) --> PA["partitioner<br/>hash(key) % 16,384"]
    PA --> L1["leader p0 · broker 3"]
    PA --> L2["leader p1 · broker 7"]

    L1 --> F1["follower · broker 9"]
    L1 --> F2["follower · broker 41"]
    F1 -. "fetch, then advance HW" .-> L1
    F2 -. "fetch" .-> L1

    subgraph SEG["one partition on one broker"]
        A["append to active segment<br/>sequential, page cache"] --> R["roll at 1 GiB"]
        R --> D["delete or compact<br/>at the retention edge"]
        IX[".index + .timeindex<br/>sparse: one entry per 4 KB"] -.-> A
    end
    L1 --> SEG

    SEG --> CG1["group A · 12,000 consumers<br/>committing offsets to __consumer_offsets"]
    SEG --> CG2["group B · analytics<br/>independent offsets"]

    CTRL["controller quorum<br/>ISR, leadership, metadata"] -.-> L1
    CTRL -.-> L2

    style SEG fill:#1d3557,color:#fff
    style CTRL fill:#495057,color:#fff

Following one message through: a producer’s record goes to a partitioner (a few lines of arithmetic in the producer library, not a server) that computes hash(key) % partition_count to pick a partition. Each partition has one leader, the only machine allowed to append to it, which writes to the tail of the active segment, rolls the segment shut at 1 GiB, and eventually deletes or compacts it at the retention edge. Two followers copy the leader using the same fetch call a consumer uses, so replication needs no separate protocol. Consumer groups read the same bytes without interfering: group A is the online fleet committing offsets, group B an analytics job that can be days behind without group A noticing. Off to the side, a controller quorum agrees among itself, via a consensus protocol, on which broker leads which partition and which replicas are caught up; no message data flows through it.

Why a log beats a table

We just claimed the log shape saves the fleet. Let’s put a number on it by costing out the design most teams reach for first: a database table with a status column, INSERT to enqueue, SELECT ... FOR UPDATE to claim a job exclusively, UPDATE/DELETE to mark it finished. The append-only shape turns out to be a roughly 300x hardware difference against it, and the reason is worth holding onto.

The intuition first: a database never touches one record in isolation. It reads and writes fixed-size pages (4 KB), never single rows, so a one-byte change costs a whole page. Moving one message through that table touches about eight pages at random locations on the drive: the row write on insert, two index-leaf writes on insert, the claim read, the claim status write, the ack status write, the ack index update, and the vacuum that later reclaims the dead row. That is eight random 4 KB I/Os per 1 KB message. (The write-ahead log, the sequential file a database appends to before touching its real structures, is sequential in both designs, so it cancels out. The database’s problem is everything it does after the log.)

Order of magnitude: at ~100 MB/s random, eight 4 KB pages is ~3,000 msg/s per device, so 1 M msg/s needs about 328 devices. The log appends 1,024 bytes sequentially at ~1 GB/s, ~976,000 msg/s per device, so one device carries the whole load. That is roughly 320x: 10x from the random-versus-sequential bandwidth gap, and another ~32x from write amplification (1 KB of payload landing in a 4 KB page, times two indexes, times the read-modify-write on the claim and the ack).

Three consequences of the log shape follow, none of them a feature anyone had to build:

• Replay is free. The consumer supplies the offset, so reprocessing yesterday is one commit at yesterday’s offset plus a restart. The table design deleted the rows, so replay there requires a second copy kept somewhere, a log built after all, badly.

• A second consumer group is free. Its entire cost is one integer per partition (16,384 × 8 bytes ≈ 131 KB) against a delete-on-read queue that must store a separate copy per reader (a day of stream ≈ 86 TB). This is why one Kafka cluster can feed the online service, the search indexer, and the warehouse loader at once, while the broker-tracks-every-message design (AMQP, which RabbitMQ implements) ends up storing a copy of the stream per consumer.

• Reads come out of RAM, until one consumer ruins it. The operating system keeps recently written file data in spare RAM, the page cache, so a consumer reading near the tail is served straight from memory by sendfile (a kernel call that copies file bytes to a socket without passing through the broker process). With ~100 GB of spare RAM per broker against ~23 MB/s of log writes, the cache holds roughly 1.19 hours of log. A consumer reading older data than that reads from disk, fine on its own at 1 GB/s, but the old pages it pulls in evict the pages every other consumer was using. One lagging consumer flips the whole broker’s read path from memory to disk. That is why consumer lag (the gap between the newest record and a group’s committed offset) is the primary health metric, not a dashboard nicety. This failure mode is where these systems actually die, and it returns in full later. First we need the knob that sets how many consumers can share the load at all.

Partitions: parallelism and ordering are the same knob

Here is the knob, and its most important property is that it only turns one way: partition count can be raised but never lowered. Lowering it changes P in hash(key) % P and re-routes every existing key, destroying per-key ordering. A one-way decision is worth deriving instead of guessing, so let’s derive it.

Ordering exists inside a partition and nowhere else. Within a partition, offsets are assigned by the leader on append, so ordering is total. Across partitions there is no shared clock, no shared sequence, and no way to say which of two records came first: they were appended by different machines and are read independently.

Global ordering is expensive. Ordering every record in a topic against every other requires exactly one partition, hence one broker whose NIC carries four copies of the stream (in, out to two followers, out to one group): 125 MB/s ÷ 4 ÷ 1 KB ≈ 31,250 msg/s, about 3% of the target. Worse, one partition means one consumer (a partition goes to at most one member of a group) so at 10 ms of processing per message the group caps at 100 msg/s, about 300x below what the single broker could serve. The fix is to order per key: hash(key) % P keeps every event for account 42 in order forever, and no realistic requirement needs account 42 ordered against account 99.

Partition count is set by consumer speed, not throughput. Two different quantities meet here, so name them before dividing. 1 M msg/s is offered load (work arriving); one consumer’s 100 msg/s is service capacity (work it can finish). Dividing them gives 10,000 consumers at 100% utilization, zero headroom, so any backlog that ever forms drains at a rate of zero and stays forever. Provisioning 20% headroom (1.2 × load) needs 12,000 consumers, and since a consumer with no partition assigned to it is idle, the partition count must be at least that. Round up to 16,384. The maximum useful consumer count is the partition count, the 16,385th consumer adds nothing.

Many partitions send the bill to the producer. Producers batch per partition, which only works if each producer sends to a given partition often enough to fill a batch. With 1,000 producers and 16,384 partitions, each producer hits any partition about 0.06 times a second: no batch ever fills, so every message becomes its own request. The sticky partitioner fixes this for keyless records: the producer sticks to one partition until its batch is full, sends it, then moves on. A 16 KiB batch holds 16 records, cutting the cluster request rate about 16x (roughly 1,000,000 down to 63,000 requests/s). Keyed records cannot use it, since their partition is fixed by the key.

Other limits are operational: 49,152 partition replicas for the controller to track, about 1,152 open file descriptors per broker, and a controller failover that must elect a new leader for every affected partition. All argue for picking the partition count once, with the consumer-latency arithmetic above, not doubling it during an incident. With the data laid out and the parallelism fixed, the next question is what the queue actually promises about each message: once, at least once, or at most once.

Delivery semantics

The three delivery guarantees sound like three different systems, but they are not three implementations. They are three orderings of the same two operations a consumer performs, apply the effect (do the real work, such as charging a card) and commit the offset (record it done), plus the question of what happens if the process dies between them:

At-least-once is the only defensible default, and its duplicates can be priced. On restart a consumer re-processes everything between its last commit and where it actually got to, set by its share of offered load, not capacity. Across the fleet, replayed messages per day is produce_rate × commit_interval × restarts_per_consumer; at a 5 s commit interval and about one restart per consumer per day that is 1,000,000 × 5 × 1 = 5 M/day ≈ 58 duplicates/s, forever. The group size cancels out entirely. So you cannot treat duplicates as an occasional surprise to clean up later; they arrive continuously. Shortening the commit interval to 100 ms cuts that about 50x (to ~1.2/s) at the cost of 120,000 offset commits/s into __consumer_offsets.

Exactly-once across two systems is impossible. If the effect (a row in MySQL, a charge at Stripe) and the offset commit live in different systems, making them agree on a single commit point needs one of two things, neither available here: two-phase commit (a coordinator asks every participant to prepare, then to commit, and blocks every participant indefinitely if it dies between the phases), or an infinite exchange of acknowledgements (the Two Generals result: two parties over a lossy link can never both become certain they agree, because the last acknowledgement always needs one of its own). The notification-system chapter works this through in full.

What Kafka’s transactions actually give is narrower than the marketing, and it is two separate mechanisms:

• The idempotent producer. Each producer gets a (producer id, epoch) (the epoch is a generation counter that increases each time the identity is re-established, so a stale incarnation is rejected) and stamps a monotonic sequence number on every record; the leader drops any sequence number it has already seen. This removes duplicates from producer retries only.
• Transactions. A coordinator writes begin/commit markers into the data partitions and into __consumer_offsets as one atomic step, and a read_committed consumer refuses to read past the earliest still-open transaction. So the atomic unit is consume-from-Kafka, produce-to-Kafka, commit-offset, all three endpoints inside the log. The instant one endpoint is your database, there is no marker to write there and the guarantee evaporates. Kafka’s exactly-once is exactly-once stream processing, not exactly-once delivery. It also adds about 50 ms of mean latency, since a read_committed consumer waits on average half a commit interval at the open-transaction boundary.

The idempotent consumer: what to build when the sink is a database

When the destination is a database, the answer is an idempotent consumer: in the same database transaction that performs the effect, insert a dedup key on a uniquely-constrained column. Because the two writes commit or roll back together, a redelivered message finds its key already present and does nothing.

The key must be something the producer minted and re-sends unchanged, not topic:partition:offset. A log position is not a message identity: after an unclean leader election (next section) the log gets shorter, offsets go backwards, and the idempotent producer re-sends, so the same logical payment reappears at a different offset, presenting a key nobody has seen, and is applied twice.

import sqlite3


class NoOpEffect(Exception):
    """Effect matched no row. Raising rolls back the whole transaction,
    including the dedup key, so a redelivery is still allowed to fix it."""


def handle(conn, record) -> str:
    # Key on the PRODUCER's identity for the message, not the offset it sits at.
    key = f"{record['producer_id']}:{record['idempotency_key']}"
    with conn:                                     # one transaction
        cur = conn.execute(
            "INSERT OR IGNORE INTO processed(key) VALUES (?)", (key,))
        if cur.rowcount == 0:
            return "duplicate"                     # already applied; do nothing
        cur = conn.execute(
            "UPDATE balances SET cents = cents + ? WHERE id = ?",
            (record["amount"], record["account"]))
        if cur.rowcount != 1:                      # matched no row: a silent no-op
            raise NoOpEffect(record["account"])
    return "applied"


db = sqlite3.connect(":memory:")
db.execute("CREATE TABLE processed(key TEXT PRIMARY KEY)")
db.execute("CREATE TABLE balances(id INTEGER PRIMARY KEY, cents INTEGER)")
db.execute("INSERT INTO balances VALUES (1, 0)")

r = {"producer_id": "p7", "idempotency_key": "txn-9",
     "topic": "t", "partition": 0, "offset": 7, "amount": 250, "account": 1}
assert handle(db, r) == "applied"
assert handle(db, r) == "duplicate"
assert db.execute("SELECT cents FROM balances WHERE id = 1").fetchone()[0] == 250

# Unclean election: the log truncated and the same logical message now sits at
# offset 3. Keyed on the offset this applies twice; keyed on producer identity
# it does not.
resent = dict(r, offset=3)
assert handle(db, resent) == "duplicate"
assert db.execute("SELECT cents FROM balances WHERE id = 1").fetchone()[0] == 250

# A no-op effect must not leave a dedup key behind, or it credits nobody and
# then refuses the redelivery that would have fixed it.
orphan = dict(r, idempotency_key="txn-10", offset=8, account=999)
try:
    handle(db, orphan)
    raise AssertionError("an effect that matched no row must not commit its key")
except NoOpEffect:
    pass
assert db.execute("SELECT count(*) FROM processed "
                  "WHERE key = 'p7:txn-10'").fetchone()[0] == 0

The second load-bearing detail is the rowcount check on the effect: an UPDATE matching no row does not raise, so without the check it would commit a dedup key for work never done and then block the redelivery that would have fixed it.

How long must the dedup table remember? Not forever, only the redelivery horizon: the session timeout (how long a consumer may go without a heartbeat before the coordinator declares it dead) plus the rebalance time. Past that point the partition has been handed to someone else and the system will never re-offer that message. At 45 s + 15 s = 60 s and 1 M msg/s × 16 bytes/key that is about 960 MB, one Redis instance or an embedded RocksDB store. Setting the window to 24 hours “to be safe” makes the same table ~1.38 TB. And the horizon covers system redeliveries only; a deliberate replay is meant to re-apply, so making replay itself idempotent needs a table sized by the full retention window (~9.7 TB), which is a different and much larger design. Decide which one you are building.

The idempotent consumer only works if we can trust that a committed offset never moves backwards, and that trust comes from how the log is replicated. That is the next piece.

Replication, ISR, and the acks knob

The whole scheme rests on one rule about what a consumer is allowed to see, so let’s build up to it. A follower that has fetched everything the leader held as of replica.lag.time.max.ms ago is in-sync; the set of such replicas is the in-sync replica set (ISR), and membership is dynamic, a follower that falls behind drops out and rejoins when it catches up. The leader tracks a high watermark, the highest offset present on every ISR member, and consumers may not read past it. That single rule is what makes failover safe: everything a consumer has seen already exists on every in-sync replica, so promoting a follower can never make a record a consumer read disappear.

The acks knob selects how many replicas the producer waits for before treating a send as successful:

The acks=1 loss is per-broker ingest (1 M ÷ 128 ≈ 7,813 msg/s) times the ~10 ms follower fetch lag ≈ 78 messages. acks=0 risks the entire 32 MiB producer buffer (about 33,554 messages, roughly 33 s of one producer’s output) held in memory with nothing durable behind it.

The two all rows share a latency because the two followers fetch concurrently; requiring the second one to hold the record buys the slower of two draws from the same distribution, not another round trip. So min.insync.replicas=3 at RF 3 costs availability, not latency. It is the same mistake as requiring W = N in a quorum store, which the key-value store chapter already prices, plus a second failure mode: if one replica drops out of the ISR the size falls to 2, the requirement can’t be met, and every producer to that partition gets NotEnoughReplicas until the follower catches up.

acks=all is not the same promise as “on disk.” Every append (leader or follower) lands in the OS’s page cache first. Kafka does not call fsync (the system call that forces cached writes onto the physical drive) on every record, because that would put a disk round trip back on the latency path the acks table just priced. So acks=all guarantees the record survives one broker dying; it does not guarantee the bytes have reached a drive anywhere. Two replicas lost at once, before the OS flushes its page cache, can still lose data that was fully acked.

Unclean leader election is the failure worth understanding precisely. A clean election promotes an ISR member, which holds everything consumers have seen. An unclean one, allowed by unclean.leader.election.enable=true, promotes a replica that had fallen out of the ISR because no ISR member is available. By definition that replica is more than replica.lag.time.max.ms (~30 s) behind, so promoting it truncates about 30 s × 61 msg/s per partition × 128 partitions ≈ 234,240 messages, silently. Worse than the loss is that the log got shorter: a consumer that committed offset 5,000 now points past the end of a log that ends at 3,170, so the broker resets it via auto.offset.reset (which decides where a consumer starts when its offset is invalid) and the records in between reach no one. No error, no exception, no red metric. This is exactly why the idempotent consumer keys on a producer-supplied id, not the offset.

The alternative, unclean.leader.election.enable=false, keeps the partition unavailable until an ISR member returns. That boolean is the choice between availability and durability written into a config file: false for payments, true for clickstream.

sequenceDiagram
    participant P as producer
    participant L as leader
    participant F1 as follower in ISR
    participant F2 as follower lagging
    P->>L: produce, acks=all, min.insync.replicas 2
    L->>L: append to active segment, page cache only
    F1->>L: fetch at offset n
    L-->>F1: records
    F1->>L: fetch at offset n+k implies it holds k
    L->>L: ISR is L and F1, so the high watermark advances
    L-->>P: ack, about 1 ms
    Note over F2: more than 30 s behind, dropped from ISR
    Note over L,F2: L and F1 both die. Promoting F2 truncates 1,830 records per partition

Consumer lag, backpressure, and the rebalance storm

This is where these systems actually fail, so it is the part worth understanding coldly. Sizing said the brokers are fine; every real incident lives on the consumer side, and it shows up as lag.

Lag is the metric that matters, because only its rate of change predicts trouble. Throughput looks perfect right up to the moment the backlog grows without bound. A saturated system still moves messages at full speed, it simply never catches up. Three forms, in increasing usefulness:

lag         = log_end_offset - committed_offset   (messages)
lag_seconds = lag / consume_rate                  (what a human acts on)
d(lag)/dt   = produce_rate - consume_rate         (what predicts)

The sign of d(lag)/dt is the whole diagnosis: positive means already broken and no waiting fixes it; negative means a finite drain time.

Drain time is set by headroom, not throughput. A 30 s stall builds a backlog of 30 s of production, and it drains at capacity minus the load still arriving. At 20% headroom: 30 M ÷ (1.2 M − 1 M) = 150 s. At 5% headroom: 30 M ÷ (1.05 M − 1 M) = 600 s. Same hardware, same throughput; 4x less headroom is 4x slower recovery, and at zero headroom the drain rate is zero and the backlog never clears. (600 s of lag is still inside the 1.19-hour page-cache window, so this recovery runs at memory speed; a backlog that exceeds that window reads from disk and evicts everyone else’s cache, and lag stops being linear.)

Backpressure lives on the producer, not the broker. The broker has no way to tell a producer to slow down; it can only be slow to respond. The real mechanism is the producer’s bounded buffer: send() returns immediately while there is room, blocks for up to max.block.ms (60 s) once the buffer fills, then throws an exception into the application. So the ~33 s of buffer is a hard operational limit: you have about 33 s of broker unavailability before application threads start seeing exceptions. That number, not “five nines,” is what a broker-recovery runbook must beat.

The rebalance storm. When group membership changes, the group coordinator (the broker tracking who is in the group and who owns which partition) revokes assignments and hands out new ones; under the original eager protocol every member stops consuming for the whole round, however small the change. A consumer that takes longer than max.poll.interval.ms between poll() calls is presumed dead and ejected, which triggers a rebalance, which stalls everyone, which grows the backlog, which makes the next batch bigger and slower, which ejects the consumer again:

flowchart LR
    A["consumer slow to call poll"] --> B["ejected from group"]
    B --> C["rebalance stalls every member"]
    C --> D["backlog grows"]
    D --> E["next batch is larger"]
    E --> F["batch takes longer to process"]
    F --> A

Staying safe means keeping max.poll.records × seconds_per_record < max.poll.interval.ms. Defaults (500 records × 10 ms = 5 s against a 300 s limit) sit 60x inside it; the storm happens when someone raises max.poll.records to 50,000 to “drain the backlog faster” and crosses the ~30,000 ceiling. The cure is the opposite of the instinct: lower the batch, or move the processing to a separate thread and use the client’s pause/resume so poll() keeps being called without delivering more work than the consumer can absorb.

Two structural fixes go further. Cooperative (incremental) rebalancing revokes only the partitions that actually move, so adding one consumer to a 12,000-member group disturbs about one consumer’s share instead of all 16,384 partitions, roughly N+1 ≈ 12,000x less disruption. Static membership gives each consumer a stable group.instance.id that survives a restart, so a consumer that returns within its session timeout rejoins with the same assignment and triggers no rebalance at all, turning a rolling deploy of 12,000 consumers from 12,000 rebalances into zero.

All of this assumes the consumer is the one asking for data. That is a choice, and it is the next one to defend.

Push vs pull

Push means the broker decides when and how much each consumer receives; pull means the consumer asks. We chose pull, and it wins for three reasons, only the last about performance:

• Backpressure belongs where capacity is known. A pushing broker facing a slow consumer must buffer (unbounded broker memory, the failure on the wrong machine) or drop. A pulling consumer that is slow simply does not ask, and the backlog sits on disk, bounded by retention and measurable as lag.
• Batching belongs to whoever knows its own capacity. The consumer requests up to fetch.max.bytes; a broker would have to guess, and a wrong guess is either no batching or an overrun.
• Replay is only expressible as pull: fetch(offset) with any offset is the replay feature.

Pull’s one real cost is idle polling: a consumer with nothing to do still asks, and every empty answer is a wasted round trip. Long poll removes it: the broker holds the request open for up to fetch.max.wait.ms and replies the moment data arrives. A naive 100 ms poll across 12,000 consumers on 2 brokers each is ~240,000 requests/s; a 500 ms long poll is ~48,000, 5x fewer requests and lower latency at the same time, because the request rate stops depending on the poll interval and starts depending on the data rate.

We now have every piece of the system. Let’s collect where each one runs out and what you do about it.

Bottlenecks and scaling

Two terms: KRaft is Kafka’s built-in metadata consensus, which replaced ZooKeeper (a separate coordination service operated alongside the brokers). RTT is round-trip time.

Growth is asymmetric. Adding brokers is easy; filling them is not, because a new broker starts empty and existing partitions do not migrate on their own. Moving one broker’s 14.18 TB over its spare half-NIC (62.5 MB/s) takes about 2.6 days. Plan capacity in “days to rebalance,” not “hours to provision.”

Failure modes

Three terms first: a poison message fails its consumer every time it is processed, blocking the partition behind it; a dead-letter topic is where you divert such a record after bounded retries so healthy traffic continues; NTP is the Network Time Protocol that keeps machine clocks roughly agreed.

The retention row is the one to watch, because every other failure here is loud while “the consumer was down for eight days and the data aged out” is silent. The alarm that catches it measures lag in seconds of retention consumed, not messages.

Alternatives considered

Three terms: SQS is Amazon’s managed queue; Pulsar is a competing log system that separates the serving brokers from the storage layer; Raft is a consensus algorithm that lets a group of machines agree on an ordered sequence of decisions while some fail.

The managed option, honestly. At $0.40 per million requests (batches of 10) versus $1.00 per broker-hour, a year is about $1.26 M managed against $1.12 M of raw instances, roughly 13% apart, a wash, and the instance figure excludes the engineers who run them. So on price, managed is competitive and probably ahead once salaries are counted. You reject it on capability, not price: SQS-style queues offer no replay, no independent consumer groups over the same data, at-least-once only, and FIFO ordering only within a single message group at a few hundred TPS, each of which is a stated requirement here.

The load-bearing assumptions

The design is only correct relative to its assumptions. The ones that would invalidate it (where a whole component appears or disappears if they are false, not just the machine count) are:

• The consumer owns its read position; the broker keeps no per-message state. This is what makes replay, independent consumer groups, and the append-only shape possible. If the product needs per-message acknowledgement, per-message delay, or priority within a stream, the broker must hold mutable per-message state and you are building the AMQP design instead.
• Ordering is per key, never global. This is what makes partitions legal. A genuine global-order requirement forces one partition (100 msg/s at the consumer) and this becomes a single-machine problem.
• Consumers read sequentially, near the tail. This is what keeps the disk at ~2%, reads in the page cache, and sendfile in play. Selective or priority consumption makes reads random, and the 328-devices penalty applies to your design.
• The message’s destination is outside the log (a database, a payment provider, an email). This is why exactly-once is unavailable and the idempotent consumer is mandatory. If every endpoint is inside the log, Kafka transactions give real atomicity and the dedup table disappears.

The one number worth pinning down early is per-message processing time (~10 ms here): it sets the consumer count and therefore the partition count that can never be lowered. At 10 µs instead of 10 ms the group is 12 consumers, not 12,000.

Conclusion

• A distributed message queue is best built as an append-only log, not a queue. The single decision everything follows from is letting the consumer own its read position, which reduces the broker to “append bytes, serve bytes by offset.”
• That shape makes replay and extra consumer groups nearly free, keeps writes sequential (one device instead of ~328), and serves most reads straight from the page cache.
• Partitions are the unit of both ordering and parallelism. Order per key; global ordering caps the system near 100 msg/s. Partition count is set by consumer speed plus headroom and cannot be lowered, so derive it once.
• Delivery is a choice, not a property: at-least-once is the honest default (~58 duplicates/s here), and an idempotent consumer keyed on a producer-supplied id gives effectively-once when the sink is a database.
• These systems fail on the consumer side (lag, rebalance storms, and unread data aging out), not on broker throughput. Watch d(lag)/dt and lag in seconds of retention.

One line to remember: hand the reader the offset, and a message queue becomes an append-only log where replay and extra consumers are nearly free and the only thing you have to watch is how fast a consumer falls behind.

Further reading

• Jay Kreps, “The Log: What every software engineer should know about real-time data’s unifying abstraction” (LinkedIn Engineering, 2013).
• Apache Kafka documentation, the Design and Implementation sections (kafka.apache.org/documentation).
• Kreps, Narkhede, and Rao, “Kafka: a Distributed Messaging System for Log Processing” (NetDB, 2011).
• Martin Kleppmann, Designing Data-Intensive Applications, chapter 11, “Stream Processing.”
• Background on this site: the database internals chapter for the write amplification behind the log-versus-table result; the key-value store chapter for the quorum arithmetic; the notification-system chapter for the exactly-once impossibility; and the scaling chapter, where the queue first appears as a featureless box.