20 — Design A Distributed Message Queue

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, instead of waiting for the work to finish.

Chapter 01 put such a queue between the web tier and the slow background work and drew it as a featureless box. This chapter opens it up and shows how to:

• explain why a queue at this scale is really an append-only file;
• size a cluster for a million messages a second from first principles;
• price each of the three delivery guarantees in duplicate messages per second;
• name the failure that actually takes these systems down — which is not the one most people expect.

The core idea is that this is not a queue but an append-only file. New records are written at the end, and nothing already written is ever modified or deleted in place. Every property people value in Kafka follows from that one choice.

(Kafka is the open-source system that made this design mainstream. “A Kafka” and “a distributed log” mean the same thing throughout this chapter, and nothing below depends on Kafka specifically.)

What goes in, and what comes out

Two calls define the entire system. Four terms come first; the rest of the chapter uses them throughout.

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

Now the two calls:

• A producer — any program with a message to hand off — calls produce(topic, key, value) with a few hundred bytes and gets back two integers: a partition number and an offset.
• A consumer — any program that wants those messages — calls fetch(topic, partition, offset) and gets back the records stored from that position onward, in order.

That is the whole interface. Everything below is a consequence of one detail in it: the reader supplies the offset, rather than the system remembering where each reader got to.

The five numbers

Each row below is one section’s headline result, with the number it produces and where it is derived.

Three results borrowed from elsewhere

These three are used but not re-derived. Each is restated here in a sentence, so you can read this chapter without opening any of them, and linked in case you want the derivation.

1. Storage-engine mechanics, from sql 03. Four terms:

• write-ahead log (WAL) — a sequential file a database appends to before touching its real data structures, so a crash can be replayed;
• log-structured merge tree (LSM) — a storage engine built from sorted files that are periodically merged;
• compaction — that merging;
• write amplification — the ratio of bytes actually written to disk to the bytes the application asked to store.

2. Quorum arithmetic, from The quorum what w and r actually buy. A quorum is how many replicas of a piece of data a write must reach, and a read must consult, before the answer is guaranteed current.

3. The impossibility of exactly-once, from Deep dive 3 exactly once is not available. Exactly-once delivery means every message takes effect once and only once; that chapter proves it is unattainable across two independent systems.

1. Framing: what decision, and what breaks

The whole design turns on one decision. From it follow the requirements everything later is sized against, and the back-of-envelope arithmetic that says how many machines the cluster needs and which resource on them runs out first.

The one decision

A message queue makes one interesting decision: who owns the read position — the bookmark recording how far a given reader has got.

There are two answers, and they build completely different systems.

If the broker owns the bookmark, it must track per-message state for every reader. Deleting data becomes a distributed problem, because the broker has to know that everyone is done with a record before it can drop it. And a message that has been read is gone.

If the consumer owns it, the bookmark is a single integer that the consumer commits somewhere — meaning it durably records “I have finished everything before this position.” The broker’s job then collapses to “append bytes, serve bytes by offset.” Everything else in this chapter follows from that collapse.

What actually breaks

It is never throughput. Brokers are network appliances that happen to write files, and the arithmetic below shows disk running at 2% of capacity.

What breaks is the consumer side, in three ways. Each gets a name now and a derivation later.

• A consumer group — a set of consumer processes that divide a topic’s partitions between them and share one set of bookmarks — that cannot keep up.
• A rebalance — the reassignment of partitions among a group’s members, triggered whenever the membership changes — that stalls every member at once.
• A retention window — the age at which the broker deletes old records whether or not anybody has read them — that expires data a slow consumer had not reached yet.

The interesting failure in this system is on the consumer, not the broker.

Requirements

The functional list is the interface; the non-functional table is the set of numbers every later derivation is anchored to.

Functional

• produce(topic, key, value) -> (partition, offset). The key selects the partition, so all the records for one key land in one place. The offset is assigned by the machine that owns that partition, and it only ever increases within a partition — offsets from two different partitions cannot be compared at all.
• fetch(topic, partition, offset, max_bytes) -> a batch of records starting at offset. The consumer names the offset, so replay — deliberately re-reading messages you already processed, in order to process them again — is a parameter, not a feature.
• commit(group, topic, partition, offset) — the durable read position for one consumer group, stored by the system on the group’s behalf.
• Retention by time or size, independent of whether anyone has read the data.
• Multiple independent consumer groups on the same topic, each with its own offsets, so adding a second reader never disturbs the first.

Non-functional

Three of those entries use terms that have to be pinned down before the arithmetic starts.

• RF 3 is a replication factor of three: every record is stored on three different machines. One of them is the leader, which accepts writes; the other two are followers, which copy from it.
• min.insync.replicas=2 says a write is not acknowledged until at least two of those three copies hold it.
• p99 means the 99th percentile: the latency that 99 out of 100 requests come in under. So p99 < 5 ms is a promise about the slowest one request in a hundred, not about the average.

Back-of-envelope

Turning the requirements into a machine count matters less for the count itself than for what it uncovers: which resource on those machines is the scarce one.

Assumptions stated once: 1,000,000 messages/s peak, 1 KB each, replication factor 3, 7-day retention, 3 consumer groups.

Seconds per day, because it changes answers by 16%. There are 86,400 seconds in a day. Ch 02 allows rounding that to 1e5 (100,000) while reasoning, because dividing by 100,000 is just moving a decimal point. It forbids 1e5 for any reported result — a fleet size, a monthly bill, a capacity headroom. The reason: 1e5 is 16% higher than 86,400, so a per-second figure computed from it is 13.6% too low. Every result in this chapter is divided by 86,400, and the places where that matters are flagged.

Storage first. Multiply the message rate up to bytes, then out to a week, then by the three copies:

ingest bytes/s:       1,000,000 x 1,000                          =  1,000,000,000
per day:              1,000,000,000 x 86,400                     =  86,400,000,000,000
7-day retention:      86,400,000,000,000 x 7                     =  604,800,000,000,000
replicated RF 3:      604,800,000,000,000 x 3                    =  1,814,400,000,000,000

That last figure is 1.81 petabytes — 1.81 million gigabytes — of disk across the cluster. (Estimated with 1e5 it comes out as 2.1 PB, and the 16% gap between the two is why the reporting convention exists.)

It is a big number and it is not the interesting one.

Network next, and this is the one that binds. Count every time a produced byte crosses a network interface. It is written once by the leader, sent out to two followers, received by those two followers, and then read once by each of the three consumer groups. Four line items, and they add up to eight times the ingest rate:

producer in:          1,000,000,000
replication out:      1,000,000,000 x 2                          =  2,000,000,000
replication in:       1,000,000,000 x 2                          =  2,000,000,000
consumer out:         1,000,000,000 x 3                          =  3,000,000,000
cluster bytes/s:      1,000,000,000 + 2,000,000,000 + 2,000,000,000 + 3,000,000,000  =  8,000,000,000

Eight gigabytes a second, cluster-wide. Turning that into a machine count needs one hardware number.

The NIC, and which end of its range this chapter takes. Every machine has a network interface card (NIC), the hardware that connects it to the network. Numbers worth memorizing cold puts a commodity box at 1-10 Gbps and tells you to default to 10 and say so, because silently taking an end of a range is how two chapters reach opposite fleet sizes.

This chapter deliberately takes the bottom of that range — 1 Gbps — and says so here. One gigabit per second is 125 million bytes per second of payload. The pessimistic NIC is the one that makes the binding constraint visible, and finding the binding constraint is the whole point of the derivation. At ch 02’s own default of 10 Gbps the same arithmetic gives 11 brokers rather than 107; that gap is restated after the disk check.

Budget 60% of the NIC for the steady path. The other 40% covers a lagging consumer catching up and the burst of traffic a rebalance causes:

usable per broker:    125,000,000 x 0.60                         =  75,000,000
brokers needed:       8,000,000,000 / 75,000,000                 =  107

Round up to 128 brokers (a power of two makes partition-to-broker assignment even).

Now check disk against that same fleet. The block below works out, in order: per-broker network load and what fraction of the NIC that is; per-broker disk write rate and what fraction of a drive’s sequential bandwidth that is; and per-broker storage. Compare the two fractions:

per-broker network:   8,000,000,000 / 128                        =  62,500,000
fraction of NIC:      62,500,000 / 125,000,000                   =  0.50
log writes/s cluster: 1,000,000,000 x 3                          =  3,000,000,000
per broker:           3,000,000,000 / 128                        =  23,437,500
fraction of 1 GB/s:   23,437,500 / 1,000,000,000                 =  0.0234
storage per broker:   1,814,400,000,000,000 / 128                =  14,175,000,000,000

The NIC is at 50% and the disk is at 2.3% of its sequential bandwidth. The network runs out first, by a wide margin, and that is the finding. It is the whole shape of the system: a message broker is a network device that happens to persist.

Two words in that sentence carry weight. Sequential means the writes land one after another at the end of a file, so the drive never has to jump to a different location. Random means each write goes somewhere unrelated to the last. A modern solid-state drive — NVMe, the interface that lets a solid-state drive talk to the processor over the PCIe bus rather than pretending to be a spinning disk — is roughly ten times faster at sequential than at random, and this design only ever does sequential. That is why the qualifier on the 2.3% is load-bearing: Why a log beats a queue 328 devices or one shows what happens the moment you give it up.

This fleet size is a hardware answer, not a workload answer. A 10 Gbps NIC — ch 02’s stated default, and the number this chapter chose not to take — would drop the fleet from 107 brokers to 11. Same workload, order-of-magnitude different machine count. State the sensitivity, and state which end of the range you took. Everything below is sized on the 128-broker, 1 Gbps fleet.

API sketch

Three method signatures, no bodies; they pin down the contract rather than being run. Two details in the parameter lists decide the rest of the design.

class Log:
    def produce(self, topic: str, key: bytes, value: bytes,
                acks: str = "all") -> tuple[int, int]:
        """Returns (partition, offset). Partition = hash(key) % n unless key is None."""

    def fetch(self, topic: str, partition: int, offset: int,
              max_bytes: int = 1 << 20, max_wait_ms: int = 500) -> list[bytes]:
        """Long poll. Returns when max_bytes are available or max_wait_ms elapses."""

    def commit(self, group: str, topic: str, partition: int, offset: int) -> None:
        """Durable read position. The consumer decides when. This is the whole design."""

fetch is a long poll: when there is nothing to return, the broker holds the request open for up to max_wait_ms instead of replying “nothing yet” straight away. Push vs pull and why pull won prices what that saves.

Two other things are load-bearing.

• fetch takes an offset. That is why replay costs nothing and why a second consumer group costs nothing.
• commit is separate from fetch. That seam is what makes delivery semantics a choice rather than a property (Delivery semantics honestly).

Data model: the log

On disk, a partition is simpler than it sounds — and its record layout produces the constant 1,024 bytes per record used everywhere later.

Segments

A partition is a directory of segment files. The log is not one enormous file: it is a series of chunks, each rolled shut once it reaches a fixed size and never touched again.

Every record carries a fixed 24-byte header before its payload. Add that to the payload to get the on-disk cost of one message, then divide a segment by it:

record header:        offset 8 + length 4 + CRC 4 + timestamp 8  =  24
record on disk:       24 + 1,000                                 =  1,024
segment size (1 GiB):                                               1,073,741,824
records per segment:  1,073,741,824 / 1,024                      =  1,048,576

The CRC in that header is a cyclic redundancy check — a short checksum computed over the record’s bytes, so a reader can tell that what came off the disk is what was written.

A 1 KB message occupies 1,024 bytes, and a 1 GiB segment holds 1,048,576 of them. The 1,024 is used everywhere later in the chapter.

GB and GiB are not the same, and this chapter labels both

Two units are in play, and mixing them silently is how a capacity plan ends up 7% out.

Every estimate in this chapter uses decimal bytes, per ch 02: 1 GB is 1e9, and 1 GB/s is 1e9 bytes a second.

But segment.bytes and batch.size are configuration values, and the software that reads them interprets them in binary units. So a “1 GB segment” is 1 GiB = 1,073,741,824 bytes, and a “16 KB batch” is 16 KiB = 16,384 bytes.

This chapter writes GiB and KiB wherever it means the binary one, and plain GB/KB everywhere else.

The two index files

Alongside each segment sit two small lookup files.

• The offset index, .index, maps an offset to the byte position in the segment where that record starts.
• The time index, .timeindex, maps a timestamp to an offset, which is what makes “start me at 9 a.m.” expressible.

Neither records every message. They hold one sparse entry per 4 KB of log, so they stay small enough to keep in memory.

A lookup is therefore a binary search — repeatedly halving the range you are searching — over that sparse table. It lands you slightly before the record you want, and a short forward scan finishes the job. That is why fetch(offset) costs a logarithmic number of comparisons plus one disk seek, rather than reading the whole partition.

One mutable integer, and nothing else

Nothing is ever updated in place. There is no per-message state, no “read” flag, and no tombstone — the marker a database writes to say “this key is deleted” and later has to clean up.

The only mutable thing in the entire system is one integer per consumer group per partition. Even that lives in an ordinary topic called __consumer_offsets, which is a log like any other.

__consumer_offsets is a compacted topic: the broker periodically discards all but the newest record for each key. So the space a group’s bookmark occupies stays constant no matter how often it commits.

High-level architecture

The diagram below is the whole system in one picture. Read it top to bottom: producers at the top, the partition log in the shaded blue box in the middle, consumer groups at the bottom, and the controller off to the side.

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

What the colours mean

Colours are ch 01’s key, unchanged.

Blue #1d3557 is the authoritative copy of the data. Here that is the partition log itself — every other box either produces bytes for it or reads bytes out of it.

Grey #495057 is the plane that watches everything and serves nothing. That is exactly the controller quorum: it decides leadership and tracks the ISR, and not one message flows through it.

Two boxes are deliberately left uncoloured, and one colour is deliberately absent.

• The partitioner is not a server at all. It is a few lines of arithmetic inside the producer library, so it holds no authoritative state and cannot be blue.
• Nothing here is green. Green means read capacity that answers a read without asking the authoritative copy, and a log has none. The followers exist for failover, not for reads, and consumers are served from the leader’s own page cache.

Following one message through

The producer picks a partition. It hands its record to a partitioner — again, code inside the producer library, not a server. The partitioner hashes the key and takes the remainder modulo the partition count. That arithmetic alone decides which of the 16,384 partitions the record belongs to.

The leader appends. Each partition has one leader, and it is the only machine allowed to append to that partition. The shaded box in the diagram is one partition on one broker: the append lands at the end of the active segment, the segment is rolled shut at 1 GiB, and much later the closed segment reaches the retention edge, where the broker will either delete or compact it.

Two followers copy the leader continuously. Each follower broker asks the leader for records exactly the way a consumer does. That is why replication needs no separate protocol — it is the same fetch call.

Two consumer groups read the same bytes without interfering. Group A is the online fleet of 12,000 consumers, committing offsets to __consumer_offsets. Group B is an analytics job with completely independent offsets, so it can be days behind without group A noticing.

Off to the side is the controller quorum. This is a small set of machines that agree among themselves — using a consensus protocol, so a minority can fail without losing the answer — about which broker leads which partition and which replicas are currently caught up. It handles metadata and leadership only, including the in-sync replica set (ISR) that Replication isr and the acks knob derives. Not one message flows through it.

2. Deep dive

The five numbers from the opening table are derived here in order, one mechanism at a time; a sixth subsection settles the design question those five make answerable.

If you read only one, read Consumer lag backpressure and the rebalance storm — it is where these systems actually fail.

1. Why a log beats a queue: 328 devices, or one

The append-only shape is not a stylistic preference but a three-hundred-fold hardware difference, and the way to see it is to cost out the design most teams write first.

The standing disk constants are ~100 MB/s random and ~1 GB/s sequential on NVMe. A 10x bandwidth gap does not sound like an architecture decision. It becomes one when you count how many random operations a queue-shaped data structure actually performs per message.

Costing the table-as-a-queue

Take the design every team writes first: a database table with a status column. INSERT to enqueue a job, SELECT ... FOR UPDATE to claim one exclusively, and UPDATE/DELETE to mark it finished.

One fact makes this expensive. Databases read and write in fixed-size pages — 4 KB here — never in single rows. A one-byte change to a row costs a whole page.

The list below counts the pages touched to move one message through that table. Each line is one 4 KB read or write at a random location on the drive:

1  heap page write   -  INSERT the row
2  primary key leaf  -  INSERT
3  status index leaf -  INSERT (the index the claim query uses)
4  heap page read    -  claim: SELECT ... FOR UPDATE
5  heap page write   -  claim: status -> in_flight
6  heap page write   -  ack:   status -> done
7  status index leaf -  ack:   the row moves within the index
8  heap page rewrite -  vacuum / compaction reclaims the dead tuple

Three of those lines deserve a translation.

• The heap page is where the row’s actual bytes live.
• The index leaf pages are the bottom level of the sorted structures the database keeps so it can find a row by primary key or by status without scanning the whole table. Both must be edited whenever the row they point at changes.
• Vacuum is the background job that reclaims the space left behind by rows that were updated or deleted. A database update does not overwrite in place — it writes a new version and abandons the old one.

So: eight random 4 KB page reads or writes to move one 1 KB message.

The write-ahead log is sequential in both designs, so it cancels out of the comparison. The database’s log is not the problem; everything the database does after the log is.

The comparison, in devices

Top half: multiply the eight pages out to bytes, divide the random bandwidth by that to get messages per second per device, then divide the target by that to get a device count. Bottom half: the same three steps for a log, whose per-message cost is the 1,024 bytes derived earlier.

random bytes/message:  8 x 4,096                                 =  32,768
messages/s per device: 100,000,000 / 32,768                      =  3,052
devices for 1 M/s:     1,000,000 / 3,052                         =  328

sequential bytes/msg:  1,000 + 24                                =  1,024
messages/s per device: 1,000,000,000 / 1,024                      =  976,563
devices for 1 M/s:     1,000,000 / 976,563                        =  1.02
ratio:                 976,563 / 3,052                            =  320

320x, or in the units that matter: one NVMe device instead of 328.

Where does 320 come from? The 10x bandwidth gap between random and sequential is only the floor. The other 32x is write amplification: 1 KB of payload landing in a 4 KB page, times two indexes, times the read-modify-write of the claim and the ack.

The block below is that arithmetic as runnable code, so you can change the assumptions and watch the ratio move. The two assert lines at the bottom pin the 3,052 and the 320 quoted above.

PAGE = 4_096
RANDOM_BPS = 100_000_000          # ~100 MB/s random on NVMe (standing constant)
SEQ_BPS = 1_000_000_000           # ~1 GB/s sequential on NVMe (standing constant)


def table_rate(pages_per_message: int = 8, page: int = PAGE) -> float:
    """Messages/s one device sustains when each message costs random pages."""
    return RANDOM_BPS / (pages_per_message * page)


def log_rate(bytes_per_message: int = 1_024) -> float:
    """Messages/s one device sustains appending to a log."""
    return SEQ_BPS / bytes_per_message


assert round(table_rate()) == 3052
assert log_rate() == 976_562.5          # round() is half-to-even: it gives 976562,
                                        # not the 976,563 the prose rounds to
assert round(log_rate() / table_rate()) == 320

Three consequences of the append-only shape

Not one of these is a feature anybody had to build. The first two are free wins. The third is the one that shows up in incident reviews.

Consequence 1: replay is free. The consumer supplies the offset. Reprocessing yesterday is commit(group, p, offset_at_midnight) and a restart.

In the table design the rows were deleted. Reprocessing requires that you kept a second copy somewhere, which means you built a log after all, badly.

Consequence 2: a second consumer group is free. Its entire cost is one integer per partition. Compare that against what a delete-on-read queue would have to store to feed a second reader — a whole extra day of the stream:

second group, log:    16,384 x 8                                 =  131,072
second group, queue:  1,000,000,000 x 86,400                     =  86,400,000,000,000

131 KB versus 86.4 TB per day. A queue that deletes on read must be written once per consumer; a log is written once and read many times.

This is why the same Kafka cluster feeds the online service, the search indexer, and the warehouse loader. And it is why the alternative shape — the classic broker-tracks-every-message design of AMQP, the Advanced Message Queuing Protocol that RabbitMQ implements — ends up storing three copies of the same stream.

Consequence 3: reads come out of RAM, until one consumer ruins it for everybody.

The operating system keeps recently touched file data in spare RAM. That mechanism is the page cache, and it means a read of data that was written a moment ago never reaches the drive at all.

Consumers on a log read the same bytes the producers just wrote, in the same order. So almost every read is such a read. How much of the log fits? Divide the RAM a broker has spare by the rate that broker writes log bytes:

page cache per broker: 100,000,000,000
log write rate/broker: 23,437,500
seconds of log in RAM: 100,000,000,000 / 23,437,500              =  4,267
hours:                 4,267 / 3,600                             =  1.19

The page cache holds about 1.19 hours of this log.

A consumer reading inside that window — at the tail of the log — is served straight out of memory by sendfile, a system call that copies bytes from a file to a network socket inside the kernel. The data never passes through the broker process’s own memory at all.

A consumer outside that window reads from disk. On its own that is fine: 1 GB/s sequential. The problem is what it does to everyone else. The cache is finite, so the old data it pulls in has to displace something, and what it displaces is the pages every other consumer was using.

One lagging consumer therefore converts the whole broker’s read path from memory to disk. That is the mechanism behind “everything got slow when the backfill job started.” It is also why Consumer lag backpressure and the rebalance storm treats consumer lag — the number of messages sitting between the newest record in a partition and the last one a group committed — as the primary health metric rather than a nice-to-have dashboard.

2. Partitions: the unit of parallelism AND of ordering

How many partitions to create is a decision you get to make exactly once, because it cannot be undone — so it is worth deriving rather than guessing.

A partition is one log, on one leader, with one append point. That single sentence sets both the parallelism — how much work can happen at the same time — and the ordering, and the two are the same knob pointed in opposite directions.

Ordering exists inside a partition and nowhere else

Offsets are assigned by the leader on append, so within a partition ordering is total: for any two records you can say which came first.

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. Even a perfect clock would not help, because the consumers read the two partitions independently.

Pricing global ordering

If you genuinely need every record in a topic ordered against every other, you need exactly one partition. One partition lives on one broker, and that broker’s NIC has to carry four copies of the stream: the ingest in, the replication out to two followers, and one consumer group’s read out.

Call the record rate r bytes per second. Then 4r must fit in the NIC:

in r + out 2r (two followers) + out r (one group)  ->  4r <= 125,000,000
max bytes/s:          125,000,000 / 4                            =  31,250,000
max messages/s:       31,250,000 / 1,000                         =  31,250
share of the target:  31,250 / 1,000,000                         =  0.031

3.1% of the requirement — and that is the optimistic bound.

The real ceiling is on the other side. One partition also means one consumer, because a partition is assigned to at most one member of a group by construction; two members reading it would lose the ordering again. So the group’s whole throughput is one process’s throughput. At 10 ms of processing per message:

one consumer:         1 / 0.010                                  =  100
consumer vs broker:   31,250 / 100                               =  313      (capacity / capacity)

Global ordering caps the system at 100 messages per second, 313x below what the single broker could even serve.

The fix is to order per key, not globally. partition = hash(key) % P puts every event for account 42 in one partition, in order, forever. No realistic business requirement needs account 42 and account 99 ordered against each other.

Sizing the partition count

Partition count is not set by throughput. It is set by how slow the consumer is.

Two rates meet here and they are different quantities, so name them before dividing:

State whether a rate is offered load or service capacity when you derive it, and never divide one by the other without tracking which is which.

1,000,000 msg/s is offered load — how much work arrives. 1 / 0.010 = 100 msg/s is one consumer’s service capacity — how much work it can finish.

Divide them and you get 10,000 consumers: a group whose capacity exactly equals its demand. That is utilization rho = 1. Utilization is offered load divided by capacity, and at rho = 1 there is zero headroom — no spare capacity at all.

Zero headroom is not “efficient”, it is broken. Any backlog that ever forms is worked off at a rate of zero and stays forever. Consumer lag backpressure and the rebalance storm shows that as an infinite drain time.

So size the group against the same 20% headroom that Consumer lag backpressure and the rebalance storm’s drain arithmetic assumes — provision for 1.2x the offered load, then divide by what one consumer can do:

offered load:                                                       1,000,000
consumer capacity:    1 / 0.010                                  =  100
capacity to provision at 20% headroom:  1,000,000 x 1.2          =  1,200,000
consumers needed:     1,200,000 / 100                            =  12,000

12,000 consumers, not 10,000. The extra 2,000 are not waste. They are the difference between a 30 s stall that drains in 150 s and one that never drains.

A consumer with no partition assigned to it does nothing, so the partition count must be at least 12,000 or the twelve-thousand-and-first consumer sits idle. Round up to 16,384 (a power of two, and comfortable room above 12,000).

The five lines below are what 16,384 partitions cost per broker across the 128-broker fleet. Note the last one: three files per partition replica — the segment, the .index, and the .timeindex:

per-partition rate:   1,000,000 / 16,384                         =  61
replicas cluster-wide: 16,384 x 3                                =  49,152
replicas per broker:  49,152 / 128                               =  384
leaders per broker:   16,384 / 128                               =  128
open file handles:    384 x 3                                    =  1,152

The maximum useful consumer count is the partition count. The 16,385th consumer on a 16,384-partition topic adds zero throughput and one idle process. This matters because “just scale out the consumers” is the reflex answer, and it stops working exactly at this line.

The bill for many partitions arrives at the producer

Assume the million messages a second come from a fleet of 1,000 producer processes, so each one produces 1,000 msg/s.

Producers batch by partition: they accumulate records destined for the same partition and send them in one request. That only works if a producer sends to a given partition often enough for a batch to fill. With 16,384 partitions it does not:

per-producer rate:    1,000,000 / 1,000                          =  1,000
per partition:        1,000 / 16,384                             =  0.061

Each producer touches any given partition 0.061 times a second — about once every 16 seconds. No batch ever fills, so every message becomes its own request.

The fix is the sticky partitioner. For keyless records — records sent with no key, where any partition is as good as another — the producer sticks to one partition until its batch is full, sends it, and only then moves to the next.

Batching returns. Divide the batch size by the per-record size to get records per batch, then work out the request rate with and without it.

Watch the units on the first line — 16,384 appears here as bytes, not as a partition count. A 16 KiB batch is 16,384 bytes, and each record is 1,024 bytes. The two 16,384s in this section are unrelated.

messages per 16 KiB batch: 16,384 / 1,024                        =  16
requests/s per producer:   1,000 / 16                            =  63
cluster request rate:      1,000 x 63                            =  63,000
unbatched:                 1,000 x 1,000                         =  1,000,000

16x fewer broker requests, from a partition-assignment policy. Keyed records cannot use the sticky partitioner, because their partition is determined by the key. That is one more reason to keep the key space wide enough that batches still fill.

The remaining ceilings, and the one-way door

The other partition-count limits are operational:

• 49,152 replicas for the controller to track as metadata;
• 1,152 open file descriptors per broker — a file descriptor is the operating system’s handle on an open file, and the per-process limit is a real ceiling;
• a controller failover that must elect a new leader for every partition the failed controller owned.

Partition count is easy to raise and impossible to lower. Lowering it would change P in hash(key) % P, which re-routes every existing key to a different partition and destroys per-key ordering. So pick it once, with the consumer-latency arithmetic above, rather than doubling it in an incident.

3. Delivery semantics, honestly

Delivery guarantees are where marketing and mechanism drift furthest apart. The honest choice produces duplicates that can be priced — and when the destination is a database, one construction gets you the effect of exactly-once.

Three names, two operations

The three semantics are not three implementations. They are three orderings of the same two operations a consumer performs:

1. Apply the effect — do the real work, such as charging a card or writing a row.
2. Commit the offset — record that this message is done.

The table below puts the two operations in each possible order and asks what happens if the process dies in between. That crash is the whole story; nothing else distinguishes the three:

Pricing at-least-once

At-least-once is the only defensible default, so price its duplicates.

When a consumer dies, it re-processes everything between its last commit and where it had actually got to. That amount is set by what the consumer actually consumed, which is its share of the offered load — not its capacity. A consumer running at 20% headroom does 83 msg/s, not the 100 it could do. Use the wrong one and the answer is 20% too big.

How often does a consumer die? Deploys, OOM kills (the operating system terminating a process that asked for more memory than the machine has), and rebalances. Assume roughly one restart per instance per day.

The block below computes the per-consumer figures first, then re-derives the daily total without passing through them, so no rounding error compounds:

per-consumer throughput (offered load, not capacity):
                      1,000,000 / 12,000                         =  83.3   (rounded)
replay per restart:   83.3 x 5                                   =  416.7  (rounded)
restarts/day:         12,000 x 1                                 =  12,000

replayed/day, without routing through either rounded figure:
  produce_rate x commit_interval x restarts_per_consumer
                      1,000,000 x 5 x 1                          =  5,000,000
duplicates/s:         5,000,000 / 86,400                         =  57.9
processed/day:        1,000,000 x 86,400                         =  86,400,000,000
duplicate rate:       5,000,000 / 86,400,000,000                 =  0.0000579

Notice what the group size did to that answer: nothing. Replay per day is produce_rate x commit_interval x restarts_per_consumer, and the consumer count cancels out. That cancellation is the check telling you the derivation is on the offered-load side rather than the capacity side.

Fifty-eight duplicate messages every second, forever. That is what “at-least-once” means at this scale, and it is why “we’ll handle duplicates if they happen” is not a plan.

Note that the last two lines both divide by 86,400 and not by 1e5. A duplicate rate is a result, and the 1e5 version of this arithmetic reports 50/s — 13.6% flattering (ch 02). Understating your own worst number is the wrong direction to be wrong in.

Turning the commit-interval dial

Shorten the commit interval from 5 s to 100 ms and the duplicates shrink by 50x. The cost is coordinator load, which grows as 1 / interval:

replayed/day:         1,000,000 x 0.1 x 1                        =  100,000
duplicates/s:         100,000 / 86,400                           =  1.2
offset commits/s:     12,000 / 0.1                               =  120,000

1.2 duplicates per second, bought with 120,000 offset commits/s into __consumer_offsets. That topic is a 50-partition compacted log, so it works out at 2,400 writes/s per partition. Nameable, survivable, and a genuine trade rather than a free win.

Why exactly-once is not on the menu

End-to-end exactly-once requires that the effect (a row in MySQL, a charge at Stripe, an email) and the offset commit either both happen or neither does. They live in different systems.

Making two independent systems agree on a single commit point needs one of two things, and neither is available here.

Option one: a transaction spanning both — two-phase commit. A coordinator first asks every participant to prepare, meaning promise it can finish; then it tells them all to commit. This needs prepare/commit support on each side, and it blocks every participant indefinitely if the coordinator dies between the two phases.

Option two: an infinite exchange of acknowledgements. This is the Two Generals result: two parties communicating over a link that can drop messages can never both become certain they agree, because the last acknowledgement always needs an acknowledgement of its own.

Deep dive 3 exactly once is not available works this through for a third-party provider that offers no idempotency key at all. The conclusion is the same here and does not need re-deriving.

What Kafka’s transactions actually guarantee

The guarantee is narrower than the marketing, and stating the boundary precisely is what the question is testing. Two separate mechanisms are involved, and they solve different problems.

1. The idempotent producer. Idempotent means an operation can be applied more than once without changing the result beyond the first application.

Each producer is issued a (producer id, epoch). The epoch is a generation counter that increases every time that producer identity is re-established, so an old incarnation can be recognised and rejected. The producer then stamps a monotonically increasing sequence number on each record it sends to each partition, and the leader remembers the last 5 sequence numbers per producer per partition and silently drops anything it has already seen.

This removes duplicates caused by producer retries: a produce call that timed out but had actually succeeded no longer lands twice. It does nothing about duplicates caused by anything else.

2. Transactions. A transaction coordinator writes begin and commit markers into the data partitions and into __consumer_offsets as one atomic step.

A consumer configured read_committed — meaning it will only surface records belonging to transactions that have committed — refuses to read past the Last Stable Offset (LSO), which is the offset of the earliest transaction still open. It also skips records belonging to transactions that were aborted.

So the atomic unit is consume-from-Kafka, produce-to-Kafka, commit-offset. All three endpoints must be inside the log. The instant one endpoint is your database, the marker mechanism has nothing to write there and the guarantee evaporates. Kafka’s exactly-once is exactly-once stream processing, not exactly-once delivery.

It is also not free. A read_committed consumer stalls at the LSO until the open transaction commits, so on average it waits half a commit interval:

transaction commit interval:  100
mean added latency:           100 / 2                            =  50

50 ms of mean added end-to-end latency, worst case 100 ms, in exchange for the marker protocol. The marker writes themselves are two datacenter round trips, 2 x 500 us = 1 ms (ch 02) — noise next to the commit interval.

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

The sink is the destination of the message. When it is a database, the answer is an idempotent consumer.

The construction is one sentence: in the same database transaction that performs the effect, write a dedup key — a unique identifier for the message, on a column with a uniqueness constraint, so a second insert of the same key fails.

Because the two writes commit or roll back together, a redelivered message finds its own key already present and does nothing.

The whole construction turns on what the key is made of, and the tempting answer is wrong.

topic:partition:offset is a log position, not a message identity. Replication isr and the acks knob is about to show the one event that separates the two: after an unclean leader election the log gets shorter, offsets go backwards, and the idempotent producer re-sends. The same logical payment then arrives at a different offset, presents a key nobody has seen, and is applied a second time.

Key on something the producer minted and re-sends unchanged — an idempotency key carried in the record. That identity survives every renumbering the log can do to it.

The code below is that consumer, in about ten lines of SQLite. Read handle first — the two rowcount checks are the load-bearing parts — then the three test cases underneath: a normal apply, a redelivery at a different offset, and an effect that matched no row.

import sqlite3


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


def handle(conn, record) -> str:
    """At-least-once delivery + a dedup key in the same transaction as the
    effect == effectively-once, with no coordinator between the two systems.

    Two things are load-bearing and both are one line each: the key is the
    PRODUCER's identity for the message rather than the offset it happens to
    sit at, and the effect's rowcount is checked so that an update matching
    nothing rolls back instead of committing a key for work never done.
    """
    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, the idempotent producer re-sent, and the
# same logical message now sits at offset 3. Keyed on the offset this applies
# twice and the balance is 500; keyed on the producer's own 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. Committing one here credits
# nobody, logs nothing, 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

How long the dedup table must remember

The answer is not “forever.” It is the maximum redelivery horizon: the session timeout — the interval a consumer may go without sending a heartbeat before the group coordinator declares it dead — plus the time the resulting rebalance takes.

Past that point the partition has been handed to somebody else, and no consumer will be offered that message again by the system. Remembering it is pure cost.

Take a 45 s session timeout and 15 s of rebalance. Multiply the horizon by the message rate to get how many keys are live, then by 16 bytes per key:

redelivery horizon:   45 + 15                                    =  60
keys in the window:   1,000,000 x 60                             =  60,000,000
bytes at 16 B/key:    60,000,000 x 16                            =  960,000,000

960 MB — one Redis instance (an in-memory key-value store) or an embedded RocksDB store (a key-value library that runs inside your own process).

Set the horizon to 24 hours “to be safe” and the same table is 1,000,000 x 86,400 x 16 = 1.38e12 bytes: 1.38 TB, and a different project. The dedup window is a real design decision with a two-order-of-magnitude cost swing, and it is derived from the consumer group’s timeout configuration, not chosen by feel.

One qualifier, and it is not pedantry: that horizon covers system-generated redeliveries only — a retry, a rebalance, a restart, a crash between apply and commit.

A deliberate replay is out of scope by construction. commit(group, p, offset_at_midnight) re-offers messages that are hours old, and it exists precisely because you want those effects applied again.

If you want a replay to be idempotent as well, that is a different and much more expensive design. It needs the same producer-supplied key, and it needs the table to remember for the full retention window rather than for a minute: 1,000,000 x 86,400 x 7 x 16 = 9.7 TB, which is the table nobody builds.

Decide which of the two you are building. A 60-second table described as making replay safe is a guard that is not there.

4. Replication, ISR, and the acks knob

Copies of a partition are kept in step by one mechanism, the producer gets one durability knob, and a single configuration flag decides whether you lose data or lose availability when everything goes wrong at once.

ISR and the high watermark

Each partition has a leader and RF - 1 followers.

A follower that has fetched everything the leader held as of replica.lag.time.max.ms ago is in-sync. The set of replicas currently in that state is the in-sync replica set (ISR). 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 known to exist on every member of the ISR. Consumers are not allowed to read past it.

That one rule is what makes failover — the promotion of a follower when the leader dies — safe. Everything a consumer has ever been shown already exists on every in-sync replica, so no promotion can make a record a consumer already read disappear.

The acks knob

The producer’s acks setting selects how many replicas the producer waits for before it treats its send as successful.

To price the middle setting, work out how much data sits on the leader but not yet on a follower. Per-broker ingest times the follower’s fetch lag:

per-broker ingest:    1,000,000 / 128                            =  7,813
follower fetch lag under load, seconds:                             0.010
unreplicated at acks=1: 7,813 x 0.010                            =  78

Now the four settings side by side. Read the third column first — it is what you actually lose when a broker dies:

The latency column is ch 02’s 500 us datacenter round trip: acks=1 is the leader’s response; acks=all adds one follower fetch round trip on top.

acks=1:               500
acks=all, min.isr=2:  500 + 500                                  =  1,000
acks=all, min.isr=3:  500 + 500                                  =  1,000

The last two rows are the same number on purpose.

The two followers fetch from the leader concurrently, not one after the other. Requiring the second one to have the record does not buy another round trip — it buys the slower of two draws from the same distribution, which moves the tail and not the mean.

So min.insync.replicas=3 costs availability, not latency. Here is why that is worse rather than better.

min.insync.replicas=3 with RF=3 is the same mistake as setting W = N in a quorum store — requiring the write to reach all N copies rather than a majority. The quorum what w and r actually buy already priced it: at 99.9% availability per node it is a thousand times less available than waiting for two, and buys nothing that waiting for two does not. Cite it, do not re-derive it.

Here it has a second failure mode the key-value version does not. If one replica falls out of the ISR, the ISR is size 2, min.insync.replicas=3 cannot be met, and every producer to that partition gets NotEnoughReplicas until the follower catches up. Durability config turned into an availability outage.

Why acks=0 loses more than you would guess

acks=0 deserves its own line, because the loss is not 78 messages. It is the entire producer buffer, and that buffer is bigger than people think.

The default buffer.memory is 32 MiB — 33,554,432 bytes. Divide it by the message size to get messages held, then by one producer’s send rate to get how much time that represents:

buffer.memory:        33,554,432
messages buffered:    33,554,432 / 1,000                         =  33,554
seconds of buffer at 1,000 msg/s: 33,554 / 1,000                 =  33.6

33,554 messages, which is 33.6 seconds of that producer’s output, held in memory with nothing durable behind it.

Unclean leader election

This is the failure worth being able to describe precisely.

An election is clean when the new leader comes from the ISR, and therefore holds everything consumers have seen.

It is unclean when no ISR member is available and the system promotes a replica that had already fallen out of the set. That happens when every ISR member for a partition is down and unclean.leader.election.enable=true.

How much is lost? By definition the promoted replica is more than replica.lag.time.max.ms behind, so multiply that window by the per-partition rate, then by the number of partitions the dead broker led:

replica.lag.time.max.ms, seconds:                                   30
per-partition rate:                                                 61
minimum messages behind: 61 x 30                                 =  1,830
partitions on that broker:                                          128
messages truncated:   128 x 1,830                                =  234,240

234,240 messages vanish, and the consumer never learns.

Worse than the loss is the shape of it: the log got shorter.

Walk through what that does to a reader. A consumer had committed offset 5,000. The new leader’s log ends at 3,170. The consumer’s bookmark now points past the end of the file, so the broker moves it to whatever auto.offset.reset says — the setting that decides where a consumer starts when its offset is invalid, usually the oldest or the newest available record. The 1,830 records in between are never delivered to anyone. No error, no exception, no metric that goes red.

Offsets going backwards is the one thing the whole design assumes cannot happen. It is exactly why the idempotent consumer in Delivery semantics honestly keys on a producer-supplied identifier rather than on topic:partition:offset: this event renumbers the offset and hands the same logical message back under a key nobody has seen.

The alternative setting is unclean.leader.election.enable=false, which leaves the partition unavailable until an ISR member returns.

That boolean is not a tuning parameter. It is the sentence “when I must choose, I choose availability / I choose durability,” written in a config file. For payments it is false. For clickstream it is true. Saying which one your topic is, and why, is the answer.

The sequence diagram below is one acks=all write, from the producer’s call to the acknowledgement. Watch two things: the ack does not come back until the in-sync follower has fetched past the record, and the lagging follower F2 takes no part in it — which is what makes the final line’s truncation possible.

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

5. Consumer lag, backpressure, and the rebalance storm

This is where these systems actually fail: one metric worth alerting on, a recovery time that depends on spare capacity rather than raw speed, and a consumer group that can destroy itself with a feedback loop that looks, from the outside, like the brokers being slow.

Lag, and why throughput is the wrong metric

Consumer lag is the only metric in this system that means anything, because it is the only one whose rate of change tells you something a snapshot cannot.

Throughput — messages per second flowing through — looks perfect right up to the moment the backlog is growing without bound. A saturated system still moves messages at full speed; it simply never catches up. (Ch 01 makes the same point about job queues.)

Three definitions, in increasing order of usefulness:

lag = log_end_offset - committed_offset          per partition, in messages
lag_seconds = lag / consume_rate                 the number a human can act on
d(lag)/dt = produce_rate - consume_rate          the number that predicts

The first is raw lag in messages: how many records sit between the newest one written and the last one this group committed. The second converts it to seconds, which is the form a human can reason about. The third, d(lag)/dt, is the rate at which lag is changing, and it is the one to alert on.

The sign of d(lag)/dt is the whole diagnosis. Positive, and the system is already broken; no amount of waiting fixes it. Negative, and the backlog has a finite drain time.

Drain time is set by headroom, not by throughput

That is the counterintuitive part, so derive it. A 30-second stall builds a backlog of 30 seconds’ worth of production. The rate that backlog shrinks is not the group’s capacity — it is capacity minus the load still arriving:

backlog from a 30 s stall: 1,000,000 x 30                        =  30,000,000
capacity at 20% headroom:  1,000,000 x 1.2                       =  1,200,000
drain rate:                1,200,000 - 1,000,000                 =  200,000
drain time:                30,000,000 / 200,000                  =  150

capacity at 5% headroom:   1,000,000 x 1.05                      =  1,050,000
drain rate:                1,050,000 - 1,000,000                 =  50,000
drain time:                30,000,000 / 50,000                   =  600

Cutting headroom by 4x multiplies recovery time by 4x, on identical hardware and identical throughput.

Three notes on where those numbers come from.

The 20% row is not a hypothetical. It is the 12,000-consumer group Partitions the unit of parallelism and of ordering provisions, whose 1,200,000 msg/s of capacity is exactly the numerator here. Every subtraction on this page is capacity minus offered load, and the group is sized so that the difference is not zero.

The 30 s stall is not hypothetical either. The rebalance storm derived at the end of this section produces exactly that.

And 600 s of lag, at 23,437,500 bytes/s per broker, is well inside the 1.19-hour page cache window from Why a log beats a queue 328 devices or one — so this recovery runs at memory speed. A backlog that exceeds 1.19 hours crosses into disk reads and evicts everyone else’s cache. That is where lag stops being linear.

The function below is the same formula, with the headroom = 0 case made explicit: no surplus means infinite drain time.

def drain_seconds(backlog: float, produce_rate: float, headroom: float) -> float:
    """Time to work off `backlog` when capacity is (1 + headroom) x produce."""
    surplus = produce_rate * headroom
    if surplus <= 0:
        return float("inf")
    return backlog / surplus


assert round(drain_seconds(30_000_000, 1_000_000, 0.20)) == 150
assert round(drain_seconds(30_000_000, 1_000_000, 0.05)) == 600
assert drain_seconds(1, 1_000_000, 0.0) == float("inf")

Where backpressure actually lives

Backpressure does not exist on the broker side, and pretending otherwise is a common wrong answer.

Backpressure is the mechanism by which an overloaded consumer of work makes its producer slow down. 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 own bounded buffer, and it degrades in three stages:

1. send() hands back a placeholder for the eventual result and returns immediately — while there is room in the buffer named by buffer.memory.
2. Buffer full: send() blocks, for up to max.block.ms = 60,000 ms.
3. Still full when that expires: send() throws an exception into the calling application.

So the 33.6 seconds derived in Replication isr and the acks knob from 33,554,432 / 1,000 / 1,000 is a hard operational service-level objective (SLO — the number the system promises to stay inside). You have 33 seconds of broker unavailability before application threads start seeing exceptions. That number, not “five nines,” is what your broker recovery runbook has to beat.

The rebalance storm

When a group’s membership changes, the group coordinator — the broker that tracks who is in the group and who owns which partition — revokes every assignment and hands out new ones. Under the original, eager protocol, every member stops consuming for the whole round, however small the change.

A rebalance storm is what happens when that stall causes the next rebalance. It has a precise trigger condition, and staying safe means keeping this inequality true:

max.poll.records x seconds_per_record < max.poll.interval.ms / 1,000

Two terms first. poll() is the call a consumer makes to fetch its next batch, and max.poll.records caps how many records that batch may hold.

A consumer that returns from poll() with a batch and then takes longer than max.poll.interval.ms to come back for the next one is presumed dead and ejected from the group. The result is a self-reinforcing loop: ejection triggers a rebalance, which stalls everyone, which grows the backlog, which makes the next batch bigger, which takes longer to process, 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

Substitute the defaults — 500 records at 10 ms each — against the 300-second limit:

default:              500 x 0.010                                =  5
limit, seconds:       300
safe ceiling:         300 / 0.010                                =  30,000

Defaults are 60x inside the limit. The storm happens when someone raises max.poll.records to 50,000 to “drain the backlog faster” and crosses 30,000.

The cure is the opposite of the instinct: lower the batch size. Or move the actual processing onto a separate thread and use the client’s pause and resume calls, which let poll() keep being called — so the group keeps hearing from the consumer — without delivering more work than the consumer can absorb.

Cooperative rebalancing and static membership

The structural fix is cooperative (incremental) rebalancing, which revokes only the partitions that actually move. Compare what each protocol disturbs when one consumer joins a 12,000-member group:

eager, partitions revoked:    16,384
cooperative, partitions moved: 16,384 / 12,001                   =  1.365  (rounded)
disruption ratio:             16,384 / (16,384 / 12,001)         =  12,001

The ratio is exactly the group size plus one, and that is not a coincidence. Eager revokes every partition; cooperative moves one consumer’s share. So the ratio is 16,384 / (16,384/(N+1)) = N+1, and the partition count cancels out entirely.

Do not compute it from the rounded middle line. 16,384 / 1.365 gives 12,003 and 16,384 / 1.4 gives 11,703 — both are significant figures manufactured out of two.

~12,000x less disruption to add one consumer, straight off the algebra as N + 1 with N = 12,000, and not one digit of it taken from the rounded 1.365.

Then add static membership: give each consumer a stable identity in group.instance.id that survives a restart, instead of a fresh one each time it connects. A consumer that comes back inside its session timeout rejoins with the same assignment and triggers no rebalance at all. That converts a rolling deploy of 12,000 consumers from 12,000 rebalances into zero.

6. Push vs pull, and why pull won

The consumer asks for data instead of the broker sending it, and the one real cost of that choice disappears with a single configuration change.

Three arguments for pull

Push means the broker decides when and how much each consumer receives. Pull means the consumer asks. Three arguments settle it, and only the third is about performance.

Backpressure has to live where the capacity is known. A pushing broker must either buffer for a slow consumer — unbounded broker memory, the failure being on the wrong machine — or drop. A pulling consumer that is slow simply does not ask, and the resulting backlog is on disk, bounded by retention, and measurable as lag. Push converts a consumer problem into a broker problem; pull leaves it where someone can fix it.

Batching has to be decided by the party that knows its own capacity. A consumer requests up to fetch.max.bytes; a pushing broker guesses, and a wrong guess is either tiny records (no batching) or an overrun (the drop above).

Replay is only expressible in a pull model. fetch(offset) with any offset is the entire replay feature. There is no push equivalent of “send me last Tuesday again” that does not amount to the consumer telling the broker a position — at which point it is a pull.

The one real cost of pull, and how long poll removes it

The cost is idle polling: a consumer with nothing to do still asks repeatedly, and every empty answer costs a round trip on both machines.

The fix is long poll. Instead of answering “nothing yet” immediately, the broker holds the request open for up to fetch.max.wait.ms and replies the moment data arrives.

Tuning the polling interval cannot compete with that, because the interval trades request rate against latency — shorten it and you pay in requests, lengthen it and you pay in latency. Long poll improves both at once. Compare a naive 100 ms poll (10 requests/s per connection) against a 500 ms long poll (2 requests/s per connection), across 12,000 consumers each connected to 2 brokers:

naive poll at 100 ms, 2 brokers per consumer:
requests/s:           12,000 x 2 x 10                            =  240,000
mean added latency:   100 / 2                                    =  50

long poll, fetch.max.wait.ms = 500, fetch.min.bytes = 1:
requests/s at idle:   12,000 x 2 x 2                             =  48,000
ratio:                240,000 / 48,000                           =  5

Long poll is 5x cheaper in requests and lower in latency at the same time. The latency win comes from the broker parking the fetch and answering the instant a byte arrives: the added delay drops from the 50 ms a message waits on average for the next scheduled poll to one datacenter round trip, 500 microseconds.

Improving two things that normally trade against each other usually means you found a real design. Here it works because the request rate stops depending on the poll interval and starts depending on the data rate.

3. Bottlenecks and scaling

Every resource in the design runs out at some scale. The table below says where each one binds and what you do about it.

Two terms in the table need naming up front. KRaft is Kafka’s built-in consensus implementation for cluster metadata; it replaced ZooKeeper, a separate coordination service that had to be operated alongside the brokers. RTT is round-trip time: how long a packet takes to reach another machine and come back.

Read the middle column for the number that says when each one bites:

Growth is not symmetric. Adding brokers is easy. Getting data onto them is not: a new broker starts empty, and existing partitions do not move to it on their own. You must move replicas explicitly.

Price that move. Each broker holds 14.18 TB (derived in Back of envelope), and half its NIC — 62,500,000 bytes/s — is spare:

seconds to move one broker's data:
                      14,175,000,000,000 / 62,500,000            =  226,800
days:                 226,800 / 86,400                           =  2.6

About 2.6 days to refill one broker. Plan capacity in units of “days to rebalance,” not “hours to provision.”

4. Failure modes

The failures below are the ones that actually happen in production, each with what it looks like from the outside.

Three names in it are worth having up front:

• a poison message is a single record that makes its consumer fail every time it is processed, so retrying forever blocks the partition behind it;
• a dead-letter topic is the separate topic you divert such a record to after a bounded number of retries, so the healthy traffic can continue;
• NTP is the Network Time Protocol, the service that keeps machine clocks roughly agreed.

The middle column is what you would actually see on a dashboard — that is the column to memorise, because it is how you recognise each failure in the room:

The one to volunteer unprompted is the retention row. Every other failure here is loud. “The consumer was down for eight days and the data aged out” is silent, and the alert that catches it is on lag measured in seconds of retention consumed, not messages.

5. Alternatives rejected

Each design below was considered and priced, because a rejection with a number attached is worth more than a preference.

Three names appear in it:

• SQS is Amazon’s Simple Queue Service, a managed queue you rent rather than run;
• Pulsar is a competing log system that separates the serving brokers from the storage layer;
• Raft is a consensus algorithm — a protocol that lets a group of machines agree on an ordered sequence of decisions even while some of them fail.

The third column is the point of the table. Every rejection carries a number:

The managed option, priced honestly

Be honest about the managed option, because the honest version is more persuasive than the reflex one.

Both sides of this comparison must use the same seconds. The broker side is priced in exact hours — 24 x 365 — so the managed side has to divide the day by 86,400 too. Run it through 1e5 and the managed bill comes out 16% too high, which is the difference between “managed loses” and “it is a wash” — decided by a rounding convention rather than by anything real.

The block below works from messages per day to a yearly bill on each side. Assume a batch of 10 messages per API request and $0.40 per million requests, against $1.00 per broker-hour:

messages/day:         1,000,000 x 86,400                         =  86,400,000,000
requests at batch 10: 86,400,000,000 / 10                        =  8,640,000,000
cost/day at $0.40/M:  8,640,000,000 / 1,000,000 x 0.40           =  3,456
cost/year:            3,456 x 365                                =  1,261,440
128 brokers at $1.00/h: 128 x 24 x 365                           =  1,121,280
ratio:                1,261,440 / 1,121,280                      =  1.13

$1.26 M/year managed against $1.12 M/year of raw instances — 13% apart, which at this precision is a wash. And the instance figure does not include the engineers who operate them.

On price, managed is competitive and probably ahead once salaries are included.

You reject it on capability instead. SQS-style queues offer no replay, no independent consumer groups over the same data, at-least-once only, and first-in-first-out (FIFO) ordering only within a single message group at a few hundred transactions per second (TPS). Every one of those is a requirement in Framing what decision and what breaks, not a preference.

Claiming a price advantage you do not have is how a candidate loses credibility. Rejecting an alternative on the capability the design actually needs is how one gains it.

6. Interviewer pushback

The same material again, in the form it gets asked. The italic text is the answer stated the way it should be delivered, not commentary.

“You said exactly-once is impossible, but Kafka advertises it. Which is it?”

Both, because they mean different scopes. Kafka’s transactional producer makes “read from topic A, write to topic B, commit the offset” atomic — all three endpoints are inside the log, so one coordinator can write markers to all of them. The moment the sink is my database, there is no marker to write there and the guarantee stops at the broker boundary. For that case I use at-least-once plus a dedup key written in the same database transaction as the effect. Two details decide whether that actually works. The key is the producer’s own idempotency key, not topic:partition:offset, because an unclean election renumbers offsets and the same message comes back at a new one. And the effect’s rowcount is checked, because an UPDATE that matches no row does not raise — it would otherwise commit the dedup key, credit nobody, and then refuse the redelivery that would have fixed it. The window is 60 seconds, from the session timeout plus rebalance time (45 + 15 = 60) — 960 MB of dedup state — and that window covers system redeliveries only; a deliberate replay is meant to re-apply, so making replay idempotent is a different table sized by the retention window. That is effectively-once, and it is the only version that survives contact with a non-Kafka sink.

“Why not just one partition? Our volume is small and ordering is simple.”

Then use one partition — but price the ceiling first, because it is lower than people expect. One partition is one broker’s NIC divided by the replication fan-out: 31,250 msg/s. And it is one consumer in the group, so at 10 ms of processing the real cap is 100 msg/s, 313x tighter than the broker. If today’s volume is 20 msg/s that is fine, and I would say so. What I would not do is choose it and discover the ceiling later, because partition count cannot be lowered and raising it breaks hash(key) % P for every existing key.

“Consumer lag is at 4 million and rising. Walk me through it.”

First I check the sign of d(lag)/dt, because that decides whether this is a capacity problem or an incident. If consumption has stopped entirely I look for a rebalance loop — the signature is the group’s generation number, which the coordinator increments once per rebalance, going up every few minutes — and the usual cause is max.poll.records times per-record processing exceeding max.poll.interval.ms; at 10 ms per record the ceiling is 30,000 records. If consumption is merely slower than production, drain time is backlog over surplus, so at 20% headroom 4 million messages takes 20 seconds and I add capacity; at 5% headroom it takes 80 seconds and the fleet is under-provisioned by design. The number I would also check immediately is lag in seconds against the 1.19-hour page-cache window, because crossing it turns memory reads into disk reads for every other consumer on those brokers.

“Is acks=all enough to say you never lose data?”

No, and the gap has a name. acks=all with min.insync.replicas=2 means two replicas have the record in their page cache — in the operating system’s memory — not yet physically on their drives. Kafka deliberately does not call fsync, the system call that forces the operating system to push cached writes onto the device, on every record, because that would put the disk back on the latency path. Two simultaneous machine losses before the OS flushes still loses data. What it does guarantee is survival of one broker loss, which is the failure that actually happens. The one that turns “never lose data” into a lie is unclean leader election: a non-ISR replica promoted after all ISR members die truncates at least 30 seconds of records — 234,240 across the 128 partitions a broker leads — and it does it silently, with offsets going backwards. So my answer is acks=all, min.insync.replicas=2, RF=3, and unclean.leader.election.enable=false on any topic where I would rather be down than wrong.

“Why is the broker not pushing? Pull seems wasteful.”

Pull costs one thing — idle polling — and long poll removes it: the broker parks the fetch until data arrives or 500 ms elapses, so idle request rate drops 5x and the latency when data does arrive is a single datacenter round trip. What pull buys is that backpressure lives on the machine that knows its own capacity. A pushing broker facing a slow consumer must buffer or drop, and both put the consumer’s problem on the broker’s memory. And replay is only expressible if the consumer names the offset, which is pull by definition.

“16,384 partitions seems like a lot. Why not 128?”

Because partition count is set by consumer latency, not by throughput. At 10 ms per message a consumer’s capacity is 100 msg/s, and a million per second is offered load — dividing them straight gives 10,000 consumers at 100% utilization, which is a group that can never work off a backlog. At the 20% headroom I use everywhere else that is 12,000 consumers, and a consumer without a partition is idle, so the partition floor is 12,000. 128 partitions would cap the group at 12,800 msg/s of useful work. The costs of 16,384 are real and I would name them: 384 replicas and 1,152 file handles per broker, 49,152 entries of controller metadata, and dead batching at the producer — which is why the sticky partitioner matters, since it restores 16-message batches and cuts the request rate 16x.

The assumption ledger

Every design is a set of assumptions with a diagram attached, and the diagram is only correct relative to them. The ledger below collects everything the chapter has relied on, so you can state the design’s foundations quickly and say what replaces the design when each one fails.

Sort each assumption into one of three bins:

• State it — you are free to pick, and being wrong costs a re-derivation, nothing more.
• Ask it — the answer moves a policy or a threshold, and is worth an interviewer’s time.
• Load-bearing — if it is wrong the design is not suboptimal, it is invalid. A box appears or disappears, rather than the count inside a box changing.

The one-line test, from ch 03: move the assumption an order of magnitude in each direction and ask whether the set of boxes changes, or only the number of machines inside them.

In the table below, the last column is the useful one — it says what you would build instead if the assumption failed:

The sentence that makes this visible to an interviewer: “This design rests on four things. One, the consumer owns its read position, which is what lets the broker keep no per-message state at all. Two, ordering is needed per key and not globally, which is what makes partitions legal. Three, consumers read sequentially near the tail, which is what keeps the disk at 2% and the reads in memory. Four, the eventual destination is not inside the log, which is what makes exactly-once unavailable and an idempotent consumer mandatory.”

Cheat sheet

Every line below is derived somewhere above. This table is the recall test, not the explanation — cover the right column and see whether you can produce each number and say where it came from.

Related: 01 — Scale From Zero To Millions is where the queue first appears and why; 06 — Key-Value Store is the quorum arithmetic min.insync.replicas reuses; sql 03 is the write amplification that makes the table-as-queue lose; 10 — Notification System is the same exactly-once theorem across a third-party boundary.