Batching and Linger

Concepts covered: Batching and compression

One request carries many batches

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.

Three ways a batch becomes ready

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 batches under load

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.

Who pays the wait, and what it buys

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.

Do

• Raise linger.ms in steps and watch requests a second fall at each step
• Size batch.size to what one partition receives inside one linger window
• Measure delay at the first record of a batch, since it waits longest

Avoid

• Assume linger.ms=0 means 1 record per request
• Raise batch.size past what a partition fills before linger expires
• Add linger to a topic whose partitions each get 1 record a second

About this lesson

This section is part of the Producers: Intermediate lesson on DataDriven, the home of data engineering. Lessons cover SQL, Python, Spark, data modeling and pipeline architecture, and each section pairs its explanation with worked examples and practice you run as you go.