Producer

What this module is responsible for

The Producer is responsible for accumulating individual records into batches to maximize throughput by reducing the number of network requests. It acts as an intermediary between the user's send calls and the BooblikConnection, managing the lifecycle of records from their arrival in a mailbox to their delivery to the broker.

Diagram

The Accumulator

The core of the producer is the accumulation loop, which manages pending records in a LinkedHashMap of Key to Batch (Producer.kt:70-71). Instead of sending records immediately, the producer waits until either the maxBatchSize is reached or the lingerMillis window expires (Producer.kt:23-32). Crucially, the timer for the lingerMillis window starts from the arrival of the first record in a batch, not the last, to prevent a steady trickle of records from indefinitely postponing a send (Producer.kt:47-49).

Command.Append and Command.Flush

The runLoop operates by consuming commands from a buffered mailbox channel (Producer.kt:59).

• Command.Append: Adds a record to the corresponding partition's batch via accumulate (Producer.kt:132-134).
• Command.Flush: Triggers an immediate sendAll() and completes the command's deferred value, forcing all currently accumulated batches to be delivered to the connection (Producer.kt:126-128).

AckPolicy and Batch Delivery

The deliver function handles the actual network transmission and the subsequent response from the broker (Producer.kt:195-220). The behavior of the caller depends on the AckPolicy:

The Lost Record Edge Case

A critical edge case occurs when a receive operation on the mailbox is cancelled (e.g., due to a timeout in select) at the exact moment a record is being delivered to the channel. If withTimeoutOrNull were used instead of select, a cancelled receive could swallow a record, leaving the caller's CompletableDeferred hanging forever (Producer.kt:143-146). This specific race condition is verified in ProducerLostRecordTest.kt:52-94, where two topics at different rates are used to force the collision.

Batching Performance and Throughput

Batching is the single largest performance factor in the project. As noted in benchmarking.md:140-146, the difference between sending records one at a time and sending them in batches of 100 can result in a performance increase of approximately 54× (Producer.kt:39-42). The ProducerConfig allows tuning this via maxBatchSize and lingerMillis (ProducerConfig.java:4-16).

Key files

Behaviour that surprises

• The lingerMillis timer is not reset on every new record; it is fixed from the moment the first record of a batch arrives (Producer.kt:47-49).
• Using select with onReceiveCatching is mandatory in the runLoop to prevent the producer from "swallowing" records during a timeout, which would cause callers to hang (Producer.kt:154-158).
• AckPolicy.NONE does not actually wait for a broker response, but it still completes the record's CompletableDeferred with Offset.ZERO to maintain a consistent API (Producer.kt:204-206).