Producers: Intermediate

A call to send() appends the record to the open batch for its partition in the record accumulator, the producer's buffer. It returns a future before any byte reaches the network. A separate sender thread drains the accumulator: it takes every batch that is ready, groups the batches by the broker that leads their partitions, and sends 1 produce request per broker. A producer writing to 12 partitions led by 3 brokers can ship 12 batches in 3 requests, and the brokers see 3 requests' worth of handling per round instead of 12.

A batch is ready once its bytes reach batch.size, which defaults to 16384. It is also ready once linger.ms has passed since its first record arrived. There is a third way. When the sender builds a request for a broker, it takes the first batch of every partition that broker leads, even a batch that is neither full nor past its linger. A 3-record batch for partition 3 rides along with a full batch for partition 7 when the same broker leads both, because sending it later would cost a second request. batch.size caps 1 batch and a batch never spans 2 partitions, so a partition that receives 1 record a second ships 1-record batches whatever the ceiling says.

linger.ms=0 tells the sender to ship whatever is ready the moment it looks, which differs from 1 record per request. max.in.flight.requests.per.connection, 5 by default, lets the sender keep 5 requests to a broker unanswered; once the 5 slots are full, records pile into the open batch until a response frees a slot. At linger.ms=0 the batch size is therefore set by how quickly the broker answers. At 1000 records a second to a partition whose broker answers in 10 ms, the 5 slots fill at 500 requests a second, so each request carries about 2 records; a broker that answers in 40 ms allows 125 requests a second and batches of 8. Linger sets a floor under the wait: a batch that would have shipped after 2 ms of accumulation holds on for 20 and ships 10 times larger.

The model behind the table: 2000 records a second of 200 bytes each to 1 partition, a broker that answers in 10 ms, and max.in.flight.requests.per.connection at 5.

The scene holds max.in.flight.requests.per.connection at 1, so each batch waits for the answer to the one before it. In it, 8 records arrive 5 ms apart. At linger.ms=0 the first record leaves alone, the records that land during its round trip leave together behind it, and the run takes more than 1 request. At linger.ms=40 the first record opens a batch and waits; the other 7 arrive inside the window, and the whole run leaves in 1 request 40 ms after the first record was sent. The first record paid 40 ms for that, the last paid about 5.

Linger charges each record in a batch a different fee. The first record waits the full linger.ms; the last arrives as the batch closes and waits nothing; a record's share of the cost depends on when it showed up. The fee buys fewer requests. A request costs the broker the same work whether it carries 1 record or 400: the broker reads it off the socket and a handler thread appends it. Each batch carries 1 header of 61 bytes on the wire and on disk, shared by every record inside it. A broker handling 50 requests a second of 40 records does a small fraction of the work of one handling 2000 requests of 1 record carrying identical data. More linger keeps helping until batches hit batch.size before linger expires; past that point it changes nothing, and the ceiling is the setting to raise.

To predict a change for keyed records, count records per partition per linger window, because the keys split the traffic across partitions before batching sees it. A topic of 60 partitions taking 6000 keyed records a second, spread evenly, gives each partition 100 a second, so a 50 ms linger gathers 5 records a batch, and reaching 100 a batch would need a full second of linger. A topic over-partitioned for its volume ships small batches at any linger, which is one more cost of the partition count you chose. Keyless records behave differently: the default partitioner sends them to 1 partition until about batch.size bytes have gone there, then moves on, so their batches fill at the topic's full rate.

compression.type on the producer takes none for no compression, or names one of the codecs gzip, snappy, lz4 and zstd. The producer compresses records into the batch as it appends them, so the unit of compression is the batch. A codec earns its ratio by finding repeats, and in event data the repeats sit across records rather than inside one: 100 order events share the same field names, the same handful of store ids and the same timestamp prefix. A batch of 100 compresses well. A batch of 1 has nothing to repeat against, and under gzip it comes out larger than it went in, because the codec writes a 10-byte header and an 8-byte trailer of its own around bytes it couldn't shrink.

The model for the table: each event is 220 bytes, 150 of them structure repeated from one event to the next and 70 that differ. The codec writes the first copy of the structure in full and each later repeat as a 3-byte reference, plus 18 bytes of gzip framing per batch.

The topic setting compression.type defaults to producer, which tells the broker to keep whatever codec the producer used. Set it to a named codec instead and the broker decompresses every batch that arrives in a different codec and recompresses it, which moves the CPU cost onto the broker, the one machine in the path that is hardest to scale out. The 1 reason to do it is a fleet of producers you do not control sending uncompressed data to a topic whose disk you pay for.

Because the ratio comes from repeats across records, the same codec on the same data gives a different answer at every batch size. The batch size depends on linger.ms and batch.size as well as on the record rate per partition from the previous section. A producer at linger.ms=0 shipping 4-record batches may measure gzip at 1.4 times and conclude compression isn't worth the CPU; the same producer at linger.ms=20 ships 40-record batches and measures 3 times. There's a second coupling. The producer checks batch.size against the batch's estimated compressed size, using a ratio it learns from recent batches, so a 16384-byte ceiling with a 3 times ratio holds around 48 KB of raw records. Turning compression on makes batches hold more records, which raises the ratio further. The growth stops when linger expires first or the record rate can't fill the batch.

A retry is only safe when the first attempt did not land. The case that breaks that assumption is a lost answer. A producer sends a batch with acks=all; the leader appends it, then waits for the followers to copy it before answering. One follower is slow, the answer misses request.timeout.ms, and the producer, which cannot tell a slow answer from a lost request, sends the batch again. The leader has no memory of the first attempt beyond the bytes in its log, so it appends the batch a second time. The partition now holds the same payment twice, and every reader will see both.

With enable.idempotence=true, the 4.4 default, the producer asks a broker for a producer id on its first send and then stamps every batch with that id and a sequence number. The sequence counts records per partition from 0, so a batch carries its first record's sequence and the leader derives the rest from the record count. The leader keeps, for each producer id on each partition, the sequence ranges of the last 5 batches it appended. A batch whose sequence it already appended is answered with the original offset and appended again never. A batch whose first sequence is higher than the next expected number is rejected with OutOfOrderSequenceException, because something before it went missing.

Idempotence needs acks=all and a positive retries. It also caps max.in.flight.requests.per.connection at 5, which matches the window the leader keeps. The 4.4 producer applies these for you when the setting is on, and refuses to start with a ConfigException if you set one of them to a conflicting value, such as acks=1.

The window of 5 is also why max.in.flight.requests.per.connection is capped at 5 for an idempotent producer. The leader recognises a duplicate only of a batch it still remembers, and with 6 requests in flight the oldest could fall out of the window before its retry arrived, turning a safe retry into a second copy.

max.in.flight.requests.per.connection lets the sender have up to 5 produce requests to 1 broker unanswered at once. Throughput depends on it: 1 in flight means 1 request per round trip, 5 means 5. It also opens a door. Batch A with records 10 to 12 for partition 0 goes out at t=0; batch B with records 13 and 14 for the same partition goes out at t=2 while A is still unanswered. The leader rejects A with NotEnoughReplicas because the in-sync set dipped below the minimum for an instant, then the follower rejoins and the leader appends B at t=5. The producer waits out retry.backoff.ms and retries A at t=102; the leader appends it at t=105. The partition now reads 13, 14, 10, 11, 12.

Setting max.in.flight.requests.per.connection=1 closes the door with no help from the broker: B can't leave until A is answered, so a rejected A is retried before B ever goes. The price is throughput per broker connection, which falls to 1 request per round trip. A producer whose broker answers in 10 ms is capped at 100 requests a second per broker. How many records each of those 100 carries is then up to linger and batch.size. Producers that wanted order with retries before idempotence existed ran this way and paid for it.

The idempotent producer keeps order at 5 in flight because the leader checks sequences exactly. Say this producer has already sent 10 records to partition 0, so A's first sequence is 10 and B's is 13, matching the record labels. The leader rejects A as before. When B arrives, the leader is expecting sequence 10. It finds 13 and rejects B with OutOfOrderSequenceException instead of appending it. The producer reads that as a signal to retry in sequence order: it re-queues A ahead of B, sends A, waits for A's acknowledgement, then sends B. The log ends with 10 through 14 in order, and at every other moment up to 5 requests stay in flight.

delivery.timeout.ms bounds the retrying, with retries left at its default. The 4.4 producer retries until that budget, 120000 ms by default, runs out, measured from the moment send() returned, and the budget covers linger, every attempt's request.timeout.ms of 30000 and every retry.backoff.ms between attempts. A batch still unacknowledged when the budget ends fails with a TimeoutException to the application, and the batches behind it in the same partition continue with their sequences adjusted so the gap does not block them.

acks sets how many copies exist when the producer counts a send as done. With acks=0 the producer counts it done when the bytes leave the socket, so a leader that dies while the request is in the air loses the record and the producer never hears. With acks=1 the leader answers after appending to its own log, before any follower has fetched the batch; if the leader dies in that gap, the new leader is a follower that never got the record, and the record is gone after the producer was told it was safe. With acks=all, the 4.4 default, the leader answers only when every replica in the in-sync set has appended the batch, which is the moment the high watermark passes it.

acks=0 has a place: a stream of sensor samples where the next reading replaces the last, sent by a producer that must never wait on the cluster. Everywhere a record is a fact rather than a sample, it's the wrong setting.

acks=all waits for the in-sync set, and the in-sync set shrinks when followers fall behind. A topic with 3 replicas whose 2 followers both fell out of sync has an in-sync set of 1, and acks=all then means the leader alone, the same promise as acks=1. min.insync.replicas on the topic sets the smallest in-sync set the leader will accept an acks=all write with. At 2, a leader whose in-sync set has dropped to itself rejects the batch with NotEnoughReplicas. That error is retriable, so the producer retries until a follower catches up or delivery.timeout.ms expires. The count includes the leader, so 2 means the leader plus 1 follower.

A sibling error, NotEnoughReplicasAfterAppend, arrives when the in-sync set shrank after the leader appended the batch. The record sits in the leader's log unacknowledged, and the retry that follows is exactly the retry the previous sections made safe: the idempotent leader answers it from the window without a second append.

The row to run is 3 replicas, min.insync.replicas=2, acks=all. Any 1 broker can die and the partition keeps taking writes, because 2 in-sync copies remain; and every acknowledged record sits on at least 2 brokers, so the death of 1 can't remove it. Lose 2 of the 3 and writes stop rather than proceed on a single copy, which is the point: the producer gets an error it can retry instead of an acknowledgement it can't trust.

Set the floor on the topic so each topic states its own contract: bin/kafka-configs.sh --bootstrap-server broker0:9092 --alter --entity-type topics --entity-name payments --add-config min.insync.replicas=2. A topic created without it inherits the broker's min.insync.replicas, which defaults to 1.

The producer's acks, the topic's min.insync.replicas and the topic's replication factor are 3 settings owned by 3 people, and the promise holds only when all 3 agree. A producer at acks=all writing to a topic with min.insync.replicas=1 has a contract of 1 copy the moment a follower lags. A topic at min.insync.replicas=2 written by a producer at acks=1 never consults the floor, because the leader checks the floor only for acks=all requests. A topic with 2 replicas and min.insync.replicas=2 holds the promise but stops writing the moment either broker restarts for an upgrade.

Readers close the loop. A consumer fetches up to the high watermark, and the high watermark advances only when every in-sync replica has the record, so a record acknowledged under acks=all with a floor of 2 is readable the instant the producer hears back and sits on 2 brokers when it's read. A record acknowledged under acks=1 becomes readable only when the followers catch up, and if the leader dies first it never does.