Producer

What this module is responsible for

The Producer is responsible for accumulating individual records into batches organized by partition to maximize throughput. Instead of sending every record as a separate network request, it uses an internal accumulator to group records, significantly reducing the overhead of the broker's write path Producer.kt:39-42.

Diagram

The Accumulator and lingerMillis

The core of the producer is the accumulation of records into Batch objects, grouped by a Key consisting of a topic and a partition Producer.kt:70-71. The timing logic is designed to prevent indefinite delays: the window for a batch is determined by the arrival of the first record in that batch, using config.lingerMillis to set a deadline Producer.kt:139-141. This ensures that a steady trickle of records cannot postpone a send indefinitely, which would turn a latency bound into a "latency hope" Producer.kt:137-138.

The runLoop and select mechanism

The runLoop is the internal execution loop that processes commands from the mailbox channel Producer.kt:129-131. To avoid a critical bug where a cancelled receive might swallow a record and leave a caller hanging forever, the loop uses a select expression Producer.kt:154-158. This ensures that the select either takes the element from the channel or takes the timeout, but never both and never neither, preventing the "lost record" scenario Producer.kt:153.

The batch function and contiguous offsets

For use cases requiring strict atomicity and contiguity, the batch function allows callers to bypass the accumulator entirely Publishing.kt:85-90. This function ensures that the records provided in the block are sent as a single request, guaranteeing that the resulting offsets are contiguous (e.g., base, base + 1, etc.) Publishing.kt:98.

Lifecycle and drainPending

When close() is called, the producer shuts down the mailbox and the underlying dispatcher Producer.kt:124-127. To prevent silent data loss, the drainPending function is called in a finally block Producer.kt:88-90. This function ensures that any records remaining in the mailbox or the pending map are either sent or completed with an exception Producer.kt:227-244.

Failure modes and ProduceFailedException

If the broker returns an error code other than ErrorCode.NONE, the producer propagates this failure to all callers in the batch Producer.kt:219-222. This is handled by completing the CompletableDeferred handles with a ProduceFailedException Producer.kt:209-211.

Performance modes: DIRECT, BATCHED, and BATCHED_NOT_AWAITED

The impact of the accumulator is measured using three distinct execution patterns Probe.kt:79-93:

Testing for record loss in the accumulator

The producer's reliability is verified by ProducerLostRecordTest, which simulates a high-collision environment ProducerLostRecordTest.kt:41-48. The test specifically drives two different topics at different rates through a single producer to ensure that the select mechanism and the runLoop do not drop records when a timeout and a new record arrival collide ProducerLostRecordTest.kt:52-58.

Key files

Behaviour that surprising

• The lingerMillis timer is reset only at the start of a batch, not on every new record arrival Producer.kt:137-138.
• newSingleThreadContext is considered a "delicate API" because the thread it creates must be manually closed via close() to avoid leaking the thread Producer.kt:49-51.
• In Producer.kt:205, AckPolicy.NONE results in an Offset.ZERO being returned to the caller to represent that the batch was sent.