dev/consumer

Generated page

Model gemma-mtp, commit f508a4b65b3f, 2026-08-16, sources: 2. Edit the code or the hand-written documentation instead.

ConsumerConfig

Configuration parameters are loaded from environment variables to define how the consumer connects to the broker and where it stores its state, as implemented in Main.kt:160-181.

Environment Variable	Default Value	Description
`CONSUMER_NAME`	`consumer`	Name of the consumer instance
`BOOBLIK_HOST`	`127.0.0.1`	Broker host address
`BOOBLIK_PORT`	`9092`	Broker port
`BOOBLIK_TOPIC`	`events`	The topic to subscribe to
`BOOBLIK_PARTITION`	`0`	The specific partition to consume
`STATE_DIR`	`/var/lib/consumer`	Directory for `FileOffsetStore`
`HTTP_PORT`	`8080`	Port for the metrics HTTP server

The main execution loop, implemented in Main.kt:56-97, maintains a continuous connection to the broker. It uses a while(true) loop to ensure that if a BooblikSubscriber fails or the broker becomes unavailable, the consumer attempts to re-establish the subscription.

More: consumeForever

Recovery after a crash

To ensure at-least-once delivery, the consumer uses FileOffsetStore to persist the last successfully processed offset. Upon startup, consumeForever checks for a saved offset in Main.kt:67-68; if found, it resumes from that position; otherwise, it starts from StartPosition.Earliest. The checkpointing mechanism is applied via .checkpointing(store) in Main.kt:83, ensuring the offset is saved only after the batch has been processed.

Reconnection logic

Connectivity issues are handled within a try-catch block inside the loop in Main.kt:76-95. If an exception occurs (such as the broker going away), the consumer increments the reconnects counter, waits for a 1000ms delay, and then attempts to reconnect.

More: Reconnection logic

Stats and monitoring

The service exposes operational metrics through an embedded Ktor server configured in Main.kt:46-52. The Stats class tracks the number of handled records, the current position, the lag, and the number of reconnection attempts, which are then serialized to JSON via the /stats endpoint.

Key files

File	Lines	What is there
`…/consumer/build.gradle.kts`	`1-8`	Dependency declarations for Ktor, Booblik client, and serialization
`…/consumer/Main.kt`	`38-54`	The `main` function entry point and server setup
`…/consumer/Main.kt`	`56-97`	The `consumeForever` loop and subscription logic
`…/consumer/Main.kt`	`109-145`	The `Stats` class and `ConsumerStats` data class
`…/consumer/Main.kt`	`160-181`	The `ConsumerConfig` data class and environment loading

Behaviour that surprise

The checkpointing function in consumeForever is designed such that the offset is saved after the collect block finishes, which guarantees at-least-once delivery by replaying batches if a crash occurs before the save is completed (Main.kt:80-82).
The StartPosition.Earliest setting used in consumeForever refers to the start of the live log (the data currently retained by the broker) rather than the absolute beginning of the log history (Main.kt:64-66).