The wire protocol reference implementation

Diagram

The Connection framing and length prefixing

The protocol uses a length-prefixed framing mechanism where each message is preceded by a 4-byte integer indicating the size of the following frame Protocol.kt:24. In the Python reference implementation, the _send method packs the API key, version, and correlation ID into a header before sending the payload, all wrapped in a length prefix wire.py:81-82. The _receive method performs the inverse, first reading the 4-byte length and then the exact number of bytes required to complete the frame wire.py:86-87.

Request decoding and DecodeResult

Decoding is a two-stage process designed to protect the broker. First, the RequestDecoder attempts to parse the header to identify the ApiKey and correlationId Requests.kt:79-96. If the header is valid, the decoder proceeds to parse the body; if the header itself is malformed, it returns a DecodeResult.Failed containing an UNKNOWN_CORRELATION_ID to prevent the broker attempting to respond to an invalid ID Requests.kt:97-98.

The Session protocol loop

The Session class manages the lifecycle of a single connection through a continuous loop in the serve function Session.kt:47-53. The loop follows a strict sequence:

1. readFrame(): Reads the next complete frame from the connection Session.kt:50.
2. handle(frame): Dispatches the decoded request Session.kt:51.
3. respondError or writeFully: Sends the response back to the client booblik-net/src/main/kotlin/ru/workinprogress/booblik/net/Session.kt:88, 151. This loop ensures that requests are served strictly in order, one at a time, to maintain consistency with the client's expectation of response ordering Session.kt:30-34.

The FetchMode and zero-copy path

The broker supports two distinct modes for transferring data during a fetch request Session.kt:219-230:

Handling partial frames and connection survival

The protocol is designed to be robust against network fragmentation. A frame may be split across multiple TCP packets; the Session handles this by reading the full length of the frame before attempting to decode it Session.kt:68-71. However, if a frame claims an "absurd" length that exceeds Protocol.MAX_FRAME_BYTES, the broker throws a CorruptRequestException and drops the connection to protect its memory booblik-net/src/main/kotlin/ru/workinprogress/booblik/net/Session.kt:69, booblik-protocol/src/commonMain/kotlin/ru/workinprogress/booblik/net/wire/Protocol.kt:31. Tests in PartialFrameTest.kt:112-128 verify that while split packets are assembled correctly, invalid lengths result in connection closure.

The awaitRecords mechanism

When a FetchRequest includes a maxWaitMillis and minBytes requirement, the broker enters a suspension loop in awaitRecords Session.kt:246-267. Instead of busy-waiting, the coroutine suspends using handle.writer.highWatermark.first { it > seen }, which resumes only when the log's watermark advances Session.kt:261. This ensures the broker does not waste CPU cycles while waiting for more data to become available in the log.

Key files

Behaviour that does surprise

• Silent Silence: In produce requests, if the ackPolicy is set to ACK_NONE, the broker returns no response at all rather than an empty response, because an offset does not exist until the writer reaches the batch wire.py:149-150.
• The Metadata Exception: Unlike other requests, MetadataRequest is dispatched before the topic and partition are even parsed, because it is the only request that does not name a specific partition Requests.kt:110-115.
• Clamped Waiting: If a client requests a maxWaitMillis longer than 60 seconds, the broker clamps the wait to MAX_FETCH_WAIT_MILLIS to prevent holding coroutines and sockets open indefinitely Protocol.kt:44.