shibuya-kafka-adapter

Kafka adapter for the Shibuya queue-processing framework.

Integrates with Apache Kafka via kafka-effectful for the consumer effect (polling, offset store, partition pause) and hw-kafka-streamly for error classification (skipNonFatal), on top of hw-kafka-client. Provides polling, offset commit semantics, partition awareness, and graceful shutdown.

Runtime Contract

Run this adapter with serial message processing only. librdkafka stores the highest offset per partition and does not track gaps, so concurrent finalization can commit past an earlier message that failed, halted, or requested retry. The adapter cannot inspect the processor concurrency policy at construction time; avoiding Async and Ahead processing is a caller contract until a gap-tracking commit layer exists.

Offset reset policy belongs to the Subscription passed to runKafkaConsumer, for example topics [TopicName "orders"] <> offsetReset Earliest. KafkaAdapterConfig.topics is metadata used for adapter naming and a construction-time warning if it differs from the live subscription.

AckRetry seeks the partition back to the failed message and does not store that message's offset. AckDeadLetter stores the offset so the consumer group moves on, prints [shibuya-kafka-adapter] WARNING: dead-lettered message DROPPED to stderr, and makes the message unrecoverable from the group's committed position. This adapter does not include a DLQ producer.

Kafka does not expose a per-message delivery-attempt counter through this consumer API, so Envelope.attempt is always Nothing. Handlers that need bounded retries must use their own store or return AckHalt to stop the stream.

AckHalt pauses the originating partition and stops the processor. Polling then stops; after max.poll.interval.ms (librdkafka default: 300000 ms, or 5 minutes) the broker may evict this consumer and rebalance the partition to another group member. A single-member group stalls until restart.

On shutdown, the adapter commits offsets stored so far and signals the source stream to stop. Let the surrounding runKafkaConsumer scope end normally after stopping the app so the consumer close path can flush offsets stored during the drain window.

For rebalance visibility, create state with newKafkaAdapterState, install Kafka.Consumer.setCallback (Kafka.Consumer.rebalanceCallback (kafkaRebalanceHandler state)) before consumer creation, and pass the same state to kafkaAdapterWith. The helper logs rebalance events and clears retry barriers for revoked partitions; in-flight fencing for cooperative rebalances is out of scope.

Packages

• shibuya-kafka-adapter — the adapter library (Shibuya.Adapter.Kafka, .Config, .Convert).
• shibuya-kafka-adapter-bench — micro-benchmarks for the conversion hot path (ConsumerRecord → Envelope, W3C header extraction, timestamps).
• shibuya-kafka-adapter-jitsurei — runnable examples: BasicConsumer, MultiTopic, MultiPartition, OffsetManagement.

Tracing

Tracing is handled by shibuya-core's supervised runner. Shibuya.Adapter.Kafka.Convert.consumerRecordToEnvelope populates Envelope.attributes with messaging.system=kafka and the Kafka-specific typed attributes messaging.kafka.destination.partition (Int64) and messaging.kafka.message.offset (Int64). The framework merges those onto its single Consumer-kind per-message span alongside messaging.destination.name, messaging.operation.type=process, and messaging.message.id.

The old opt-in Shibuya.Adapter.Kafka.Tracing module was removed in 0.5.0.0; callers should run the Kafka adapter through Shibuya's normal runApp or runWithMetrics path instead of wrapping the stream.

Building

The repo ships a Nix flake and direnv config for a reproducible toolchain.

direnv allow        # or: nix develop
cabal build all
cabal test shibuya-kafka-adapter

Benchmarks and examples:

cabal bench shibuya-kafka-adapter-bench
cabal run BasicConsumer

Layout

shibuya-kafka-adapter/            library sources and tests
shibuya-kafka-adapter-bench/      tasty-bench micro-benchmarks
shibuya-kafka-adapter-jitsurei/   runnable usage examples
docs/plans/                       execution plans
mori.dhall                        project manifest (mori registry)

License

MIT. See package cabal files for details.