kafka-effectful 0.1.0.0 → 0.2.0.0

raw patch · 23 files changed

+2412/−11 lines, 23 filesdep +case-insensitivedep +hs-opentelemetry-apidep +hs-opentelemetry-exporter-in-memorynew-component:exe:example-otel-tracingnew-component:exe:example-sync-publishnew-component:exe:example-transactional-etlPVP ok

version bump matches the API change (PVP)

Dependencies added: case-insensitive, hs-opentelemetry-api, hs-opentelemetry-exporter-in-memory, hs-opentelemetry-exporter-otlp, hs-opentelemetry-sdk, hs-opentelemetry-semantic-conventions, http-types, kafka-effectful, tasty, tasty-hunit, unliftio-core, unordered-containers

API changes (from Hackage documentation)

Files

• CHANGELOG.md +50/−0
• README.md +228/−0
• examples/OtelTracing.hs +223/−0
• examples/SyncPublish.hs +51/−0
• examples/TransactionalEtl.hs +118/−0
• kafka-effectful.cabal +145/−6
• src/Kafka/Effectful.hs +26/−0
• src/Kafka/Effectful/Consumer.hs +4/−1
• src/Kafka/Effectful/Consumer/Effect.hs +21/−0
• src/Kafka/Effectful/Consumer/Interpreter.hs +1/−0
• src/Kafka/Effectful/OpenTelemetry.hs +56/−0
• src/Kafka/Effectful/OpenTelemetry/Consumer/Interpreter.hs +211/−0
• src/Kafka/Effectful/OpenTelemetry/Producer/Interpreter.hs +191/−0
• src/Kafka/Effectful/OpenTelemetry/Propagation.hs +92/−0
• src/Kafka/Effectful/OpenTelemetry/Semantic.hs +199/−0
• src/Kafka/Effectful/Producer.hs +39/−1
• src/Kafka/Effectful/Producer/Effect.hs +190/−3
• src/Kafka/Effectful/Producer/Interpreter.hs +45/−0
• src/Kafka/Effectful/Producer/Transaction.hs +42/−0
• test/Kafka/Effectful/OpenTelemetry/PropagationTest.hs +187/−0
• test/Kafka/Effectful/OpenTelemetry/SemanticTest.hs +171/−0
• test/Kafka/Effectful/OpenTelemetry/ShibuyaCompatibilityTest.hs +104/−0
• test/Main.hs +18/−0

@@ -4,6 +4,56 @@  This package follows the [Haskell Package Versioning Policy](https://pvp.haskell.org/). +## 0.2.0.0 — 2026-05-06++Additive release. No breaking changes to existing modules.++- Add `produceMessage'` and `produceMessageSync` to the+  `KafkaProducer` effect. `produceMessage'` mirrors+  `Kafka.Producer.produceMessage'`, taking a per-message+  `DeliveryReport -> IO ()` callback. `produceMessageSync` blocks+  until the broker acknowledges the record and returns the assigned+  `Offset`. Both throw `KafkaError` via the `Error` effect on+  failure.+- Add `produceMessageBatch` to the `KafkaProducer` effect. Returns+  only the records that failed to enqueue, paired with their+  `KafkaError`. The interpreter inlines the upstream definition+  (`mapM` over the list) because Hackage `hw-kafka-client-5.3.0`+  does not re-export `Kafka.Producer.produceMessageBatch`.+- Add the transaction API to the `KafkaProducer` effect —+  `initTransactions`, `beginTransaction`, `commitTransaction`,+  `abortTransaction` — plus the cross-effect helper+  `commitOffsetMessageTransaction` (in new module+  `Kafka.Effectful.Producer.Transaction`) that commits consumer+  offsets as part of the producer's open transaction. Re-exports+  `TxError` with its three accessors+  (`kafkaErrorTxnRequiresAbort`, `kafkaErrorIsRetriable`,+  `kafkaErrorIsFatal`).+- Add narrow handle-ask escape hatches `askProducerHandle` and+  `askConsumerHandle` on the scoped facades. Reachable only from+  `Kafka.Effectful.Producer` and `Kafka.Effectful.Consumer`; not+  re-exported from the combined `Kafka.Effectful` facade.+- Add OpenTelemetry tracing support via opt-in interpreter variants+  `runKafkaProducerTraced` and `runKafkaConsumerTraced`. New modules+  under `Kafka.Effectful.OpenTelemetry.*` provide the+  attribute-builder helpers (`producerRecordAttributes`,+  `consumerRecordAttributes`) and the W3C trace-context header+  bridges (`extractTraceContextFromRecord`,+  `injectTraceContextIntoRecord`). The default interpreters+  `runKafkaProducer` and `runKafkaConsumer` are unchanged and remain+  zero-cost for users who do not want tracing. The attribute keys+  and value types match what `shibuya-kafka-adapter` already emits,+  so layering the two remains compatible.+- Add `kafka-effectful-test` test suite covering the OpenTelemetry+  attribute helpers and trace-context propagation bridges.+- Add example projects: `example-sync-publish`,+  `example-transactional-etl`, and `example-otel-tracing`+  demonstrating an end-to-end traced producer/consumer pipeline.+- Expand the README with a "Producer scenarios" walkthrough covering+  all eight best-practice cases from upstream's+  `producer-best-practices.md`, and document the new tracing+  support.+ ## 0.1.0.0 — 2026-04-16  Initial release.

@@ -27,6 +27,158 @@     flushProducer ``` +### Producer scenarios++The eight scenarios below mirror the+[producer-best-practices](https://github.com/haskell-works/hw-kafka-client/blob/master/docs/producer-best-practices.md)+guide from `hw-kafka-client`. Each snippet uses only symbols exported+from `Kafka.Effectful`; when a snippet needs `produceMessage'` or+`askProducerHandle`, it also imports `Kafka.Effectful.Producer`.++##### Scenario 1 — Fire-and-forget++Register a global delivery callback via `ProducerProperties` and+enqueue without blocking. `runKafkaProducer` flushes on scope exit.++```haskell+fireAndForgetProps =+  brokersList ["localhost:9092"]+  <> setCallback (deliveryCallback logFailure)++runKafkaProducer fireAndForgetProps $+  forM_ events (produceMessage . toRecord)+```++##### Scenario 2 — Synchronous delivery confirmation++`produceMessageSync` allocates an `MVar`, flushes the producer, and+returns the broker-assigned `Offset` — throwing `KafkaError` on any+failure.++```haskell+runKafkaProducer producerProps $ do+  offset <- produceMessageSync record+  liftIO $ putStrLn ("stored at offset " <> show offset)+```++##### Scenario 3 — Idempotent producer++Configuration only — no new call site is needed. Safe to enable by+default; the broker deduplicates retries by `(producer-id, sequence)`.++```haskell+idempotentProps =+  brokersList ["localhost:9092"]+  <> extraProp "enable.idempotence" "true"+  <> extraProp "acks" "all"+  <> extraProp "max.in.flight.requests.per.connection" "5"+```++##### Scenario 4 — High-throughput batching++Combine `produceMessageBatch` with `linger.ms`, `batch.size`, and+`compression` to trade a few milliseconds of latency for substantially+higher throughput. The result contains only records that failed to+enqueue.++```haskell+batchProps =+  brokersList ["localhost:9092"]+  <> compression Snappy+  <> extraProp "linger.ms" "10"+  <> extraProp "batch.size" "65536"++runKafkaProducer batchProps $ do+  failures <- produceMessageBatch records+  unless (null failures) $+    liftIO $ putStrLn ("enqueue failures: " <> show (length failures))+```++##### Scenario 5 — Transactional ETL++Consume, transform, produce, and commit consumer offsets — all inside+one producer transaction. `commitOffsetMessageTransaction` requires+both the `KafkaProducer` and `KafkaConsumer` effects. `TxError` must+be dispatched on in a fixed order: `kafkaErrorTxnRequiresAbort`,+`kafkaErrorIsRetriable`, `kafkaErrorIsFatal`.++```haskell+txProps =+  brokersList ["localhost:9092"]+  <> extraProp "transactional.id" "etl-1"+  <> extraProp "enable.idempotence" "true"+  <> extraProp "acks" "all"++etl = runKafkaProducer txProps $ runKafkaConsumer consumerProps sub $ do+  initTransactions (Timeout 10000)+  forever $ do+    msgs <- pollMessageBatch (Timeout 500) (BatchSize 100)+    let records = rights msgs+    unless (null records) $ do+      beginTransaction+      forM_ records (produceMessage . transform)+      forM_ (lastPerPartition records) $ \r ->+        commitOffsetMessageTransaction r (Timeout 5000)+          >>= handleTxResult+      commitTransaction (Timeout 5000) >>= handleTxResult++handleTxResult Nothing  = pure ()+handleTxResult (Just e)+  | kafkaErrorTxnRequiresAbort e = abortTransaction (Timeout 5000)+  | kafkaErrorIsRetriable e      = liftIO $ putStrLn "retry"+  | kafkaErrorIsFatal e          = throwError (getKafkaError e)+  | otherwise                    = liftIO $ putStrLn (show (getKafkaError e))+```++##### Scenario 6 — Keyed partitioning for ordering++Set `prKey` and leave `prPartition = UnassignedPartition` so the+default hash partitioner routes every record for the same key to the+same partition. Enable idempotence alongside, so a retry does not+reorder.++```haskell+orderedByKey userId event = ProducerRecord+  { prTopic     = TopicName "user-events"+  , prPartition = UnassignedPartition+  , prKey       = Just userId+  , prValue     = Just (encode event)+  , prHeaders   = mempty+  }+```++##### Scenario 7 — Custom partitioning and headers++Target a specific partition with `SpecifiedPartition` and attach+per-message metadata via `headersFromList`.++```haskell+shardedRecord (Shard n) payload = ProducerRecord+  { prTopic     = TopicName "sharded-events"+  , prPartition = SpecifiedPartition n+  , prKey       = Nothing+  , prValue     = Just payload+  , prHeaders   = headersFromList+      [ ("schema-version", "v3")+      , ("source",         "billing-api")+      ]+  }+```++##### Scenario 8 — Graceful shutdown++`runKafkaProducer` already brackets the handle — it flushes and closes+the producer on normal scope exit, so enqueued records drain before+the program continues. No explicit cleanup code is needed.++```haskell+main =+  runEff . runError @KafkaError $+    runKafkaProducer props $ do+      forM_ events (produceMessage . toRecord)+      -- no explicit flush needed; runKafkaProducer flushes on exit+```+ ### Consumer  ```haskell@@ -78,6 +230,76 @@ `ProducerRecord` values (see the `Kafka.Effectful.Producer` module for the available builders). +### OpenTelemetry tracing++Swap `runKafkaProducer` for `runKafkaProducerTraced tracer` and+`runKafkaConsumer` for `runKafkaConsumerTraced tracer` to add+distributed tracing without changing any effect-level code. The+traced interpreters open a `Producer`-kind span around every+record-sending operation and a `Consumer`-kind span around every+successful `pollMessage` / per-record success of `pollMessageBatch`,+populated with the OpenTelemetry messaging semantic conventions+(`messaging.system`, `messaging.destination.name`,+`messaging.operation`, `messaging.kafka.destination.partition`,+`messaging.kafka.message.offset`, `messaging.kafka.message.key`,+`messaging.kafka.consumer.group`). The current OTel context is+injected into the outgoing record's headers as W3C `traceparent` /+`tracestate` so downstream consumers can extract it and continue the+trace.++```haskell+{-# LANGUAGE TypeApplications #-}++import Effectful+import Effectful.Error.Static (runError)+import Kafka.Effectful+import Kafka.Effectful.OpenTelemetry (runKafkaProducerTraced)+import OpenTelemetry.Trace+  ( initializeGlobalTracerProvider+  , makeTracer+  , tracerOptions+  )++main :: IO ()+main = do+  tp <- initializeGlobalTracerProvider+  let tracer = makeTracer tp "my-app" tracerOptions+  result <- runEff . runError @KafkaError $+    runKafkaProducerTraced tracer producerProps $ do+      produceMessage record+      flushProducer+  case result of+    Left (_, err) -> putStrLn ("Kafka error: " <> show err)+    Right ()      -> pure ()+```++Span names are `"send <topic>"` and `"process <topic>"`. The default+interpreters (`runKafkaProducer`, `runKafkaConsumer`) are unchanged+and zero-cost for users who do not want tracing.++**Compatibility with `shibuya-kafka-adapter`.** The attribute keys+this library emits (`messaging.system`,+`messaging.kafka.destination.partition`,+`messaging.kafka.message.offset`) agree with what+`shibuya-kafka-adapter`'s envelope-level attributes already produce.+Layering the two yields a Receive→Process span split:+`kafka-effectful`'s poll span as parent, Shibuya's framework+per-message span as child. If you want only one span per message+instead of two, use either `kafka-effectful`'s traced runner *or*+Shibuya's framework span — not both.++The end-to-end demo in+`examples/OtelTracing.hs` produces a record through the traced+producer and reads it back through the traced consumer, printing+both trace IDs and asserting they match. Run it against a local+broker:++```+cabal run example-otel-tracing -f examples -- \+  --bootstrap-servers localhost:9092 \+  --topic otel-demo+```+ ## Module Structure  | Module | Description |@@ -87,8 +309,14 @@ | `Kafka.Effectful.Consumer` | Consumer effect, interpreter, and types | | `Kafka.Effectful.Producer.Effect` | `KafkaProducer` effect definition and operations | | `Kafka.Effectful.Producer.Interpreter` | `runKafkaProducer` interpreter |+| `Kafka.Effectful.Producer.Transaction` | Cross-effect `commitOffsetMessageTransaction` helper | | `Kafka.Effectful.Consumer.Effect` | `KafkaConsumer` effect definition and operations | | `Kafka.Effectful.Consumer.Interpreter` | `runKafkaConsumer` interpreter |+| `Kafka.Effectful.OpenTelemetry` | Single-import facade for the traced interpreters and helpers |+| `Kafka.Effectful.OpenTelemetry.Producer.Interpreter` | `runKafkaProducerTraced` |+| `Kafka.Effectful.OpenTelemetry.Consumer.Interpreter` | `runKafkaConsumerTraced` |+| `Kafka.Effectful.OpenTelemetry.Semantic` | Pure attribute-builder helpers |+| `Kafka.Effectful.OpenTelemetry.Propagation` | W3C trace-context header bridges |  ## Requirements 

@@ -0,0 +1,223 @@+{- | End-to-end OpenTelemetry tracing demo.++This program produces a single record through 'runKafkaProducerTraced'+and then consumes it through 'runKafkaConsumerTraced'. It captures the+producer-side trace ID (from an outer parent span we open so its trace+ID propagates to the produce span the interpreter opens) and the+consumer-side trace ID (from the inbound parent span the consumer+interpreter attaches to the thread context after extracting the W3C+@traceparent@ header), then compares the two. A successful run prints+two equal trace IDs and exits zero. A mismatch (or any kafka error)+exits non-zero.++Manual run procedure:++@+  cabal run example-otel-tracing -f examples -- \\+    --bootstrap-servers localhost:9092 \\+    --topic otel-demo+@++Expected output (trace IDs are 32-hex-char strings, regenerated each+run):++@+  [otel-tracing] producer trace id: 4bf92f3577b34da6a3ce929d0e0e4736+  [otel-tracing] consumer trace id: 4bf92f3577b34da6a3ce929d0e0e4736+  [otel-tracing] trace IDs match — context propagated through Kafka headers+@++If a Jaeger v2 instance is reachable at @http:\/\/localhost:4318@+(the default OTLP HTTP endpoint), the same trace appears with a+producer span as the parent and a child consumer span. Otherwise the+flow still works locally and the OTLP exporter silently drops the+spans.+-}+module Main (main) where++import Data.IORef (IORef, newIORef, readIORef, writeIORef)+import Data.Text (Text)+import Data.Text qualified as Text+import Effectful (Eff, IOE, runEff, (:>))+import Effectful qualified+import Effectful.Error.Static (runError)+import Kafka.Effectful+import Kafka.Effectful.Consumer qualified as KEC+import Kafka.Effectful.OpenTelemetry (+    runKafkaConsumerTraced,+    runKafkaProducerTraced,+ )+import Kafka.Effectful.Producer qualified as KEP+import OpenTelemetry.Attributes (emptyAttributes)+import OpenTelemetry.Context qualified as Context+import OpenTelemetry.Context.ThreadLocal (getContext)+import OpenTelemetry.Trace (+    initializeGlobalTracerProvider,+    makeTracer,+    shutdownTracerProvider,+    tracerOptions,+ )+import OpenTelemetry.Trace.Core (+    InstrumentationLibrary (..),+    SpanContext (traceId),+    Tracer,+    defaultSpanArguments,+    getSpanContext,+    inSpan'',+ )+import OpenTelemetry.Trace.Id (Base (Base16), traceIdBaseEncodedText)+import System.Environment (getArgs)+import System.Exit (exitFailure, exitSuccess)+import System.IO (hPutStrLn, stderr)++-- CLI++data Args = Args+    { bootstrapServers :: Text+    , topic :: Text+    }++parseArgs :: [String] -> Maybe Args+parseArgs = go (Args "" "")+  where+    go acc [] =+        if Text.null acc.bootstrapServers || Text.null acc.topic+            then Nothing+            else Just acc+    go acc ("--bootstrap-servers" : v : rest) =+        go acc{bootstrapServers = Text.pack v} rest+    go acc ("--topic" : v : rest) = go acc{topic = Text.pack v} rest+    go _ _ = Nothing++usage :: String+usage =+    "usage: example-otel-tracing --bootstrap-servers HOST:PORT --topic NAME"++-- Wiring++producerProps :: Text -> ProducerProperties+producerProps host =+    KEP.brokersList [BrokerAddress host]+        <> KEP.sendTimeout (Timeout 10000)+        <> KEP.extraProp "enable.idempotence" "true"+        <> KEP.extraProp "acks" "all"++consumerProps :: Text -> ConsumerProperties+consumerProps host =+    KEC.brokersList [BrokerAddress host]+        <> KEC.groupId (ConsumerGroupId "kafka-effectful-otel-demo")++mkRecord :: Text -> ProducerRecord+mkRecord t =+    ProducerRecord+        { prTopic = TopicName t+        , prPartition = UnassignedPartition+        , prKey = Just "k1"+        , prValue = Just "hello otel"+        , prHeaders = mempty+        }++instrumentation :: InstrumentationLibrary+instrumentation =+    InstrumentationLibrary+        { libraryName = "kafka-effectful-example"+        , libraryVersion = "0.2.0.0"+        , librarySchemaUrl = ""+        , libraryAttributes = emptyAttributes+        }++-- Capture the trace ID of whatever span is currently active on the+-- thread. Used on both sides: the producer wraps its work in an+-- outer 'inSpan''' so a parent span is active; the consumer+-- interpreter attaches the inbound parent span to the thread+-- context after extracting the @traceparent@ header.+captureCurrentTraceId :: (IOE :> es) => IORef (Maybe Text) -> Eff es ()+captureCurrentTraceId ref = Effectful.liftIO $ do+    ctx <- getContext+    case Context.lookupSpan ctx of+        Nothing -> pure ()+        Just s -> do+            sc <- getSpanContext s+            writeIORef ref (Just (traceIdBaseEncodedText Base16 (traceId sc)))++-- Poll up to 30 one-second timeouts looking for the record we just+-- produced. Captures the consumer-side trace ID after the first+-- successful poll: at that point the consumer interpreter has+-- attached the inbound parent span, so the active context's trace+-- ID equals what the producer published on the wire.+pollUntilRecord ::+    (KafkaConsumer :> es, IOE :> es) =>+    IORef (Maybe Text) ->+    Eff es ()+pollUntilRecord ref = loop (30 :: Int)+  where+    loop 0 = pure ()+    loop n = do+        mbRecord <- pollMessage (Timeout 1000)+        case mbRecord of+            Nothing -> loop (n - 1)+            Just _ -> captureCurrentTraceId ref++main :: IO ()+main = do+    rawArgs <- getArgs+    args <- case parseArgs rawArgs of+        Nothing -> hPutStrLn stderr usage *> exitFailure+        Just a -> pure a++    tp <- initializeGlobalTracerProvider+    let tracer :: Tracer+        tracer = makeTracer tp instrumentation tracerOptions++    producerTraceRef <- newIORef Nothing+    consumerTraceRef <- newIORef Nothing++    producerResult <-+        runEff . runError @KafkaError $+            runKafkaProducerTraced tracer (producerProps args.bootstrapServers) $+                inSpan'' tracer "publish" defaultSpanArguments $ \_parentSpan -> do+                    captureCurrentTraceId producerTraceRef+                    produceMessage (mkRecord args.topic)+                    flushProducer+    case producerResult of+        Left (_cs, err) -> do+            hPutStrLn stderr $ "producer error: " <> show err+            shutdownTracerProvider tp+            exitFailure+        Right () -> pure ()++    consumerResult <-+        runEff . runError @KafkaError $+            runKafkaConsumerTraced+                tracer+                (consumerProps args.bootstrapServers)+                (topics [TopicName args.topic] <> offsetReset Earliest)+                (pollUntilRecord consumerTraceRef)+    case consumerResult of+        Left (_cs, err) -> do+            hPutStrLn stderr $ "consumer error: " <> show err+            shutdownTracerProvider tp+            exitFailure+        Right () -> pure ()++    pid <- readIORef producerTraceRef+    cid <- readIORef consumerTraceRef+    shutdownTracerProvider tp+    case (pid, cid) of+        (Just p, Just c) -> do+            putStrLn $ "[otel-tracing] producer trace id: " <> Text.unpack p+            putStrLn $ "[otel-tracing] consumer trace id: " <> Text.unpack c+            if p == c+                then do+                    putStrLn+                        "[otel-tracing] trace IDs match — context propagated through Kafka headers"+                    exitSuccess+                else do+                    putStrLn "[otel-tracing] trace IDs differ"+                    exitFailure+        (Nothing, _) -> do+            hPutStrLn stderr "[otel-tracing] failed to capture producer trace id"+            exitFailure+        (_, Nothing) -> do+            hPutStrLn stderr "[otel-tracing] consumer never received the record (timed out)"+            exitFailure

@@ -0,0 +1,51 @@+{- | Scenario 2 of @producer-best-practices.md@: publish one record+synchronously and print the broker-assigned offset.++Assumes a Kafka broker at localhost:9092 with the topic+@kafka-effectful-sync-demo@ already created (or auto-create enabled).+Change the 'brokerHost' and 'topicName' bindings below if your local+broker lives somewhere else.+-}+module Main (main) where++import Effectful (runEff)+import Effectful.Error.Static (runError)+import Kafka.Effectful+import Kafka.Effectful.Producer (brokersList, extraProp, sendTimeout)+import System.Exit (exitFailure)+import System.IO (hPutStrLn, stderr)++brokerHost :: BrokerAddress+brokerHost = BrokerAddress "localhost:9092"++topicName :: TopicName+topicName = TopicName "kafka-effectful-sync-demo"++producerProps :: ProducerProperties+producerProps =+    brokersList [brokerHost]+        <> sendTimeout (Timeout 10000)+        <> extraProp "enable.idempotence" "true"+        <> extraProp "acks" "all"++record :: ProducerRecord+record =+    ProducerRecord+        { prTopic = topicName+        , prPartition = UnassignedPartition+        , prKey = Just "demo-key"+        , prValue = Just "hello from produceMessageSync"+        , prHeaders = mempty+        }++main :: IO ()+main = do+    result <- runEff . runError @KafkaError $+        runKafkaProducer producerProps $ do+            produceMessageSync record+    case result of+        Right (Offset offset) ->+            putStrLn $ "delivered offset " <> show offset+        Left (_callStack, err) -> do+            hPutStrLn stderr $ "kafka error: " <> show err+            exitFailure

@@ -0,0 +1,118 @@+{- | Scenario 5 of @producer-best-practices.md@: consume from one topic,+uppercase the value, and produce to another topic — all inside a+producer transaction, with consumer offsets committed as part of the+same transaction.++Assumes a Kafka broker at localhost:9092 with topics @source@ and+@destination@ already created (or auto-create enabled).++To exercise the exactly-once guarantee:++  1. seed records into @source@ (e.g. @kcat -P -b localhost:9092 -t source@)+  2. run this example; watch @destination@ (@kcat -C -e -b localhost:9092 -t destination@)+  3. kill -9 the process mid-batch, restart it, and confirm no duplicates.+-}+module Main (main) where++import Control.Monad (forever, unless)+import Data.ByteString (ByteString)+import Data.ByteString.Char8 qualified as BS8+import Data.Char (toUpper)+import Data.Either (rights)+import Data.Foldable (for_)+import Data.Map.Strict qualified as Map+import Effectful (Eff, IOE, liftIO, runEff, (:>))+import Effectful.Error.Static (Error, runError, throwError)+import Kafka.Effectful+import Kafka.Effectful.Consumer qualified as C+import Kafka.Effectful.Producer qualified as P+import System.Exit (exitFailure)+import System.IO (hPutStrLn, stderr)++brokerHost :: BrokerAddress+brokerHost = BrokerAddress "localhost:9092"++sourceTopic, destinationTopic :: TopicName+sourceTopic = TopicName "source"+destinationTopic = TopicName "destination"++producerProps :: ProducerProperties+producerProps =+    P.brokersList [brokerHost]+        <> P.sendTimeout (Timeout 30000)+        <> P.extraProp "transactional.id" "kafka-effectful-etl-1"+        <> P.extraProp "enable.idempotence" "true"+        <> P.extraProp "acks" "all"++consumerProps :: ConsumerProperties+consumerProps =+    C.brokersList [brokerHost]+        <> C.groupId (ConsumerGroupId "kafka-effectful-etl-group")+        <> C.noAutoCommit+        <> C.extraProp "isolation.level" "read_committed"++sourceSubscription :: Subscription+sourceSubscription = topics [sourceTopic] <> offsetReset Earliest++-- | Uppercase ASCII bytes in the value; drop anything else.+transform ::+    ConsumerRecord (Maybe ByteString) (Maybe ByteString) ->+    ProducerRecord+transform msg =+    ProducerRecord+        { prTopic = destinationTopic+        , prPartition = UnassignedPartition+        , prKey = crKey msg+        , prValue = fmap (BS8.map toUpper) (crValue msg)+        , prHeaders = mempty+        }++{- | Keep only the last record per source partition — that is the offset+to commit into the transaction.+-}+lastPerPartition ::+    [ConsumerRecord (Maybe ByteString) (Maybe ByteString)] ->+    [ConsumerRecord (Maybe ByteString) (Maybe ByteString)]+lastPerPartition =+    Map.elems . Map.fromList . fmap (\r -> ((crTopic r, crPartition r), r))++handleTxResult ::+    (KafkaProducer :> es, Error KafkaError :> es, IOE :> es) =>+    Maybe TxError ->+    Eff es ()+handleTxResult Nothing = pure ()+handleTxResult (Just err)+    | kafkaErrorTxnRequiresAbort err = abortTransaction (Timeout 5000)+    | kafkaErrorIsRetriable err =+        liftIO $ hPutStrLn stderr "retriable tx error — retry the whole transaction"+    | kafkaErrorIsFatal err = throwError (getKafkaError err)+    | otherwise =+        liftIO $ hPutStrLn stderr $ "tx error: " <> show (getKafkaError err)++etlLoop ::+    (KafkaProducer :> es, KafkaConsumer :> es, Error KafkaError :> es, IOE :> es) =>+    Eff es ()+etlLoop = do+    initTransactions (Timeout 10000)+    forever $ do+        msgs <- pollMessageBatch (Timeout 500) (BatchSize 100)+        let records = rights msgs+        unless (null records) $ do+            beginTransaction+            for_ records (produceMessage . transform)+            for_ (lastPerPartition records) $ \r ->+                commitOffsetMessageTransaction r (Timeout 5000)+                    >>= handleTxResult+            commitTransaction (Timeout 5000) >>= handleTxResult++main :: IO ()+main = do+    result <-+        runEff . runError @KafkaError $+            runKafkaProducer producerProps $+                runKafkaConsumer consumerProps sourceSubscription etlLoop+    case result of+        Right () -> pure ()+        Left (_callStack, err) -> do+            hPutStrLn stderr $ "kafka error: " <> show err+            exitFailure

@@ -1,6 +1,6 @@ cabal-version:   3.4 name:            kafka-effectful-version:         0.1.0.0+version:         0.2.0.0 synopsis:        Effectful effects for hw-kafka-client description:   Effectful effects and interpreters for hw-kafka-client, a Haskell@@ -21,6 +21,11 @@   type:     git   location: https://github.com/shinzui/kafka-effectful.git +flag examples+  description: Build example executables+  manual:      True+  default:     False+ common warnings   ghc-options:     -Wall -Wcompat -Widentities -Wincomplete-uni-patterns@@ -35,29 +40,163 @@     Kafka.Effectful.Consumer     Kafka.Effectful.Consumer.Effect     Kafka.Effectful.Consumer.Interpreter+    Kafka.Effectful.OpenTelemetry+    Kafka.Effectful.OpenTelemetry.Consumer.Interpreter+    Kafka.Effectful.OpenTelemetry.Producer.Interpreter+    Kafka.Effectful.OpenTelemetry.Propagation+    Kafka.Effectful.OpenTelemetry.Semantic     Kafka.Effectful.Producer     Kafka.Effectful.Producer.Effect     Kafka.Effectful.Producer.Interpreter+    Kafka.Effectful.Producer.Transaction    build-depends:+    , base                                   >=4.21 && <5+    , bytestring                             >=0.11 && <0.13+    , case-insensitive                       >=1.2  && <1.3+    , containers                             >=0.6  && <0.8+    , effectful-core                         >=2.5  && <2.7+    , hs-opentelemetry-api                   ^>=0.3+    , hs-opentelemetry-semantic-conventions  ^>=0.1+    , http-types                             >=0.12 && <0.13+    , hw-kafka-client                        >=5.3  && <6+    , text                                   >=2.0  && <2.2+    , unliftio-core                          >=0.2  && <0.3+    , unordered-containers                   >=0.2  && <0.3++  hs-source-dirs:     src+  default-language:   GHC2024+  default-extensions:+    DataKinds+    DuplicateRecordFields+    GADTs+    GeneralisedNewtypeDeriving+    ImportQualifiedPost+    LambdaCase+    NoFieldSelectors+    OverloadedRecordDot+    OverloadedStrings+    TypeFamilies+    TypeOperators++executable example-sync-publish+  import:             warnings+  main-is:            SyncPublish.hs+  hs-source-dirs:     examples+  default-language:   GHC2024+  default-extensions:+    ImportQualifiedPost+    LambdaCase+    OverloadedStrings+    TypeApplications++  if !flag(examples)+    buildable: False++  build-depends:     , base             >=4.21 && <5     , bytestring       >=0.11 && <0.13+    , effectful-core   >=2.5  && <2.7+    , hw-kafka-client  >=5.3  && <6+    , kafka-effectful++executable example-transactional-etl+  import:             warnings+  main-is:            TransactionalEtl.hs+  hs-source-dirs:     examples+  default-language:   GHC2024+  default-extensions:+    ImportQualifiedPost+    LambdaCase+    OverloadedStrings+    TypeApplications++  if !flag(examples)+    buildable: False++  build-depends:+    , base             >=4.21 && <5+    , bytestring       >=0.11 && <0.13     , containers       >=0.6  && <0.8     , effectful-core   >=2.5  && <2.7     , hw-kafka-client  >=5.3  && <6-    , text             >=2.0  && <2.2+    , kafka-effectful -  hs-source-dirs:     src+executable example-otel-tracing+  import:             warnings+  main-is:            OtelTracing.hs+  hs-source-dirs:     examples++  -- The hs-opentelemetry batch span processor requires the threaded+  -- runtime; without it, initializeGlobalTracerProvider raises at+  -- runtime.+  ghc-options:        -threaded   default-language:   GHC2024   default-extensions:+    DuplicateRecordFields+    ImportQualifiedPost+    LambdaCase+    NoFieldSelectors+    OverloadedRecordDot+    OverloadedStrings+    TypeApplications++  if !flag(examples)+    buildable: False++  build-depends:+    , base                            >=4.21 && <5+    , bytestring                      >=0.11 && <0.13+    , effectful-core                  >=2.5  && <2.7+    , hs-opentelemetry-api            ^>=0.3+    , hs-opentelemetry-exporter-otlp  ^>=0.1+    , hs-opentelemetry-sdk            ^>=0.1+    , hw-kafka-client                 >=5.3  && <6+    , kafka-effectful+    , text                            >=2.0  && <2.2++test-suite kafka-effectful-test+  import:             warnings+  type:               exitcode-stdio-1.0+  main-is:            Main.hs+  hs-source-dirs:     test++  -- The hs-opentelemetry batch span processor requires the threaded+  -- runtime. Without -threaded, even initializing the global tracer+  -- provider in PropagationTest's withResource block raises an+  -- exception at runtime.+  ghc-options:        -threaded+  default-language:   GHC2024+  default-extensions:     DataKinds     DuplicateRecordFields     GADTs-    GeneralisedNewtypeDeriving     ImportQualifiedPost     LambdaCase-    NoFieldSelectors     OverloadedRecordDot     OverloadedStrings-    TypeFamilies+    TypeApplications     TypeOperators++  other-modules:+    Kafka.Effectful.OpenTelemetry.PropagationTest+    Kafka.Effectful.OpenTelemetry.SemanticTest+    Kafka.Effectful.OpenTelemetry.ShibuyaCompatibilityTest++  build-depends:+    , base                                   >=4.21   && <5+    , bytestring                             >=0.11   && <0.13+    , case-insensitive                       >=1.2    && <1.3+    , containers                             >=0.6    && <0.8+    , effectful-core                         >=2.5    && <2.7+    , hs-opentelemetry-api                   ^>=0.3+    , hs-opentelemetry-exporter-in-memory    ^>=0.0.1+    , hs-opentelemetry-sdk                   ^>=0.1+    , hs-opentelemetry-semantic-conventions  ^>=0.1+    , http-types                             >=0.12   && <0.13+    , hw-kafka-client                        >=5.3    && <6+    , kafka-effectful+    , tasty                                  ^>=1.5+    , tasty-hunit                            ^>=0.10+    , text                                   >=2.0    && <2.2+    , unordered-containers                   >=0.2    && <0.3

@@ -3,8 +3,22 @@     KafkaProducer,     runKafkaProducer,     produceMessage,+    produceMessageSync,+    produceMessageBatch,     flushProducer, +    -- ** Transactions+    initTransactions,+    beginTransaction,+    commitTransaction,+    abortTransaction,+    commitOffsetMessageTransaction,+    TxError,+    getKafkaError,+    kafkaErrorIsFatal,+    kafkaErrorIsRetriable,+    kafkaErrorTxnRequiresAbort,+     -- * Consumer Effect     KafkaConsumer,     runKafkaConsumer,@@ -119,8 +133,20 @@     ProducePartition (..),     ProducerProperties (..),     ProducerRecord (..),+    TxError,+    abortTransaction,+    beginTransaction,+    commitOffsetMessageTransaction,+    commitTransaction,     flushProducer,+    getKafkaError,+    initTransactions,+    kafkaErrorIsFatal,+    kafkaErrorIsRetriable,+    kafkaErrorTxnRequiresAbort,     produceMessage,+    produceMessageBatch,+    produceMessageSync,     runKafkaProducer,  ) import Kafka.Types (

@@ -28,6 +28,9 @@     assignment,     subscription, +    -- * Raw handle escape hatch+    askConsumerHandle,+     -- * Consumer Types     ConsumerRecord (..),     Offset (..),@@ -96,6 +99,6 @@ import Kafka.Consumer.ConsumerProperties qualified as K import Kafka.Consumer.Subscription (Subscription (..), extraSubscriptionProps, offsetReset, topics) import Kafka.Consumer.Types (ConsumerGroupId (..), ConsumerRecord (..), Offset (..), OffsetCommit (..), OffsetReset (..), PartitionOffset (..), RebalanceEvent (..), SubscribedPartitions (..), Timestamp (..), TopicPartition (..))-import Kafka.Effectful.Consumer.Effect (KafkaConsumer, assign, assignment, commitAllOffsets, commitOffsetMessage, commitPartitionsOffsets, committed, pausePartitions, pollMessage, pollMessageBatch, position, resumePartitions, seekPartitions, storeOffsetMessage, storeOffsets, subscription)+import Kafka.Effectful.Consumer.Effect (KafkaConsumer, askConsumerHandle, assign, assignment, commitAllOffsets, commitOffsetMessage, commitPartitionsOffsets, committed, pausePartitions, pollMessage, pollMessageBatch, position, resumePartitions, seekPartitions, storeOffsetMessage, storeOffsets, subscription) import Kafka.Effectful.Consumer.Interpreter (runKafkaConsumer) import Kafka.Types (BatchSize (..), BrokerAddress (..), ClientId (..), Headers, KafkaCompressionCodec (..), KafkaDebug (..), KafkaError (..), KafkaLogLevel (..), Millis (..), PartitionId (..), Timeout (..), TopicName (..), headersFromList, headersToList)

@@ -24,6 +24,9 @@     position,     assignment,     subscription,++    -- * Internal — cross-effect plumbing+    askConsumerHandle, ) where @@ -37,6 +40,7 @@     SubscribedPartitions,     TopicPartition,  )+import Kafka.Consumer.Types qualified as KC import Kafka.Types (     BatchSize,     KafkaError,@@ -95,6 +99,8 @@         KafkaConsumer m (Map TopicName [PartitionId])     Subscription ::         KafkaConsumer m [(TopicName, SubscribedPartitions)]+    AskConsumerHandle ::+        KafkaConsumer m KC.KafkaConsumer  type instance DispatchOf KafkaConsumer = 'Dynamic @@ -186,3 +192,18 @@ -- | Get the current topic subscription. subscription :: (KafkaConsumer :> es) => Eff es [(TopicName, SubscribedPartitions)] subscription = send Subscription++{- | Escape hatch: return the raw @Kafka.Consumer.KafkaConsumer@ handle+acquired by @runKafkaConsumer@.++Exposed to enable the cross-effect+'Kafka.Effectful.Producer.Transaction.commitOffsetMessageTransaction'+helper, which must reach both the producer and consumer handles to+call the underlying transactional offset-commit primitive. New+operations should go through the 'KafkaConsumer' effect rather than+this handle.++@since 0.2.0.0+-}+askConsumerHandle :: (KafkaConsumer :> es) => Eff es KC.KafkaConsumer+askConsumerHandle = send AskConsumerHandle

@@ -87,6 +87,7 @@     Position parts -> throwOnLeft $ K.position consumer parts     Assignment -> throwOnLeft $ K.assignment consumer     Subscription -> throwOnLeft $ K.subscription consumer+    AskConsumerHandle -> pure consumer   where     throwOnJust action' = do         mbErr <- Effectful.liftIO action'

@@ -0,0 +1,56 @@+{- | Single-import facade for the OpenTelemetry-aware variants of+@kafka-effectful@\'s producer and consumer interpreters, plus the+pure attribute-builder helpers and W3C trace-context propagation+helpers that the traced interpreters are built from.++Typical wiring:++> import Kafka.Effectful.OpenTelemetry+> import OpenTelemetry.Trace (initializeGlobalTracerProvider, makeTracer, tracerOptions)+>+> tracer <- do+>   tp <- initializeGlobalTracerProvider+>   pure (makeTracer tp \"my-app\" tracerOptions)+>+> runEff . runError . runKafkaProducerTraced tracer producerProps $ do+>   produceMessage record++The facade does not re-export the upstream @Tracer@, @Span@, etc.+types; users who need those import them from @OpenTelemetry.Trace@+or @OpenTelemetry.Trace.Core@ directly.++@since 0.2.0.0+-}+module Kafka.Effectful.OpenTelemetry (+    -- * Traced interpreters+    runKafkaProducerTraced,+    runKafkaConsumerTraced,++    -- * Attribute helpers+    producerRecordAttributes,+    consumerRecordAttributes,+    producerSpanName,+    consumerSpanName,++    -- * Trace-context propagation+    extractTraceContextFromRecord,+    injectTraceContextIntoRecord,+    kafkaHeadersToRequestHeaders,+    requestHeadersToKafkaHeaders,+)+where++import Kafka.Effectful.OpenTelemetry.Consumer.Interpreter (runKafkaConsumerTraced)+import Kafka.Effectful.OpenTelemetry.Producer.Interpreter (runKafkaProducerTraced)+import Kafka.Effectful.OpenTelemetry.Propagation (+    extractTraceContextFromRecord,+    injectTraceContextIntoRecord,+    kafkaHeadersToRequestHeaders,+    requestHeadersToKafkaHeaders,+ )+import Kafka.Effectful.OpenTelemetry.Semantic (+    consumerRecordAttributes,+    consumerSpanName,+    producerRecordAttributes,+    producerSpanName,+ )

@@ -0,0 +1,211 @@+-- The 'EffectHandler' type synonym in effectful-core expands to a+-- constraint that GHC's redundant-constraint check flags on the+-- handler's signature, even though the constraint is required for+-- 'interpret' to type-check. Suppress the warning at the file level.+{-# OPTIONS_GHC -Wno-redundant-constraints #-}++{- | OpenTelemetry-traced interpreter for the 'KafkaConsumer' effect.++Drop-in alternative to 'Kafka.Effectful.Consumer.Interpreter.runKafkaConsumer'+that opens a Consumer-kind span on every successful record return from+'pollMessage' \/ 'pollMessageBatch', rooted at the W3C trace context+extracted from the record\'s Kafka headers (or as a new root span when+no inbound context is present). Polls that return @Nothing@ on timeout+do not open a span, preserving the existing timeout-returns-@Nothing@+semantics.++Non-polling operations (offset commit, partition assignment, etc.) are+passed through unchanged.++The design parallels the upstream+@hs-opentelemetry-instrumentation-hw-kafka-client@\'s+@OpenTelemetry.Instrumentation.Kafka.pollMessage@.++@since 0.2.0.0+-}+module Kafka.Effectful.OpenTelemetry.Consumer.Interpreter (+    -- * Interpreter+    runKafkaConsumerTraced,+)+where++import Control.Monad (void)+import Data.ByteString (ByteString)+import Data.Foldable (for_)+import Data.Map.Strict qualified as Map+import Effectful (Eff, IOE, (:>))+import Effectful qualified+import Effectful.Dispatch.Dynamic (EffectHandler, interpret)+import Effectful.Error.Static (Error, throwError)+import Effectful.Exception (ExitCase (..))+import Effectful.Exception qualified as Exception+import Kafka.Consumer (RdKafkaRespErrT (..))+import Kafka.Consumer qualified as K+import Kafka.Consumer.ConsumerProperties (ConsumerProperties (cpProps))+import Kafka.Consumer.Subscription (Subscription)+import Kafka.Consumer.Types (ConsumerRecord (crTopic))+import Kafka.Effectful.Consumer.Effect (KafkaConsumer (..))+import Kafka.Effectful.OpenTelemetry.Propagation (+    extractTraceContextFromRecord,+ )+import Kafka.Effectful.OpenTelemetry.Semantic (+    consumerRecordAttributes,+    consumerSpanName,+ )+import Kafka.Types (KafkaError (..))+import OpenTelemetry.Attributes.Attribute (toAttribute)+import OpenTelemetry.Attributes.Map (AttributeMap, insertAttributeByKey)+import OpenTelemetry.Context.ThreadLocal (attachContext, getContext)+import OpenTelemetry.SemanticConventions (messaging_kafka_consumer_group)+import OpenTelemetry.Trace.Core (+    SpanArguments (kind),+    SpanKind (Consumer),+    Tracer,+    addAttributesToSpanArguments,+    defaultSpanArguments,+    inSpan'',+ )++{- | Run the 'KafkaConsumer' effect with OpenTelemetry tracing.++Identical in shape to 'Kafka.Effectful.Consumer.Interpreter.runKafkaConsumer',+plus an additional 'Tracer' argument. On every successful record return+from 'pollMessage' \/ 'pollMessageBatch' the interpreter:++* extracts a W3C trace context from the record\'s headers and+  attaches it as the current thread\'s 'OpenTelemetry.Context.Context';+* opens a Consumer-kind span named @\"process \<topic\>\"@,+  populated with the spec-aligned @messaging.*@ attribute set plus+  the @messaging.kafka.consumer.group@ attribute (read from the+  @group.id@ entry of the supplied 'ConsumerProperties').++Polls that time out (return @Nothing@) do not open a span. Per-record+errors in batch polls (@Left err@ entries) are kept in place in the+returned list and do not get spans either. The consumer handle is+acquired and released via 'Exception.generalBracket', exactly as+'runKafkaConsumer' does.++@since 0.2.0.0+-}+runKafkaConsumerTraced ::+    (IOE :> es, Error KafkaError :> es) =>+    Tracer ->+    ConsumerProperties ->+    Subscription ->+    Eff (KafkaConsumer : es) a ->+    Eff es a+runKafkaConsumerTraced tracer props sub action =+    fst+        <$> Exception.generalBracket+            acquire+            release+            ( \consumer ->+                interpret (handleTracedConsumer tracer props consumer) action+            )+  where+    acquire = do+        result <- Effectful.liftIO $ K.newConsumer props sub+        case result of+            Left err -> throwError err+            Right consumer -> pure consumer++    release consumer = \case+        ExitCaseSuccess _ -> do+            mbErr <- Effectful.liftIO $ K.closeConsumer consumer+            for_ mbErr throwError+        ExitCaseException _ ->+            Effectful.liftIO . void $ K.closeConsumer consumer+        ExitCaseAbort ->+            Effectful.liftIO . void $ K.closeConsumer consumer++handleTracedConsumer ::+    (IOE :> es, Error KafkaError :> es) =>+    Tracer ->+    ConsumerProperties ->+    K.KafkaConsumer ->+    EffectHandler KafkaConsumer es+handleTracedConsumer tracer props consumer _env = \case+    PollMessage timeout -> do+        result <- Effectful.liftIO $ K.pollMessage consumer timeout+        case result of+            Left (KafkaResponseError RdKafkaRespErrTimedOut) -> pure Nothing+            Left err -> throwError err+            Right cr -> Just <$> withConsumerSpan tracer props cr (pure cr)+    PollMessageBatch timeout batchSize -> do+        results <-+            Effectful.liftIO $+                K.pollMessageBatch consumer timeout batchSize+        traverse openSpanForResult results+      where+        openSpanForResult (Left err) = pure (Left err)+        openSpanForResult (Right cr) =+            Right <$> withConsumerSpan tracer props cr (pure cr)+    CommitOffsetMessage oc cr -> throwOnJust $ K.commitOffsetMessage oc consumer cr+    CommitAllOffsets oc -> throwOnJust $ K.commitAllOffsets oc consumer+    CommitPartitionsOffsets oc tps -> throwOnJust $ K.commitPartitionsOffsets oc consumer tps+    StoreOffsets tps -> throwOnJust $ K.storeOffsets consumer tps+    StoreOffsetMessage cr -> throwOnJust $ K.storeOffsetMessage consumer cr+    Assign tps -> throwOnJust $ K.assign consumer tps+    PausePartitions parts ->+        throwOnKafkaErr (K.pausePartitions consumer parts)+    ResumePartitions parts ->+        throwOnKafkaErr (K.resumePartitions consumer parts)+    SeekPartitions tps timeout -> throwOnJust $ K.seekPartitions consumer tps timeout+    Committed timeout parts -> throwOnLeft $ K.committed consumer timeout parts+    Position parts -> throwOnLeft $ K.position consumer parts+    Assignment -> throwOnLeft $ K.assignment consumer+    Subscription -> throwOnLeft $ K.subscription consumer+    AskConsumerHandle -> pure consumer+  where+    throwOnJust action' = do+        mbErr <- Effectful.liftIO action'+        for_ mbErr throwError++    throwOnLeft action' = do+        result <- Effectful.liftIO action'+        case result of+            Left err -> throwError err+            Right a -> pure a++    throwOnKafkaErr action' = do+        err <- Effectful.liftIO action'+        case err of+            KafkaResponseError RdKafkaRespErrNoError -> pure ()+            _ -> throwError err++{- | Open a Consumer-kind span around an action that processes a single+record.++Extracts the W3C trace context from the record\'s headers, attaches+it as the current thread context, then opens a span named+@\"process \<topic\>\"@ populated with the @messaging.*@ attribute+set (including @messaging.kafka.consumer.group@ when known).+-}+withConsumerSpan ::+    (IOE :> es) =>+    Tracer ->+    ConsumerProperties ->+    ConsumerRecord (Maybe ByteString) (Maybe ByteString) ->+    Eff es a ->+    Eff es a+withConsumerSpan tracer props cr action = do+    inboundCtx <- Effectful.liftIO $ do+        currentCtx <- getContext+        extractTraceContextFromRecord cr currentCtx+    void $ attachContext inboundCtx+    inSpan'' tracer (consumerSpanName (crTopic cr)) spanArgs $ \_span -> action+  where+    spanArgs =+        addAttributesToSpanArguments+            (withConsumerGroup (consumerRecordAttributes cr))+            defaultSpanArguments{kind = Consumer}++    withConsumerGroup :: AttributeMap -> AttributeMap+    withConsumerGroup attrs =+        case Map.lookup "group.id" (cpProps props) of+            Just groupId ->+                insertAttributeByKey+                    messaging_kafka_consumer_group+                    (toAttribute groupId)+                    attrs+            Nothing -> attrs

@@ -0,0 +1,191 @@+-- The 'EffectHandler' type synonym in effectful-core expands to a+-- constraint that GHC's redundant-constraint check flags on the+-- handler's signature, even though the constraint is required for+-- 'interpret' to type-check. Suppress the warning at the file level.+{-# OPTIONS_GHC -Wno-redundant-constraints #-}++{- | OpenTelemetry-traced interpreter for the 'KafkaProducer' effect.++Drop-in alternative to 'Kafka.Effectful.Producer.Interpreter.runKafkaProducer'+that opens a Producer-kind span around every record-sending operation+('produceMessage', 'produceMessage'', 'produceMessageSync',+'produceMessageBatch'), populates the span with the spec-aligned+@messaging.*@ attribute set, and injects the current OTel context as+W3C @traceparent@\/@tracestate@ headers on the record before handing+it off to the underlying @hw-kafka-client@ produce call.++Non-sending operations (flush, transactional begin\/commit\/abort, etc.)+are passed through unchanged — they do not represent message sends and+therefore do not get a span.++The design parallels the upstream+@hs-opentelemetry-instrumentation-hw-kafka-client@\'s+@OpenTelemetry.Instrumentation.Kafka.produceMessage@.++@since 0.2.0.0+-}+module Kafka.Effectful.OpenTelemetry.Producer.Interpreter (+    -- * Interpreter+    runKafkaProducerTraced,+)+where++import Control.Concurrent.MVar qualified as Concurrent+import Data.Foldable (for_)+import Effectful (Eff, IOE, (:>))+import Effectful qualified+import Effectful.Dispatch.Dynamic (EffectHandler, interpret)+import Effectful.Error.Static (Error, throwError)+import Effectful.Exception qualified as Exception+import Kafka.Effectful.OpenTelemetry.Propagation (+    injectTraceContextIntoRecord,+ )+import Kafka.Effectful.OpenTelemetry.Semantic (+    producerRecordAttributes,+    producerSpanName,+ )+import Kafka.Effectful.Producer.Effect (KafkaProducer (..))+import Kafka.Producer (ProducerRecord (prTopic))+import Kafka.Producer qualified as K+import Kafka.Producer.ProducerProperties (ProducerProperties)+import Kafka.Transaction qualified as K+import Kafka.Types (KafkaError)+import OpenTelemetry.Context qualified as Context+import OpenTelemetry.Context.ThreadLocal (getContext)+import OpenTelemetry.Trace.Core (+    SpanArguments (kind),+    SpanKind (Producer),+    Tracer,+    addAttributesToSpanArguments,+    defaultSpanArguments,+    inSpan'',+ )++{- | Run the 'KafkaProducer' effect with OpenTelemetry tracing.++Identical in shape to 'Kafka.Effectful.Producer.Interpreter.runKafkaProducer',+plus an additional 'Tracer' argument used to open a Producer-kind+span around every record-sending operation. The current+'OpenTelemetry.Context.Context' is injected as W3C trace-context+headers on the outgoing record before the underlying+@hw-kafka-client@ produce call runs, so that downstream consumers+can extract the context and continue the trace.++The producer handle is acquired and released via 'Exception.bracket',+exactly as 'runKafkaProducer' does. Errors are thrown via the+'Error' effect.++@since 0.2.0.0+-}+runKafkaProducerTraced ::+    (IOE :> es, Error KafkaError :> es) =>+    Tracer ->+    ProducerProperties ->+    Eff (KafkaProducer : es) a ->+    Eff es a+runKafkaProducerTraced tracer props action =+    Exception.bracket+        acquire+        (Effectful.liftIO . K.closeProducer)+        (\producer -> interpret (handleTracedProducer tracer producer) action)+  where+    acquire = do+        result <- Effectful.liftIO $ K.newProducer props+        case result of+            Left err -> throwError err+            Right producer -> pure producer++handleTracedProducer ::+    (IOE :> es, Error KafkaError :> es) =>+    Tracer ->+    K.KafkaProducer ->+    EffectHandler KafkaProducer es+handleTracedProducer tracer producer _env = \case+    ProduceMessage record ->+        withProducerSpan tracer record $ \instrumentedRecord -> do+            mbErr <- Effectful.liftIO $ K.produceMessage producer instrumentedRecord+            for_ mbErr throwError+    ProduceMessage' record cb ->+        withProducerSpan tracer record $ \instrumentedRecord -> do+            res <-+                Effectful.liftIO $+                    K.produceMessage' producer instrumentedRecord cb+            case res of+                Left (K.ImmediateError err) -> throwError err+                Right () -> pure ()+    ProduceMessageSync record ->+        withProducerSpan tracer record $ \instrumentedRecord -> do+            var <- Effectful.liftIO Concurrent.newEmptyMVar+            res <-+                Effectful.liftIO $+                    K.produceMessage' producer instrumentedRecord (Concurrent.putMVar var)+            case res of+                Left (K.ImmediateError err) -> throwError err+                Right () -> do+                    Effectful.liftIO $ K.flushProducer producer+                    report <- Effectful.liftIO $ Concurrent.takeMVar var+                    case report of+                        K.DeliverySuccess _ offset -> pure offset+                        K.DeliveryFailure _ err -> throwError err+                        K.NoMessageError err -> throwError err+    ProduceMessageBatch records -> do+        -- One span per record so the Producer-kind attributes+        -- (partition, key) are per-record, matching the upstream+        -- reference. Spans are opened sequentially as the list is+        -- traversed.+        results <-+            traverse+                ( \r ->+                    withProducerSpan tracer r $ \instrumentedRecord -> do+                        mbErr <-+                            Effectful.liftIO $+                                K.produceMessage producer instrumentedRecord+                        pure (r, mbErr)+                )+                records+        pure [(r, err) | (r, Just err) <- results]+    FlushProducer ->+        Effectful.liftIO $ K.flushProducer producer+    InitTransactions timeout ->+        throwOnJust $ K.initTransactions producer timeout+    BeginTransaction ->+        throwOnJust $ K.beginTransaction producer+    CommitTransaction timeout ->+        Effectful.liftIO $ K.commitTransaction producer timeout+    AbortTransaction timeout ->+        throwOnJust $ K.abortTransaction producer timeout+    SendOffsetsToTransaction consumer record timeout ->+        Effectful.liftIO $+            K.commitOffsetMessageTransaction producer consumer record timeout+    AskProducerHandle ->+        pure producer+  where+    throwOnJust action' = do+        mbErr <- Effectful.liftIO action'+        for_ mbErr throwError++{- | Open a Producer-kind span around a record-sending action.++Builds the @messaging.*@ attribute set from the record, opens a+span named @\"send \<topic\>\"@, injects the current OTel context+as W3C trace-context headers onto a clone of the record, and runs+the supplied action with that instrumented record.+-}+withProducerSpan ::+    (IOE :> es) =>+    Tracer ->+    ProducerRecord ->+    (ProducerRecord -> Eff es a) ->+    Eff es a+withProducerSpan tracer record action =+    inSpan'' tracer (producerSpanName (prTopic record)) spanArgs $ \newSpan -> do+        ctx <- getContext+        instrumentedRecord <-+            Effectful.liftIO $+                injectTraceContextIntoRecord (Context.insertSpan newSpan ctx) record+        action instrumentedRecord+  where+    spanArgs =+        addAttributesToSpanArguments+            (producerRecordAttributes record)+            defaultSpanArguments{kind = Producer}

@@ -0,0 +1,92 @@+{- | Bridges between @hw-kafka-client@\'s 'Kafka.Types.Headers' (a list+of @(ByteString, ByteString)@) and @http-types@\'s 'RequestHeaders'+(a list of @(CI ByteString, ByteString)@), plus convenience helpers+that fetch the global propagator and inject\/extract a W3C trace+context against a record\'s headers in one call.++The header-bridge functions are pure and round-trip the underlying+bytes verbatim; only the case-sensitivity envelope changes (Kafka+headers are case-sensitive, HTTP headers are not).++@since 0.2.0.0+-}+module Kafka.Effectful.OpenTelemetry.Propagation (+    -- * Header bridges+    kafkaHeadersToRequestHeaders,+    requestHeadersToKafkaHeaders,++    -- * W3C trace-context round-trip+    extractTraceContextFromRecord,+    injectTraceContextIntoRecord,+)+where++import Data.Bifunctor (first)+import Data.CaseInsensitive qualified as CI+import Kafka.Consumer.Types (ConsumerRecord (crHeaders))+import Kafka.Producer.Types (ProducerRecord (prHeaders))+import Kafka.Types (Headers, headersFromList, headersToList)+import Network.HTTP.Types (RequestHeaders)+import OpenTelemetry.Context (Context)+import OpenTelemetry.Propagator (extract, inject)+import OpenTelemetry.Trace.Core (+    getGlobalTracerProvider,+    getTracerProviderPropagators,+ )++{- | Convert @hw-kafka-client@ 'Headers' (case-sensitive) to+@http-types@ 'RequestHeaders' (case-insensitive). Each+@(bsKey, bsVal)@ becomes @(CI.mk bsKey, bsVal)@.+-}+kafkaHeadersToRequestHeaders :: Headers -> RequestHeaders+kafkaHeadersToRequestHeaders = map (first CI.mk) . headersToList++{- | Convert @http-types@ 'RequestHeaders' (case-insensitive) back to+@hw-kafka-client@ 'Headers' (case-sensitive). The+'CI.foldedCase' lower-case form of each key is used as the+resulting Kafka header key.+-}+requestHeadersToKafkaHeaders :: RequestHeaders -> Headers+requestHeadersToKafkaHeaders = headersFromList . map (first CI.foldedCase)++{- | Extract a W3C trace context from a 'ConsumerRecord'\'s headers+and merge it into the supplied 'Context'.++This fetches the global tracer provider\'s propagator, hands it the+record\'s headers (after the case-insensitivity bridge), and returns+the resulting 'Context'. If the record carries no @traceparent@+header the propagator returns the input 'Context' unchanged.++This is the building block that the traced consumer interpreter uses+to root a per-message Consumer-kind span at the inbound trace+context.+-}+extractTraceContextFromRecord ::+    ConsumerRecord k v ->+    Context ->+    IO Context+extractTraceContextFromRecord record ctx = do+    propagator <- getTracerProviderPropagators <$> getGlobalTracerProvider+    extract propagator (kafkaHeadersToRequestHeaders (crHeaders record)) ctx++{- | Inject the supplied 'Context'\'s W3C trace context into a+'ProducerRecord'\'s headers, returning the augmented record.++Existing headers on the record are preserved; the propagator-emitted+headers (typically @traceparent@ and optionally @tracestate@) are+appended via 'Headers'\'s 'Semigroup' instance. The original record is+not mutated.++This is the building block that the traced producer interpreter uses+to publish the current span\'s context on the wire so that downstream+consumers can extract it and continue the trace.+-}+injectTraceContextIntoRecord ::+    Context ->+    ProducerRecord ->+    IO ProducerRecord+injectTraceContextIntoRecord ctx record = do+    propagator <- getTracerProviderPropagators <$> getGlobalTracerProvider+    extraHeaders <- inject propagator ctx []+    let merged = prHeaders record <> requestHeadersToKafkaHeaders extraHeaders+    pure record{prHeaders = merged}

@@ -0,0 +1,199 @@+{- | Pure helpers that translate Kafka producer and consumer records+into the OpenTelemetry messaging-semantic-conventions attribute set.++The functions here do not perform any I\/O and do not depend on a+'OpenTelemetry.Trace.Core.Tracer'. They exist as building blocks for+the traced interpreters in+"Kafka.Effectful.OpenTelemetry.Producer.Interpreter" and+"Kafka.Effectful.OpenTelemetry.Consumer.Interpreter", and are also+exposed so users who want to write a custom interpreter or a+framework wrapper can produce the same spec-aligned attribute set+without having to redefine the keys themselves.++The attribute keys and value types follow the OpenTelemetry messaging+spec v1.24 as exposed by+@hs-opentelemetry-semantic-conventions@, and they intentionally+agree with what+@shibuya-kafka-adapter@\'s @Shibuya.Adapter.Kafka.Convert@ produces+(@kafkaSpanAttributes@), so the two libraries can be layered without+attribute key drift.++The design parallels the upstream+@hs-opentelemetry-instrumentation-hw-kafka-client@\'s+@OpenTelemetry.Instrumentation.Kafka@ module.++@since 0.2.0.0+-}+module Kafka.Effectful.OpenTelemetry.Semantic (+    -- * Constants+    kafkaMessagingSystem,+    producerOperationName,+    consumerOperationName,++    -- * Span names+    producerSpanName,+    consumerSpanName,++    -- * Attribute builders+    producerRecordAttributes,+    consumerRecordAttributes,+)+where++import Data.ByteString (ByteString)+import Data.Int (Int64)+import Data.Text (Text)+import Data.Text.Encoding (decodeUtf8')+import Kafka.Consumer.Types (+    ConsumerRecord (crKey, crOffset, crPartition, crTopic),+    Offset (unOffset),+ )+import Kafka.Producer.Types (+    ProducePartition (SpecifiedPartition, UnassignedPartition),+    ProducerRecord (prKey, prPartition, prTopic),+ )+import Kafka.Types (PartitionId (unPartitionId), TopicName (..))+import OpenTelemetry.Attributes.Attribute (toAttribute)+import OpenTelemetry.Attributes.Map (AttributeMap, insertAttributeByKey)+import OpenTelemetry.SemanticConventions (+    messaging_destination_name,+    messaging_kafka_destination_partition,+    messaging_kafka_message_key,+    messaging_kafka_message_offset,+    messaging_operation,+    messaging_system,+ )++{- | The constant @"kafka"@ used as the value of the+@messaging.system@ attribute.+-}+kafkaMessagingSystem :: Text+kafkaMessagingSystem = "kafka"++{- | The operation name carried by producer-side spans+(@\"send\"@). The full span name is built by 'producerSpanName'.+-}+producerOperationName :: Text+producerOperationName = "send"++{- | The operation name carried by consumer-side spans+(@\"process\"@). The full span name is built by 'consumerSpanName'.+-}+consumerOperationName :: Text+consumerOperationName = "process"++{- | Build the canonical producer span name @\"send \<topic\>\"@.++This is the name shape the OTel messaging semantic conventions+recommend for Producer-kind spans (e.g. @"send orders"@).+-}+producerSpanName :: TopicName -> Text+producerSpanName (TopicName t) = producerOperationName <> " " <> t++{- | Build the canonical consumer span name @\"process \<topic\>\"@.++This is the name shape the OTel messaging semantic conventions+recommend for Consumer-kind spans (e.g. @"process orders"@).+-}+consumerSpanName :: TopicName -> Text+consumerSpanName (TopicName t) = consumerOperationName <> " " <> t++{- | Build the OpenTelemetry attribute map for a Producer-kind span+describing a single 'ProducerRecord'.++The map always includes @messaging.system=kafka@,+@messaging.destination.name=<topic>@, and+@messaging.operation=send@. It also includes+@messaging.kafka.destination.partition@ (Int64) when the record+targets a 'SpecifiedPartition' (omitted for 'UnassignedPartition',+since the broker will pick the partition), and+@messaging.kafka.message.key@ (Text) when the record\'s @prKey@+is present and decodes as UTF-8.+-}+producerRecordAttributes :: ProducerRecord -> AttributeMap+producerRecordAttributes record =+    addOperation+        . addDestination+        . addPartition+        . addKey+        . addSystem+        $ mempty+  where+    addSystem =+        insertAttributeByKey messaging_system $+            toAttribute kafkaMessagingSystem+    addOperation =+        insertAttributeByKey messaging_operation $+            toAttribute producerOperationName+    addDestination =+        insertAttributeByKey messaging_destination_name $+            toAttribute (unTopicName (prTopic record))+    addPartition = case prPartition record of+        SpecifiedPartition p ->+            insertAttributeByKey messaging_kafka_destination_partition $+                toAttribute (fromIntegral p :: Int64)+        UnassignedPartition -> id+    addKey = case prKey record >>= decodeKey of+        Just k ->+            insertAttributeByKey messaging_kafka_message_key $+                toAttribute k+        Nothing -> id++{- | Build the OpenTelemetry attribute map for a Consumer-kind span+describing a single 'ConsumerRecord'.++The map always includes @messaging.system=kafka@,+@messaging.destination.name=<topic>@,+@messaging.operation=process@,+@messaging.kafka.destination.partition@ (Int64), and+@messaging.kafka.message.offset@ (Int64). It also includes+@messaging.kafka.message.key@ (Text) when the record\'s @crKey@+is present and decodes as UTF-8.++Note: @messaging.kafka.consumer.group@ is not added here because+the consumer-group identifier lives on the+@Kafka.Consumer.ConsumerProperties@ value, not on the record.+Callers that need it (the traced interpreter does) add it on top+of what this helper returns.+-}+consumerRecordAttributes ::+    ConsumerRecord (Maybe ByteString) (Maybe ByteString) ->+    AttributeMap+consumerRecordAttributes record =+    addOperation+        . addDestination+        . addPartition+        . addOffset+        . addKey+        . addSystem+        $ mempty+  where+    addSystem =+        insertAttributeByKey messaging_system $+            toAttribute kafkaMessagingSystem+    addOperation =+        insertAttributeByKey messaging_operation $+            toAttribute consumerOperationName+    addDestination =+        insertAttributeByKey messaging_destination_name $+            toAttribute (unTopicName (crTopic record))+    addPartition =+        insertAttributeByKey messaging_kafka_destination_partition $+            toAttribute (fromIntegral (unPartitionId (crPartition record)) :: Int64)+    addOffset =+        insertAttributeByKey messaging_kafka_message_offset $+            toAttribute (unOffset (crOffset record))+    addKey = case crKey record >>= decodeKey of+        Just k ->+            insertAttributeByKey messaging_kafka_message_key $+                toAttribute k+        Nothing -> id++{- | Decode message-key bytes as UTF-8, returning 'Nothing' if the+bytes are not valid UTF-8. Matches the behavior of upstream+@OpenTelemetry.Instrumentation.Kafka@\'s @prKey \/ crKey@ handling.+-}+decodeKey :: ByteString -> Maybe Text+decodeKey bs = case decodeUtf8' bs of+    Right t -> Just t+    Left _ -> Nothing

@@ -7,8 +7,26 @@      -- * Operations     produceMessage,+    produceMessage',+    produceMessageSync,+    produceMessageBatch,     flushProducer, +    -- * Transactions+    initTransactions,+    beginTransaction,+    commitTransaction,+    abortTransaction,+    commitOffsetMessageTransaction,+    TxError,+    getKafkaError,+    kafkaErrorIsFatal,+    kafkaErrorIsRetriable,+    kafkaErrorTxnRequiresAbort,++    -- * Raw handle escape hatch+    askProducerHandle,+     -- * Types     ProducerRecord (..),     ProducePartition (..),@@ -52,11 +70,31 @@ ) where -import Kafka.Effectful.Producer.Effect (KafkaProducer, flushProducer, produceMessage)+import Kafka.Effectful.Producer.Effect (+    KafkaProducer,+    abortTransaction,+    askProducerHandle,+    beginTransaction,+    commitTransaction,+    flushProducer,+    initTransactions,+    produceMessage,+    produceMessage',+    produceMessageBatch,+    produceMessageSync,+ ) import Kafka.Effectful.Producer.Interpreter (runKafkaProducer)+import Kafka.Effectful.Producer.Transaction (commitOffsetMessageTransaction) import Kafka.Producer.ProducerProperties (ProducerProperties (..)) import Kafka.Producer.ProducerProperties qualified as K import Kafka.Producer.Types (DeliveryReport (..), ImmediateError (..), ProducePartition (..), ProducerRecord (..))+import Kafka.Transaction (+    TxError,+    getKafkaError,+    kafkaErrorIsFatal,+    kafkaErrorIsRetriable,+    kafkaErrorTxnRequiresAbort,+ ) import Kafka.Types (BrokerAddress (..), Headers, KafkaCompressionCodec (..), KafkaDebug (..), KafkaError (..), KafkaLogLevel (..), Timeout (..), TopicName (..), headersFromList, headersToList)  -- Offset is in Consumer.Types

@@ -4,18 +4,67 @@      -- * Operations     produceMessage,+    produceMessage',+    produceMessageSync,+    produceMessageBatch,     flushProducer,++    -- * Transactions+    initTransactions,+    beginTransaction,+    commitTransaction,+    abortTransaction,++    -- ** Internal — cross-effect plumbing+    sendOffsetsToTransaction,+    askProducerHandle, ) where  import Effectful (Dispatch (..), DispatchOf, Eff, Effect, (:>)) import Effectful.Dispatch.Dynamic (send)-import Kafka.Producer.Types (ProducerRecord)+import Kafka.Consumer.Types (ConsumerRecord, Offset)+import Kafka.Consumer.Types qualified as KC+import Kafka.Producer.Types (DeliveryReport, ProducerRecord)+import Kafka.Producer.Types qualified as KP+import Kafka.Transaction (TxError)+import Kafka.Types (KafkaError, Timeout)  -- | Effect for Kafka producer operations. data KafkaProducer :: Effect where-    ProduceMessage :: ProducerRecord -> KafkaProducer m ()-    FlushProducer :: KafkaProducer m ()+    ProduceMessage ::+        ProducerRecord ->+        KafkaProducer m ()+    ProduceMessage' ::+        ProducerRecord ->+        (DeliveryReport -> IO ()) ->+        KafkaProducer m ()+    ProduceMessageSync ::+        ProducerRecord ->+        KafkaProducer m Offset+    ProduceMessageBatch ::+        [ProducerRecord] ->+        KafkaProducer m [(ProducerRecord, KafkaError)]+    FlushProducer ::+        KafkaProducer m ()+    InitTransactions ::+        Timeout ->+        KafkaProducer m ()+    BeginTransaction ::+        KafkaProducer m ()+    CommitTransaction ::+        Timeout ->+        KafkaProducer m (Maybe TxError)+    AbortTransaction ::+        Timeout ->+        KafkaProducer m ()+    SendOffsetsToTransaction ::+        KC.KafkaConsumer ->+        ConsumerRecord k v ->+        Timeout ->+        KafkaProducer m (Maybe TxError)+    AskProducerHandle ::+        KafkaProducer m KP.KafkaProducer  type instance DispatchOf KafkaProducer = 'Dynamic @@ -25,6 +74,144 @@ produceMessage :: (KafkaProducer :> es) => ProducerRecord -> Eff es () produceMessage = send . ProduceMessage +{- | Send a single message with a per-message 'DeliveryReport' callback.++The callback runs on a librdkafka-forked thread, so blocking operations+(such as writing to an @MVar@) are safe. Throws 'KafkaError' via the+'Error' effect when the underlying send fails to enqueue+(@ImmediateError@).++This is the low-level primitive for Scenario 2 of+@hw-kafka-client@'s producer best practices. Callers that only need a+single synchronous send should prefer 'produceMessageSync'.++@since 0.2.0.0+-}+produceMessage' ::+    (KafkaProducer :> es) =>+    ProducerRecord ->+    (DeliveryReport -> IO ()) ->+    Eff es ()+produceMessage' record cb = send (ProduceMessage' record cb)++{- | Send a single message and block until the broker acknowledges it,+returning the broker-assigned 'Offset'.++Throws 'KafkaError' via the 'Error' effect on enqueue failure+(@ImmediateError@) or on delivery failure reported via the+'DeliveryReport'.++This is the high-level convenience for Scenario 2 of+@hw-kafka-client@'s producer best practices.++@since 0.2.0.0+-}+produceMessageSync ::+    (KafkaProducer :> es) =>+    ProducerRecord ->+    Eff es Offset+produceMessageSync = send . ProduceMessageSync++{- | Send many records in one call.++Returns only the records that failed to enqueue, paired with the+error librdkafka reported for that record. Successful records are+omitted. This mirrors @Kafka.Producer.produceMessageBatch@.++Combined with @linger.ms@ and @batch.size@ set on the+@ProducerProperties@, this is the throughput-oriented path — Scenario+4 of @hw-kafka-client@'s producer best practices.++@since 0.2.0.0+-}+produceMessageBatch ::+    (KafkaProducer :> es) =>+    [ProducerRecord] ->+    Eff es [(ProducerRecord, KafkaError)]+produceMessageBatch = send . ProduceMessageBatch+ -- | Flush the producer's outbound queue, blocking until all messages are sent. flushProducer :: (KafkaProducer :> es) => Eff es () flushProducer = send FlushProducer++{- | Initialise the transactional producer.++Must be called exactly once per producer, after @runKafkaProducer@+has acquired the handle and before any call to 'beginTransaction'.+The producer's @ProducerProperties@ must set @transactional.id@,+@enable.idempotence=true@, and @acks=all@.++Throws 'KafkaError' via the 'Error' effect on failure.++@since 0.2.0.0+-}+initTransactions :: (KafkaProducer :> es) => Timeout -> Eff es ()+initTransactions = send . InitTransactions++{- | Open a new transaction.++Must be preceded by exactly one successful 'initTransactions' on the+same producer handle. Throws 'KafkaError' via the 'Error' effect on+failure.++@since 0.2.0.0+-}+beginTransaction :: (KafkaProducer :> es) => Eff es ()+beginTransaction = send BeginTransaction++{- | Commit the currently-open transaction.++Returns @Nothing@ on success or @Just TxError@ on failure. The+caller must branch on the three 'TxError' discriminators+(@kafkaErrorTxnRequiresAbort@ first, then @kafkaErrorIsRetriable@,+then @kafkaErrorIsFatal@) to decide whether to abort, retry, or+crash.++@since 0.2.0.0+-}+commitTransaction :: (KafkaProducer :> es) => Timeout -> Eff es (Maybe TxError)+commitTransaction = send . CommitTransaction++{- | Abort the currently-open transaction.++Throws 'KafkaError' via the 'Error' effect on failure.++@since 0.2.0.0+-}+abortTransaction :: (KafkaProducer :> es) => Timeout -> Eff es ()+abortTransaction = send . AbortTransaction++{- | Send the consumer offsets for a single 'ConsumerRecord' to the+open transaction.++This is plumbing used by+'Kafka.Effectful.Producer.Transaction.commitOffsetMessageTransaction'+and is not intended to be called directly — end-users should use the+helper, which picks up the consumer handle automatically via+@askConsumerHandle@.++@since 0.2.0.0+-}+sendOffsetsToTransaction ::+    (KafkaProducer :> es) =>+    KC.KafkaConsumer ->+    ConsumerRecord k v ->+    Timeout ->+    Eff es (Maybe TxError)+sendOffsetsToTransaction consumer record timeout =+    send (SendOffsetsToTransaction consumer record timeout)++{- | Escape hatch: return the raw @Kafka.Producer.KafkaProducer@ handle+acquired by @runKafkaProducer@.++Exposed to enable the cross-effect+'Kafka.Effectful.Producer.Transaction.commitOffsetMessageTransaction'+helper, which must reach both the producer and consumer handles to+call the underlying transactional offset-commit primitive. New+operations should go through the 'KafkaProducer' effect rather than+this handle.++@since 0.2.0.0+-}+askProducerHandle :: (KafkaProducer :> es) => Eff es KP.KafkaProducer+askProducerHandle = send AskProducerHandle

@@ -10,6 +10,7 @@ ) where +import Control.Concurrent.MVar qualified as Concurrent import Data.Foldable (for_) import Effectful (Eff, IOE, (:>)) import Effectful qualified@@ -19,6 +20,7 @@ import Kafka.Effectful.Producer.Effect (KafkaProducer (..)) import Kafka.Producer qualified as K import Kafka.Producer.ProducerProperties (ProducerProperties)+import Kafka.Transaction qualified as K import Kafka.Types (KafkaError)  {- | Run the 'KafkaProducer' effect.@@ -51,5 +53,48 @@     ProduceMessage record -> do         mbErr <- Effectful.liftIO $ K.produceMessage producer record         for_ mbErr throwError+    ProduceMessage' record cb -> do+        res <- Effectful.liftIO $ K.produceMessage' producer record cb+        case res of+            Left (K.ImmediateError err) -> throwError err+            Right () -> pure ()+    ProduceMessageBatch records -> Effectful.liftIO $ do+        -- Hackage hw-kafka-client-5.3.0 does not export+        -- 'Kafka.Producer.produceMessageBatch', so we inline the same+        -- definition it ships on master: mapM over the list and keep+        -- only the records that failed to enqueue.+        results <- mapM (\r -> (r,) <$> K.produceMessage producer r) records+        pure [(r, err) | (r, Just err) <- results]+    ProduceMessageSync record -> do+        var <- Effectful.liftIO Concurrent.newEmptyMVar+        res <-+            Effectful.liftIO $+                K.produceMessage' producer record (Concurrent.putMVar var)+        case res of+            Left (K.ImmediateError err) -> throwError err+            Right () -> do+                Effectful.liftIO $ K.flushProducer producer+                report <- Effectful.liftIO $ Concurrent.takeMVar var+                case report of+                    K.DeliverySuccess _ offset -> pure offset+                    K.DeliveryFailure _ err -> throwError err+                    K.NoMessageError err -> throwError err     FlushProducer ->         Effectful.liftIO $ K.flushProducer producer+    InitTransactions timeout ->+        throwOnJust $ K.initTransactions producer timeout+    BeginTransaction ->+        throwOnJust $ K.beginTransaction producer+    CommitTransaction timeout ->+        Effectful.liftIO $ K.commitTransaction producer timeout+    AbortTransaction timeout ->+        throwOnJust $ K.abortTransaction producer timeout+    SendOffsetsToTransaction consumer record timeout ->+        Effectful.liftIO $+            K.commitOffsetMessageTransaction producer consumer record timeout+    AskProducerHandle ->+        pure producer+  where+    throwOnJust action' = do+        mbErr <- Effectful.liftIO action'+        for_ mbErr throwError

@@ -0,0 +1,42 @@+module Kafka.Effectful.Producer.Transaction (+    commitOffsetMessageTransaction,+)+where++import Effectful (Eff, (:>))+import Kafka.Consumer.Types (ConsumerRecord)+import Kafka.Effectful.Consumer.Effect (KafkaConsumer, askConsumerHandle)+import Kafka.Effectful.Producer.Effect (KafkaProducer, sendOffsetsToTransaction)+import Kafka.Transaction (TxError)+import Kafka.Types (Timeout)++{- | Commit the offset of a single 'ConsumerRecord' inside the current+transaction of the surrounding 'KafkaProducer'.++This is the cross-effect counterpart to the consumer's+@commitOffsetMessage@ — the record's offset is not stored via the+consumer's offset-commit path but instead shipped as part of the+producer's open transaction, which preserves the exactly-once+semantics required by Scenario 5 of @hw-kafka-client@'s producer+best practices.++Returns @Nothing@ on success or @Just TxError@ on failure. The+caller must dispatch on the three 'TxError' discriminators+(@kafkaErrorTxnRequiresAbort@ first, then @kafkaErrorIsRetriable@,+then @kafkaErrorIsFatal@) to decide whether to abort the+transaction, retry the commit, or crash.++The helper picks up the consumer handle automatically via+@askConsumerHandle@, so callers only need to have both 'KafkaProducer'+and 'KafkaConsumer' in the effect stack.++@since 0.2.0.0+-}+commitOffsetMessageTransaction ::+    (KafkaProducer :> es, KafkaConsumer :> es) =>+    ConsumerRecord k v ->+    Timeout ->+    Eff es (Maybe TxError)+commitOffsetMessageTransaction record timeout = do+    consumer <- askConsumerHandle+    sendOffsetsToTransaction consumer record timeout

@@ -0,0 +1,187 @@+module Kafka.Effectful.OpenTelemetry.PropagationTest (tests) where++import Data.ByteString (ByteString)+import Data.ByteString.Char8 qualified as BSC+import Data.CaseInsensitive qualified as CI+import Data.Text (Text)+import Data.Text qualified as Text+import Kafka.Consumer.Types (+    ConsumerRecord (..),+    Offset (..),+    Timestamp (NoTimestamp),+ )+import Kafka.Effectful.OpenTelemetry.Propagation (+    extractTraceContextFromRecord,+    injectTraceContextIntoRecord,+    kafkaHeadersToRequestHeaders,+    requestHeadersToKafkaHeaders,+ )+import Kafka.Producer.Types (+    ProducePartition (UnassignedPartition),+    ProducerRecord (..),+ )+import Kafka.Types (+    PartitionId (..),+    TopicName (..),+    headersFromList,+    headersToList,+ )+import OpenTelemetry.Context qualified as Context+import OpenTelemetry.Trace (initializeGlobalTracerProvider)+import OpenTelemetry.Trace.Core (+    SpanContext (..),+    defaultTraceFlags,+    getSpanContext,+    wrapSpanContext,+ )+import OpenTelemetry.Trace.Id (+    Base (Base16),+    SpanId,+    TraceId,+    baseEncodedToSpanId,+    baseEncodedToTraceId,+    traceIdBaseEncodedText,+ )+import OpenTelemetry.Trace.TraceState qualified as TraceState+import Test.Tasty (TestTree, testGroup, withResource)+import Test.Tasty.HUnit (assertBool, assertEqual, assertFailure, testCase)++{- | Known traceparent value derived from the W3C trace-context+specification\'s example. Trace ID+@0af7651916cd43dd8448eb211c80319c@, span ID+@b7ad6b7169203331@, sampled (flag @01@).+-}+sampleTraceparent :: ByteString+sampleTraceparent =+    "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01"++sampleTraceIdHex :: Text+sampleTraceIdHex = "0af7651916cd43dd8448eb211c80319c"++sampleSpanIdHex :: Text+sampleSpanIdHex = "b7ad6b7169203331"++tests :: TestTree+tests =+    -- Initialize the global tracer provider exactly once for the whole+    -- test group. The SDK\'s default propagator pipeline includes+    -- @w3cTraceContextPropagator@, which is what+    -- @injectTraceContextIntoRecord@ \/ @extractTraceContextFromRecord@+    -- consult under the hood.+    withResource initializeGlobalTracerProvider (\_ -> pure ()) $ \_ ->+        testGroup+            "Propagation"+            [ testCase "round-trip kafka headers <-> request headers (case-fold)" $ do+                let original =+                        headersFromList+                            [ ("traceparent", sampleTraceparent)+                            , ("Custom-Header", "value")+                            ]+                    roundTripped =+                        requestHeadersToKafkaHeaders+                            (kafkaHeadersToRequestHeaders original)+                lookup "traceparent" (headersToList roundTripped)+                    `shouldBeJust` sampleTraceparent+                lookup "custom-header" (headersToList roundTripped)+                    `shouldBeJust` "value"+            , testCase "kafkaHeadersToRequestHeaders preserves traceparent value" $ do+                let h = headersFromList [("traceparent", sampleTraceparent)]+                    rh = kafkaHeadersToRequestHeaders h+                lookup (CI.mk "traceparent") rh+                    `shouldBeJust` sampleTraceparent+            , testCase "injectTraceContextIntoRecord adds W3C traceparent" $ do+                tid <- decodeHexTraceId sampleTraceIdHex+                sid <- decodeHexSpanId sampleSpanIdHex+                let ctx =+                        Context.insertSpan+                            (wrapSpanContext (frozenContextWith tid sid))+                            Context.empty+                injected <-+                    injectTraceContextIntoRecord+                        ctx+                        emptyProducerRecord+                let injectedHeaders = headersToList (prHeaders injected)+                case lookup "traceparent" injectedHeaders of+                    Nothing ->+                        assertFailure+                            "expected the producer record to carry a traceparent header"+                    Just header ->+                        assertBool+                            ( "traceparent did not contain the trace-id "+                                <> Text.unpack sampleTraceIdHex+                                <> ", got: "+                                <> BSC.unpack header+                            )+                            (BSC.pack (Text.unpack sampleTraceIdHex) `BSC.isInfixOf` header)+            , testCase "extractTraceContextFromRecord recovers parent context" $ do+                let cr = consumerRecordWithHeaders sampleTraceparent+                ctx <-+                    extractTraceContextFromRecord cr Context.empty+                case Context.lookupSpan ctx of+                    Nothing ->+                        assertFailure+                            "expected the extracted context to carry a span"+                    Just span_ -> do+                        sc <- getSpanContext span_+                        let recovered = traceIdBaseEncodedText Base16 (traceId sc)+                        assertEqual+                            "recovered trace-id should match the inbound traceparent"+                            sampleTraceIdHex+                            recovered+            ]++shouldBeJust :: (Eq a, Show a) => Maybe a -> a -> IO ()+shouldBeJust Nothing expected =+    assertFailure ("expected Just " <> show expected <> ", got Nothing")+shouldBeJust (Just actual) expected =+    assertEqual "values differ" expected actual++decodeHexTraceId :: Text -> IO TraceId+decodeHexTraceId hex =+    case baseEncodedToTraceId Base16 (BSC.pack (Text.unpack hex)) of+        Right tid -> pure tid+        Left err -> assertFailure ("invalid hex trace id: " <> err) >> error "unreachable"++decodeHexSpanId :: Text -> IO SpanId+decodeHexSpanId hex =+    case baseEncodedToSpanId Base16 (BSC.pack (Text.unpack hex)) of+        Right sid -> pure sid+        Left err -> assertFailure ("invalid hex span id: " <> err) >> error "unreachable"++{- | Build a 'SpanContext' that carries the supplied trace and span+IDs. Trace flags are 'defaultTraceFlags' (unsampled); the W3C+propagator preserves the bytes anyway.+-}+frozenContextWith :: TraceId -> SpanId -> SpanContext+frozenContextWith tid sid =+    SpanContext+        { traceId = tid+        , spanId = sid+        , traceFlags = defaultTraceFlags+        , isRemote = False+        , traceState = TraceState.empty+        }++emptyProducerRecord :: ProducerRecord+emptyProducerRecord =+    ProducerRecord+        { prTopic = TopicName "demo"+        , prPartition = UnassignedPartition+        , prKey = Nothing+        , prValue = Just "value"+        , prHeaders = headersFromList []+        }++consumerRecordWithHeaders ::+    ByteString ->+    ConsumerRecord (Maybe ByteString) (Maybe ByteString)+consumerRecordWithHeaders traceparent =+    ConsumerRecord+        { crTopic = TopicName "demo"+        , crPartition = PartitionId 0+        , crOffset = Offset 0+        , crTimestamp = NoTimestamp+        , crHeaders = headersFromList [("traceparent", traceparent)]+        , crKey = Nothing+        , crValue = Just "value"+        }

@@ -0,0 +1,171 @@+module Kafka.Effectful.OpenTelemetry.SemanticTest (tests) where++import Data.ByteString (ByteString)+import Data.ByteString qualified as BS+import Data.HashMap.Strict qualified as HashMap+import Data.Int (Int64)+import Data.Text (Text)+import Kafka.Consumer.Types (+    ConsumerRecord (..),+    Offset (..),+    Timestamp (NoTimestamp),+ )+import Kafka.Effectful.OpenTelemetry.Semantic (+    consumerRecordAttributes,+    producerRecordAttributes,+ )+import Kafka.Producer.Types (+    ProducePartition (SpecifiedPartition, UnassignedPartition),+    ProducerRecord (..),+ )+import Kafka.Types (+    PartitionId (..),+    TopicName (..),+    headersFromList,+ )+import OpenTelemetry.Attributes.Attribute (+    Attribute (AttributeValue),+    PrimitiveAttribute (IntAttribute, TextAttribute),+ )+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit (assertEqual, assertFailure, testCase, (@?=))++tests :: TestTree+tests =+    testGroup+        "Semantic"+        [ producerTests+        , consumerTests+        ]++-- Producer-side attribute coverage++producerTests :: TestTree+producerTests =+    testGroup+        "producerRecordAttributes"+        [ testCase "adds messaging.system=kafka" $ do+            let attrs = producerRecordAttributes (sampleProducerRecord (Just "k1"))+            HashMap.lookup "messaging.system" attrs+                @?= Just (textAttr "kafka")+        , testCase "adds messaging.destination.name from topic" $ do+            let attrs = producerRecordAttributes (sampleProducerRecord (Just "k1"))+            HashMap.lookup "messaging.destination.name" attrs+                @?= Just (textAttr "orders")+        , testCase "adds messaging.operation=send" $ do+            let attrs = producerRecordAttributes (sampleProducerRecord (Just "k1"))+            HashMap.lookup "messaging.operation" attrs+                @?= Just (textAttr "send")+        , testCase "adds messaging.kafka.destination.partition for SpecifiedPartition" $ do+            let attrs = producerRecordAttributes (sampleProducerRecord (Just "k1"))+            HashMap.lookup "messaging.kafka.destination.partition" attrs+                @?= Just (intAttr 42)+        , testCase "omits messaging.kafka.destination.partition for UnassignedPartition" $ do+            let attrs =+                    producerRecordAttributes+                        (sampleProducerRecord (Just "k1"))+                            { prPartition = UnassignedPartition+                            }+            assertEqual+                "partition attribute should not be present"+                Nothing+                (HashMap.lookup "messaging.kafka.destination.partition" attrs)+        , testCase "adds messaging.kafka.message.key when valid UTF-8" $ do+            let attrs = producerRecordAttributes (sampleProducerRecord (Just "hello"))+            HashMap.lookup "messaging.kafka.message.key" attrs+                @?= Just (textAttr "hello")+        , testCase "omits messaging.kafka.message.key when invalid UTF-8" $ do+            let attrs =+                    producerRecordAttributes+                        (sampleProducerRecord (Just (BS.pack [0xC3, 0x28])))+            case HashMap.lookup "messaging.kafka.message.key" attrs of+                Nothing -> pure ()+                Just _ ->+                    assertFailure+                        "expected messaging.kafka.message.key to be omitted for invalid UTF-8"+        , testCase "omits messaging.kafka.message.key when key is Nothing" $ do+            let attrs = producerRecordAttributes (sampleProducerRecord Nothing)+            assertEqual+                "key attribute should not be present"+                Nothing+                (HashMap.lookup "messaging.kafka.message.key" attrs)+        ]++-- Consumer-side attribute coverage++consumerTests :: TestTree+consumerTests =+    testGroup+        "consumerRecordAttributes"+        [ testCase "adds messaging.system=kafka" $ do+            let attrs = consumerRecordAttributes (sampleConsumerRecord (Just "k1"))+            HashMap.lookup "messaging.system" attrs+                @?= Just (textAttr "kafka")+        , testCase "adds messaging.destination.name from topic" $ do+            let attrs = consumerRecordAttributes (sampleConsumerRecord (Just "k1"))+            HashMap.lookup "messaging.destination.name" attrs+                @?= Just (textAttr "orders")+        , testCase "adds messaging.operation=process" $ do+            let attrs = consumerRecordAttributes (sampleConsumerRecord (Just "k1"))+            HashMap.lookup "messaging.operation" attrs+                @?= Just (textAttr "process")+        , testCase "adds messaging.kafka.destination.partition" $ do+            let attrs = consumerRecordAttributes (sampleConsumerRecord (Just "k1"))+            HashMap.lookup "messaging.kafka.destination.partition" attrs+                @?= Just (intAttr 7)+        , testCase "adds messaging.kafka.message.offset" $ do+            let attrs = consumerRecordAttributes (sampleConsumerRecord (Just "k1"))+            HashMap.lookup "messaging.kafka.message.offset" attrs+                @?= Just (intAttr 42)+        , testCase "adds messaging.kafka.message.key when valid UTF-8" $ do+            let attrs = consumerRecordAttributes (sampleConsumerRecord (Just "hello"))+            HashMap.lookup "messaging.kafka.message.key" attrs+                @?= Just (textAttr "hello")+        , testCase "omits messaging.kafka.message.key when invalid UTF-8" $ do+            let attrs =+                    consumerRecordAttributes+                        (sampleConsumerRecord (Just (BS.pack [0xC3, 0x28])))+            case HashMap.lookup "messaging.kafka.message.key" attrs of+                Nothing -> pure ()+                Just _ ->+                    assertFailure+                        "expected messaging.kafka.message.key to be omitted for invalid UTF-8"+        , testCase "omits messaging.kafka.message.key when key is Nothing" $ do+            let attrs = consumerRecordAttributes (sampleConsumerRecord Nothing)+            assertEqual+                "key attribute should not be present"+                Nothing+                (HashMap.lookup "messaging.kafka.message.key" attrs)+        ]++-- Helpers++sampleProducerRecord :: Maybe ByteString -> ProducerRecord+sampleProducerRecord key =+    ProducerRecord+        { prTopic = TopicName "orders"+        , prPartition = SpecifiedPartition 42+        , prKey = key+        , prValue = Just "value"+        , prHeaders = headersFromList []+        }++sampleConsumerRecord ::+    Maybe ByteString ->+    ConsumerRecord (Maybe ByteString) (Maybe ByteString)+sampleConsumerRecord key =+    ConsumerRecord+        { crTopic = TopicName "orders"+        , crPartition = PartitionId 7+        , crOffset = Offset 42+        , crTimestamp = NoTimestamp+        , crHeaders = headersFromList []+        , crKey = key+        , crValue = Just "value"+        }++textAttr :: Text -> Attribute+textAttr = AttributeValue . TextAttribute++intAttr :: Int64 -> Attribute+intAttr = AttributeValue . IntAttribute

@@ -0,0 +1,104 @@+module Kafka.Effectful.OpenTelemetry.ShibuyaCompatibilityTest (tests) where++import Data.HashMap.Strict qualified as HashMap+import Data.Int (Int64)+import Data.Text (Text)+import Kafka.Consumer.Types (+    ConsumerRecord (..),+    Offset (..),+    Timestamp (NoTimestamp),+ )+import Kafka.Effectful.OpenTelemetry.Semantic (consumerRecordAttributes)+import Kafka.Types (+    PartitionId (..),+    TopicName (..),+    headersFromList,+ )+import OpenTelemetry.Attributes (Attribute, toAttribute, unkey)+import OpenTelemetry.SemanticConventions (+    messaging_kafka_destination_partition,+    messaging_kafka_message_offset,+ )+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit (assertEqual, testCase)++{- | Tests pinning @Kafka.Effectful.OpenTelemetry.Semantic@\'s+'consumerRecordAttributes' against the attribute set that+@shibuya-kafka-adapter@\'s+@Shibuya.Adapter.Kafka.Convert.kafkaSpanAttributes@ produces.++Both libraries depend on @hs-opentelemetry-semantic-conventions@,+so if upstream renames a key in a future release this test fails+together with shibuya — the desired failure mode for a+compatibility pin.+-}+tests :: TestTree+tests =+    testGroup+        "ShibuyaCompatibility"+        [ testCase "consumerRecordAttributes agrees on messaging.system" $ do+            let actual = HashMap.lookup "messaging.system" sampleAttrs+                expected = HashMap.lookup "messaging.system" shibuyaAttrs+            assertEqual "messaging.system attribute differs" expected actual+        , testCase "consumerRecordAttributes agrees on messaging.kafka.destination.partition" $ do+            let key = unkey messaging_kafka_destination_partition+                actual = HashMap.lookup key sampleAttrs+                expected = HashMap.lookup key shibuyaAttrs+            assertEqual "partition attribute differs" expected actual+        , testCase "consumerRecordAttributes agrees on messaging.kafka.message.offset" $ do+            let key = unkey messaging_kafka_message_offset+                actual = HashMap.lookup key sampleAttrs+                expected = HashMap.lookup key shibuyaAttrs+            assertEqual "offset attribute differs" expected actual+        ]++-- Sample inputs — partition 7, offset 42 — chosen so the resulting+-- attribute values are non-zero and easy to spot in failures.++samplePid :: PartitionId+samplePid = PartitionId 7++sampleOffset :: Offset+sampleOffset = Offset 42++sampleAttrs :: HashMap.HashMap Text Attribute+sampleAttrs =+    consumerRecordAttributes+        ConsumerRecord+            { crTopic = TopicName "orders"+            , crPartition = samplePid+            , crOffset = sampleOffset+            , crTimestamp = NoTimestamp+            , crHeaders = headersFromList []+            , crKey = Nothing+            , crValue = Just "value"+            }++{- | Local replica of @shibuya-kafka-adapter@\'s+@Shibuya.Adapter.Kafka.Convert.kafkaSpanAttributes@. Mirrors the+exact source-level expression at+@shibuya-kafka-adapter\/src\/Shibuya\/Adapter\/Kafka\/Convert.hs:78-89@:++> kafkaSpanAttributes :: PartitionId -> Offset -> HashMap Text Attribute+> kafkaSpanAttributes (PartitionId pid) (Offset off) =+>     HashMap.fromList+>         [ (\"messaging.system\", toAttribute (\"kafka\" :: Text))+>         , (unkey messaging_kafka_destination_partition, toAttribute (fromIntegral pid :: Int64))+>         , (unkey messaging_kafka_message_offset, toAttribute (off :: Int64))+>         ]+-}+shibuyaAttrs :: HashMap.HashMap Text Attribute+shibuyaAttrs =+    let PartitionId pid = samplePid+        Offset off = sampleOffset+     in HashMap.fromList+            [ ("messaging.system", toAttribute ("kafka" :: Text))+            ,+                ( unkey messaging_kafka_destination_partition+                , toAttribute (fromIntegral pid :: Int64)+                )+            ,+                ( unkey messaging_kafka_message_offset+                , toAttribute (off :: Int64)+                )+            ]

@@ -0,0 +1,18 @@+module Main (main) where++import Kafka.Effectful.OpenTelemetry.PropagationTest qualified as PropagationTest+import Kafka.Effectful.OpenTelemetry.SemanticTest qualified as SemanticTest+import Kafka.Effectful.OpenTelemetry.ShibuyaCompatibilityTest qualified as ShibuyaCompatibilityTest+import Test.Tasty (TestTree, defaultMain, testGroup)++main :: IO ()+main = defaultMain tests++tests :: TestTree+tests =+    testGroup+        "Kafka.Effectful.OpenTelemetry"+        [ SemanticTest.tests+        , PropagationTest.tests+        , ShibuyaCompatibilityTest.tests+        ]