Directional modes

Diagram

Directional Configuration and Environment

The behavior of the relay is determined by the Direction enum, which is parsed from the RELAY_DIRECTION environment variable (defaulting to KAFKA_TO_BOOBLIK) in Main.kt:159-163. The RelayConfig class encapsulates all necessary parameters, including bootstrap servers and topic names, which are loaded from the environment in Main.kt:157-174.

Kafka to booblik: Consumer Group and Manual Commit

When operating in the KAFKA_TO_BOOBLIK direction, the relay acts as a Kafka consumer. To ensure delivery guarantees, ENABLE_AUTO_COMMIT_CONFIG is explicitly set to false in Directions.kt:49. The process follows a strict lifecycle: it polls records, sends them to booblik via a Producer, and only after the booblik acknowledgements are received does it call consumer.commitSync() in Directions.kt:71. This ensures that the Kafka consumer group position only advances once the data is safely in booblik.

booblik to Kafka: FileOffsetStore and Checkpointing

In the BOOBLIK_TO_KAFKA direction, the broker does not track the reader's position. Instead, the relay uses a FileOffsetStore to persist the state locally on a volume, as seen in Directions.kt:93. The booblikToKafka function uses a checkpointing operator that saves the offset only after the collector has successfully handled the batch, ensuring that the FileOffsetStore is updated only after the Kafka producer has finished its work (Directions.kt:112).

The Key-to-Partition Mapping

The relay handles Kafka keys differently depending on the direction. In kafkaToBooblik, the Kafka record key is passed to handle.send(record.value(), key = record.key()) in Directions.kt:66, which allows the key to influence partition selection in booblik, preserving per-key ordering. However, because the booblik wire format does not contain a field for the key, the key itself is not stored and is lost for any subsequent round trip (Main.kt:37-42).

Key files

Behaviour that surprises

• The kafkaToBooblik function uses withContext(Dispatchers.IO) to wrap the blocking consumer.poll and consumer.commitSync calls to avoid blocking the coroutine scope (Directions.kt:62, 71).
• In the BOOBLIK_TO_KAFKA direction, the relay uses kafka.flush() before calling stats.observe to ensure that all buffered records are actually sent to Kafka before the position is updated (Directions.kt:119).
• The RelayConfig uses replace logic for environment variables, where the direction string is transformed by converting it to uppercase and replacing hyphens with underscores (Main.kt:160-162).