booblikToKafka

What this module is responsible for

The booblikToKafka relay acts as a bridge that consumes data from a booblik topic and produces it into a Kafka topic. Unlike the reverse direction, this relay is responsible for managing its own state because the booblik broker does not track consumer positions.

Diagram

The booblikToKafka lifecycle

The lifecycle begins with the BooblikSubscriber.follow method, which initiates a stream from the booblik topic using a position retrieved from the FileOffsetStore (Directions.kt:110-111). The stream is processed via the collect operator, which receives batches of records. The checkpointing operator is applied to the flow, ensuring that the OffsetStore is updated only after the collector has successfully processed the batch (Directions.kt:112).

The FileOffsetStore mechanism

Since the booblik broker does not remember readers, the relay maintains state using a FileOffsetStore on its own volume (Directions.kt:93). To ensure position integrity, the save method writes the offset to a temporary file and then performs an atomic move using StandardCopyOption.ATOMIC_MOVE (FileOffsetStore.kt:57). This prevents a partially written or truncated offset from being parsed during a restart (FileOffsetStore.kt:25).

At-least-once delivery and kafka.flush

The relay guarantees at-least-once delivery through a specific sequence of operations within the collect block. First, all records in a batch are sent to Kafka via kafka.send (Directions.kt:117). Second, kafka.flush() is called to ensure all buffered records are actually acknowledged by the Kafka broker (Directions.kt:119). Only after the flush is successful does the checkpointing mechanism save the batch.nextOffset to the FileOffsetStore (Directions.kt:120).

The mirrored topic and partition mapping

When moving data from booblik to Kafka, the relay faces a limitation: booblik's wire protocol does not store keys, only the partitionId (README.md:177-179). Consequently, when the relay calls kafka.send(ProducerRecord(config.kafkaTopic, record)), it does not provide a key (Directions.kt:117). This means the distribution of records into Kafka partitions is determined by Kafka's default partitioning logic rather than any specific key-based rule chosen by the original producer.

Key files

Behaviour that surprises

• At-least-once duplicates: Because the checkpointing happens after kafka.flush(), a crash between these two calls will result in the same batch being re-sent to Kafka upon restart (Directions.kt:119-120).
• Key loss: While the TopicHandle.send in the kafkaToBooblik direction uses the Kafka key to pick a booblik partition, the booblikToKafka direction loses this key entirely because it is not part of the booblik wire protocol (README.md:177-180).
• Silent corruption prevention: The FileOffsetStore uses a .tmp file and an atomic move specifically to avoid the "half-saved" offset bug where a truncated file could lead to silent data replay (FileOffsetStore.kt:24-25).