# shibuya-kafka-adapter
Consume Kafka messages through the [Shibuya](mori://shinzui/shibuya)
queue-processing framework, with explicit acknowledgements, partition-aware
retries, offset management, and OpenTelemetry context propagation.
**Release:** [0.12.0.0 on Hackage](https://hackage.haskell.org/package/shibuya-kafka-adapter-0.12.0.0)
· [API documentation](https://hackage.haskell.org/package/shibuya-kafka-adapter-0.12.0.0/docs/Shibuya-Adapter-Kafka.html)
· [Changelog](CHANGELOG.md)
The adapter requires `shibuya-core ^>=0.12.0.0` and GHC 9.12. The repository's
Nix shell uses GHC 9.12.4. It builds on
[kafka-effectful](mori://shinzui/kafka-effectful) for consumer operations and
[hw-kafka-streamly](mori://shinzui/hw-kafka-streamly) for poll-error classification.
## Using the adapter
Add these dependencies to your application:
```cabal
build-depends:
base ^>=4.21,
effectful-core,
hw-kafka-client >=5.3 && <6,
kafka-effectful ^>=0.3.1.0,
shibuya-core ^>=0.12.0.0,
shibuya-kafka-adapter ^>=0.12.0.0
```
The consumer scope owns the Kafka connection and subscription. Create the
adapter inside that scope and use `mkProcessor`, which defaults to serial
processing. Disable automatic offset storage so a record's offset is stored
only after acknowledgement.
This example prints messages from `orders` on `localhost:9092`:
```haskell
{-# LANGUAGE GHC2024 #-}
{-# LANGUAGE OverloadedRecordDot #-}
{-# LANGUAGE OverloadedStrings #-}
module Main (main) where
import Control.Monad.IO.Class (liftIO)
import Effectful (runEff)
import Effectful.Error.Static (runError)
import Kafka.Consumer.Types (OffsetReset (Earliest))
import Kafka.Effectful.Consumer qualified as Kafka
import Shibuya.Adapter.Kafka
( BrokerAddress (..), ConsumerGroupId (..), KafkaError, TopicName (..)
, defaultConfig, kafkaAdapter
)
import Shibuya.App (ProcessorId (..), defaultAppConfig, mkProcessor, runApp, waitApp)
import Shibuya.Core.Ack (AckDecision (AckOk))
import Shibuya.Core.Ingested (Message (..))
import Shibuya.Core.Types (Envelope (..))
import Shibuya.Telemetry.Effect (runTracingNoop)
main :: IO ()
main = do
result <- runEff . runError @KafkaError . runTracingNoop $ do
let props = Kafka.brokersList [BrokerAddress "localhost:9092"]
<> Kafka.groupId (ConsumerGroupId "orders-consumer")
<> Kafka.noAutoOffsetStore
sub = Kafka.topics [TopicName "orders"] <> Kafka.offsetReset Earliest
Kafka.runKafkaConsumer props sub $ do
adapter <- kafkaAdapter (defaultConfig [TopicName "orders"])
let handler message = do
liftIO (print message.envelope.payload)
pure AckOk
app <- runApp defaultAppConfig
[(ProcessorId "orders", mkProcessor adapter handler)]
case app of
Left err -> liftIO (fail (show err))
Right handle -> waitApp handle
case result of
Left err -> fail (show err)
Right () -> pure ()
```
`defaultConfig` polls with a 1,000 ms timeout and a maximum batch size of 100.
Its `topics` field supplies adapter metadata and is checked against the live
subscription. Subscription topics and offset-reset policy belong to
`runKafkaConsumer`.
## Acknowledgements and ordering
**Use serial processing.** The adapter does not support Shibuya's `Async` or
`Ahead` policies. librdkafka stores the highest offset per partition without
tracking gaps; concurrent finalization can commit past unresolved work. The
adapter cannot inspect or enforce the processor's concurrency policy.
| Decision | Adapter behavior |
| --- | --- |
| `AckOk` | Store the record's offset locally; auto-commit or consumer close flushes it to the broker. |
| `AckRetry` | Keep the earliest unresolved delivery as the retry boundary, seek back to it, and leave its offset unstored. |
| `AckDeadLetter` | Store the offset and log a `dead-lettered message DROPPED` warning. The adapter has no DLQ producer. |
| `AckHalt` | Pause the originating partition without storing its offset; the processor stops. |
With serial single-message processing, a retry blocks later messages in the
same partition. Pre-seek records are discarded, already-buffered successors
are skipped before handler entry, and replayed successors wait until the
retried record resolves. Later callbacks cannot replace the earliest retry
boundary or store offsets past it. A retry cannot undo messages already
admitted together in the same handler batch.
If retention or compaction removed the retry offset, the adapter warns and
resumes at the next available record. Kafka provides no delivery-attempt
counter here: `Envelope.attempt` is always `Nothing`. Use an external store
for bounded retry counts, or halt processing.
A dead-letter acknowledgement advances the group's position without publishing
the message elsewhere. Arrange any required DLQ publication before returning
`AckDeadLetter`.
## Rebalances and shutdown
For assignment fencing, create state with `newKafkaAdapterState`, add
`Kafka.Consumer.setCallback (Kafka.Consumer.rebalanceCallback (kafkaRebalanceHandler state))`
to the consumer properties before consumer creation, and construct the adapter
with `kafkaAdapterWith state config`.
The callback logs assignment changes, clears retry state on revocation, and
fences old delivery callbacks so they cannot store, seek, or pause a later
assignment. Deliveries revoked before their admission check are skipped. A
revocation after that check cannot cancel a handler already admitted. The
helper is optional; `kafkaAdapter` alone does not provide this ownership fence.
Keep the consumer scope open while stopping the Shibuya application. Adapter
shutdown signals the source to stop and commits offsets already stored. Let
`runKafkaConsumer` close normally afterward so it can flush offsets stored
during draining.
After `AckHalt`, polling stops. The consumer can lose its assignment when
`max.poll.interval.ms` expires; a single-member group needs a restart to resume.
Partition pause state belongs to the consumer session.
## Errors and tracing
Non-fatal poll errors, including timeouts and partition EOFs, are filtered out.
Fatal poll errors terminate the source through `Error KafkaError`. Exhausted
acknowledgement operations throw `KafkaAcknowledgementException` at the
finalizer boundary so Shibuya can retain the processor failure.
The adapter preserves Kafka headers, extracts W3C `traceparent` and `tracestate`,
and adds `messaging.system=kafka`, `messaging.kafka.destination.partition`, and
`messaging.kafka.message.offset` to the envelope. Shibuya's runner uses these
attributes on its per-message span. The example above disables tracing with
`runTracingNoop`; use the configured tracing interpreter to export spans.
## Trace correlation
When a record carries a W3C `traceparent` header, Shibuya opens a Consumer span
named `process <processor>` around the handler and always adds a *link* to the
producer's span. The `telemetry` field of `KafkaAdapterConfig` decides whether
that span is also the producer's child:
```haskell
import Shibuya.Adapter.Kafka
-- Default (ParentAndLink): the handler runs inside the producer's trace.
config = defaultConfig [TopicName "orders"]
-- LinkOnly: each record's processing starts its own trace, linked to the producer.
replaying =
(defaultConfig [TopicName "orders"])
{ telemetry = defaultAdapterTelemetry {traceCorrelation = LinkOnly}
}
```
`ParentAndLink` suits records consumed close to real time: producer and
consumer appear in one connected trace. Choose `LinkOnly` when the producer's
trace has usually ended before processing starts, so the consumer span would
only stretch it:
- replay from an old offset, or a reprocessing job over retained data;
- sustained consumer lag;
- many consumer groups on one topic, each of which would otherwise add its
processing to the same producer trace.
A `LinkOnly` trace is sampled on its own, independently of the producer's
sampling decision, and a tail-sampling collector sees it as a new trace. The
link still lets a trace backend navigate to the producer.
OpenTelemetry's messaging conventions allow creation-context parenting for a
single message and recommend against it by default only when processing runs
inside another span; Shibuya records its choice in its ADR 0014
(`mori://shinzui/shibuya`, `docs/adr/0014-define-consumer-correlation-and-bounded-batch-link-accounting.md`;
artifact URI pending). Without a usable `traceparent` neither setting adds a
link, and the span keeps whatever span is active on the processing thread as
its parent.
Batch processors ignore the setting: a batch span is always a root with one
link per record (the first 128 records with a valid context), each link
carrying the record's `messaging.message.id`, and
`messaging.batch.message_count` on the span.
## Examples and development
See the [repository README](https://github.com/shinzui/shibuya-kafka-adapter#development)
for runnable examples, the Nix toolchain, and live Kafka integration tests. Only
the adapter library is published to Hackage.
## License
MIT. See [LICENSE](LICENSE).