kafkaToBooblik

What this module is responsible for

The kafkaToBooblik function acts as a bridge that translates data from a Kafka topic into the booblik format. It consumes batches of records from Kafka and pushes them into a booblik topic, ensuring that the data is safely acknowledged by the booblik broker before the Kafka consumer advances its position.

Diagram

KafkaConsumer configuration and properties

The Kafka consumer is configured to ensure manual control over the message lifecycle. It uses ByteArrayDeserializer for both keys and values to treat all data as raw bytes Directions.kt:47-48. Crucially, ENABLE_AUTO_COMMIT_CONFIG is set to false to prevent the consumer from automatically advancing offsets before the data is safely stored in booblik Directions.kt:49. The AUTO_OFFSET_RESET_CONFIG is set to earliest to ensure that a new relay can catch up on all existing data in the topic Directions.kt:51.

The at-least-once delivery guarantee

The relay implements an at-least-once delivery guarantee through a strict sequence of operations in the polling loop. First, it maps the records to handle.send calls, which are asynchronous acknowledgements Directions.kt:66. It then calls await() on these acknowledgements to ensure every record in the batch has been processed by the booblik broker Directions.kt:67. Only after all records are acknowledged does it call consumer.commitSync() Directions.kt:71. This order ensures that if a crash occurs between sending and committing, the batch will be re-delivered upon restart Directions.kt:70.

Per-key ordering preservation

Even though the booblik wire protocol does not store Kafka keys research-usecases.md:37-41, the relay preserves per-key ordering during the crossing. When handle.send(record.value(), key = record.key()) is called, the Kafka key is used to determine the target booblik partition Directions.kt:66. Because the key determines the partition, all records with the same key are routed to the same booblik partition, maintaining their relative order Directions.kt:34-35.

The relay restart loop and error handling

The relay is designed to be resilient against transient failures on either side of the connection. The main loop is wrapped in a while (true) block within a CoroutineScope initialized with a SupervisorJob Main.kt:54-55. If an exception occurs, the error is caught, logged, and the loop restarts after a 2-second delay Main.kt:61-68. This prevents the relay from dying permanently if one side (Kafka or booblik) experiences a temporary outage Main.kt:62-63.

Relay statistics and observability

Observability is provided through the Stats class, which tracks the progress of the relay. The observe function updates several metrics whenever a batch is processed Main.kt:102-111.

The RelayStats data class provides a serializable snapshot of these metrics for the HTTP /stats endpoint Main.kt:129-140.

Key files

Behaviour that surprises

• Key Loss: While handle.send uses the key to maintain partition ordering, the key itself is not stored in booblik, meaning it cannot be recovered when reading from booblik back to Kafka research-usecases.md:37-41.
• At-least-once Duplicates: Because consumer.commitSync() happens after the booblik write, a crash between these two steps results in the same records being sent again upon restart, making duplicates a legal and expected behavior Directions.kt:70-72.