BooblikConnection

Diagram

BooblikConnection

The BooblikConnection class provides a blocking, synchronous connection to a single broker using POSIX sockets via Socket.kt:33. It is designed to be non-thread-safe because requests and responses are matched by a correlation ID in the order they were sent, meaning two callers sharing a connection would read each other's answers (Connection.kt:23-24).

The correlationId matching mechanism

To ensure data integrity in a pipelined environment, the connection uses a correlationId that is incremented for every request (BooblikConnection.kt:108). Because the broker is guaranteed to answer in strict FIFO order, the client uses a ConcurrentLinkedQueue of Pending objects to match responses to their original callers (BooblikConnection.kt:60). If a response arrives with a correlation ID that does not match the head of the queue, the client fails loudly to prevent delivering the wrong data to a caller (BooblikConnection.kt:35-37).

AckPolicy.NONE behavior

When using AckPolicy.NONE, the produce function returns null immediately (Connection.kt:59). This is because no response is expected from the broker, and since no offset exists until the writer actually reaches the batch on the broker side, there is no answer to wait for (Connection.kt:44-45). In this mode, the request is enqueued to the outbound channel but no Pending object is registered to wait for a result (BooblikConnection.kt:111).

Topic and Partition routing

Routing is handled through the Topic class, which resolves partitions using metadata fetched from the broker (Connection.kt:145). Partition selection follows these rules:

Frame assembly and partial reads

The client is designed to handle fragmented network traffic where a single frame might arrive across multiple packets. The readFrame function in Socket.kt:123 first reads a length prefix and then continues to call recv until the exact number of bytes requested is received (Socket.kt:137-140). Tests in PartialFrameTest.kt:40-60 verify that frames split across two packets or even delivered one byte at a time are correctly assembled into a single request.

Connection failure and error handling

The connection lifecycle includes several error states:

• Absurd Frame Lengths: If a broker sends a frame length exceeding Protocol.MAX_FRAME_BYTES, a ConnectionException is thrown to prevent massive memory allocations (Connection.kt:168-170).
• Unknown API Keys: If a header is valid but the API key is unknown, the broker responds with UNSUPPORTED_VERSION while still echoing the correlationId (ErrorCodeTest.kt:37-39).
• Socket Closures: If the socket closes mid-frame, the client throws a ConnectionException indicating how many bytes were still expected (Socket.kt:137-138).

Key files

Behaviour that inspires

• Topic.partitionFor(null): Calling this with a null key advances the internal roundRobin counter, meaning that even just "asking" for a partition can change the result of the next call (Connection.kt:192-193).
• BooblikConnection.produce: When ackPolicy is AckPolicy.NONE, the function returns null instead of an empty result, because no offset exists until the batch is written (Connection.kt:59).