Full Buffers and Blocking
Concepts covered: Batching and compression
send() returns at once because it never talks to a broker. It hands the record to an in-memory accumulator, sized by buffer.memory, 32 MiB by default, and the producer's single I/O thread drains batches from there to the leaders. The application fills, the brokers drain, and when the fill outruns the drain the accumulator grows until it reaches buffer.memory. Then send() stops returning at once.
| Quantity | Value | Where it comes from |
|---|---|---|
| record size | 1,024 bytes | measured |
| application send rate | 10,000 records/s | the loop calling `send()` |
| cluster drain rate | 6,000 records/s | acks coming back, limited by broker disks and the network |
| net growth | 4,000 records/s | fill minus drain |
| `buffer.memory` | 33,554,432 bytes | 32,768 records of 1,024 bytes |
| time until full | 8.2 s | 32,768 records at 4,000 per second |
The accumulator is full 8 seconds after the loop starts. The next send() cannot allocate a batch, so it blocks the calling thread until a batch is acknowledged and its memory comes back. From that point the application runs at 6,000 records a second, because the brokers set the pace. That blocking is the producer's backpressure, and it works as long as the brokers keep draining. The wait has a limit: max.block.ms, 60 seconds by default. A send() that waits that long throws TimeoutException, whose message says it failed to allocate memory within the configured max blocking time, and the record it was handed never entered the buffer.
| Clock | Accumulator | What send() does |
|---|---|---|
| 0 s | empty | returns in microseconds |
| 8.2 s | 32,768 records, full | blocks until a batch frees |
| 8.2 s to 60 s | full, draining at 6,000/s | returns after a short wait each, so the loop runs at the drain rate |
| 60 s, network to every leader cut | full, draining at 0/s | blocks for the whole `max.block.ms` |
| 120 s | full | throws `TimeoutException`; the record is dropped on the application's side |
The buffer is 1 pool shared by every partition. A batch for a partition whose leader is down stays in the accumulator while the producer retries it, up to delivery.timeout.ms, 120 seconds by default. With 12 partitions and 1 dead leader, the dead partition's batches stop being freed while the other 11 keep draining, so its share of the pool grows until its batches expire and the memory returns. Whether the pool fills first depends on the rate. With keys evenly spread over the 6,000 records a second the brokers were draining, at 1,024 bytes each, the dead partition's twelfth is 500 records a second. Holding 120 seconds of them would take 60,000 records, about 59 MiB, nearly twice the 32 MiB pool, so the pool fills after about 65 seconds, long before the first batch expires. From then on a send() to a healthy partition blocks as well, because memory has no partition, and 1 slow partition has stalled the whole producer.
What to change, and what each change buys
Raising buffer.memory is the first instinct and the weakest fix. Doubling it to 64 MiB delays the block from 8.2 seconds to 16.4 seconds, after which the application runs at the drain rate anyway: the buffer sizes a burst, never a rate. Rate fixes live on the drain side: compression shrinks what crosses the wire, a longer linger.ms fills larger batches, and more partitions on more brokers add leaders to write in parallel; each raises the 6,000. On the fill side, the application can read the block as a signal: watch the producer's bufferpool-wait-ratio metric and slow the loop that calls send() when it climbs. A max.block.ms of 2,000 turns a silent minute into an exception the application can act on by shedding the record, spilling it locally or pausing its intake.
- Java client
- buffer.memory: 32 MiB of batches, shared across partitions
- send() blocks when the pool is empty
- max.block.ms 60 s, then TimeoutException
- the calling thread pays the wait
- librdkafka
- queue.buffering.max.messages 100,000 and queue.buffering.max.kbytes 1,048,576
- produce() returns RD_KAFKA_RESP_ERR__QUEUE_FULL at once
- no wait: the application polls, then retries the call
- the application decides how long to retry
The 2 libraries hand the same problem to the application in 2 ways. The Java client absorbs it as time: the thread stops. librdkafka returns it as an error: the call fails and the record is still the application's. A working librdkafka loop polls for delivery reports to free queue slots, then calls produce() again; a working Java loop catches TimeoutException and does the same. Either way the refused record is lost unless the application keeps it, and delivery.timeout.ms never starts for it, because that clock begins when a record enters the buffer.
- Do
- Measure the drain rate before sizing the buffer; the buffer covers bursts above it.
- Set max.block.ms to what the caller can tolerate, and handle TimeoutException.
- Avoid
- Raise buffer.memory past a few seconds of fill without raising the drain rate.
- Call send() from a request handler with a 60 s block and no timeout of your own.
About this lesson
This section is part of the Producers: Advanced 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.