shibuya-kafka-adapter 0.12.0.0 → 0.13.0.0

raw patch · 11 files changed

+418/−60 lines, 11 filesdep +hs-opentelemetry-exporter-in-memorydep ~shibuya-corePVP ok

version bump matches the API change (PVP)

Dependencies added: hs-opentelemetry-exporter-in-memory

Dependency ranges changed: shibuya-core

API changes (from Hackage documentation)

Files

• CHANGELOG.md +34/−0
• README.md +177/−41
• shibuya-kafka-adapter.cabal +5/−3
• src/Shibuya/Adapter/Kafka.hs +8/−2
• src/Shibuya/Adapter/Kafka/Config.hs +16/−2
• test/Kafka/TestEnv.hs +3/−2
• test/Main.hs +2/−0
• test/Shibuya/Adapter/Kafka/AckHandleTest.hs +5/−3
• test/Shibuya/Adapter/Kafka/CorrelationTest.hs +158/−0
• test/Shibuya/Adapter/Kafka/IntegrationTest.hs +7/−4
• test/Shibuya/Adapter/Kafka/SourceGateTest.hs +3/−3

@@ -1,5 +1,39 @@ # Changelog +## 0.13.0.0 — 2026-11-14++### Breaking Changes++- Require the published `shibuya-core ^>=0.13.0.0` from+  `mori://shinzui/shibuya` across the library, tests, examples and benchmarks.+  Core 0.13 renames processing spans from `<processor> process` to+  `process <processor>` and adds `messaging.operation.name = "process"`; update+  trace queries, dashboards and alerts that match the old span name.+- Add the strict field `telemetry :: AdapterTelemetry` to `KafkaAdapterConfig`.+  Configurations built from `defaultConfig` with record update need no change;+  a full `KafkaAdapterConfig { .. }` literal must set it, normally to+  `defaultAdapterTelemetry`.++### New Features++- Choose how each record's processing span relates to the producer's span with+  `telemetry = defaultAdapterTelemetry {traceCorrelation = LinkOnly}`. The+  default, `ParentAndLink`, keeps the span in the producer's trace as its child+  and now also links to the producer. `LinkOnly` starts a new trace per record,+  linked to the producer, for replay, sustained lag and wide consumer-group+  fan-out. Without a usable `traceparent` neither adds a link. Re-export+  `AdapterTelemetry (traceCorrelation)`, `defaultAdapterTelemetry` and+  `TraceCorrelation (..)` from `Shibuya.Adapter.Kafka`.++### Other Changes++- Add broker-free correlation tests over the scripted log, document the setting+  in the README, and let `otel-demo` select `LinkOnly` with+  `OTEL_DEMO_LINK_ONLY=true`.+- Refresh the README usage and runtime guidance.+- Update `nix-haskell-flake` to 0.40.0 so the `redpanda-local-*` scripts use+  the host's container CLI.+ ## 0.12.0.0 — 2026-10-26  ### Breaking Changes

@@ -1,70 +1,206 @@ # shibuya-kafka-adapter -Kafka adapter for the [Shibuya](https://github.com/shinzui/shibuya) queue-processing framework.+Consume Kafka messages through the [Shibuya](mori://shinzui/shibuya)+queue-processing framework, with explicit acknowledgements, partition-aware+retries, offset management, and OpenTelemetry context propagation. -Integrates with Apache Kafka via [`kafka-effectful`](https://github.com/shinzui/kafka-effectful) for the consumer effect (polling, offset store, partition pause) and [`hw-kafka-streamly`](https://hackage.haskell.org/package/hw-kafka-streamly) for error classification (`skipNonFatal`), on top of [`hw-kafka-client`](https://github.com/haskell-works/hw-kafka-client). Provides polling, offset commit semantics, partition awareness, and graceful shutdown.+**Release:** [0.12.0.0 on Hackage](https://hackage.haskell.org/package/shibuya-kafka-adapter-0.12.0.0)+· [API documentation](https://hackage.haskell.org/package/shibuya-kafka-adapter-0.12.0.0/docs/Shibuya-Adapter-Kafka.html)+· [Changelog](CHANGELOG.md) -## Runtime Contract+The adapter requires `shibuya-core ^>=0.12.0.0` and GHC 9.12. The repository's+Nix shell uses GHC 9.12.4. It builds on+[kafka-effectful](mori://shinzui/kafka-effectful) for consumer operations and+[hw-kafka-streamly](mori://shinzui/hw-kafka-streamly) for poll-error classification. -Run this adapter with serial message processing only. librdkafka stores the highest offset per partition and does not track gaps, so concurrent finalization can commit past an earlier message that failed, halted, or requested retry. The adapter cannot inspect the processor concurrency policy at construction time; avoiding `Async` and `Ahead` processing is a caller contract until a gap-tracking commit layer exists.+## Using the adapter -Offset reset policy belongs to the `Subscription` passed to `runKafkaConsumer`, for example `topics [TopicName "orders"] <> offsetReset Earliest`. `KafkaAdapterConfig.topics` is metadata used for adapter naming and a construction-time warning if it differs from the live subscription.+Add these dependencies to your application: -`AckRetry` seeks the partition back to the failed message and does not store that message's offset. Records polled before the seek are discarded. Replayed successors wait behind the retried record until it is resolved; if retention or compaction removed that offset, the adapter warns, abandons the missing record, and resumes at the next available offset. `AckDeadLetter` stores the offset so the consumer group moves on, prints `[shibuya-kafka-adapter] WARNING: dead-lettered message DROPPED` to stderr, and makes the message unrecoverable from the group's committed position. This adapter does not include a DLQ producer.+```cabal+build-depends:+  base ^>=4.21,+  effectful-core,+  hw-kafka-client >=5.3 && <6,+  kafka-effectful ^>=0.3.1.0,+  shibuya-core ^>=0.12.0.0,+  shibuya-kafka-adapter ^>=0.12.0.0+``` -Under serial single-message processing, already-buffered successors are skipped before handler entry and redelivered after the retried record resolves. This preserves partition handler order, subject to the missing-offset recovery policy. A retry cannot undo successors already admitted in the same batch; later superseded batches are skipped. Concurrent processing is unsupported and can admit a successor before the retry is recorded.+The consumer scope owns the Kafka connection and subscription. Create the+adapter inside that scope and use `mkProcessor`, which defaults to serial+processing. Disable automatic offset storage so a record's offset is stored+only after acknowledgement. -Kafka does not expose a per-message delivery-attempt counter through this consumer API, so `Envelope.attempt` is always `Nothing`. Handlers that need bounded retries must use their own store or return `AckHalt` to stop the stream.+This example prints messages from `orders` on `localhost:9092`: -`AckHalt` pauses the originating partition and stops the processor. Polling then stops; after `max.poll.interval.ms` (librdkafka default: 300000 ms, or 5 minutes) the broker may evict this consumer and rebalance the partition to another group member. A single-member group stalls until restart.+```haskell+{-# LANGUAGE GHC2024 #-}+{-# LANGUAGE OverloadedRecordDot #-}+{-# LANGUAGE OverloadedStrings #-} -On shutdown, the adapter commits offsets stored so far and signals the source stream to stop. Let the surrounding `runKafkaConsumer` scope end normally after stopping the app so the consumer close path can flush offsets stored during the drain window.+module Main (main) where -For rebalance visibility, create state with `newKafkaAdapterState`, install `Kafka.Consumer.setCallback (Kafka.Consumer.rebalanceCallback (kafkaRebalanceHandler state))` before consumer creation, and pass the same state to `kafkaAdapterWith`. The helper logs rebalance events and fences deliveries from revoked assignments. Deliveries already revoked when their status is checked are skipped before handler entry. A revocation after that admission check cannot cancel an admitted handler.+import Control.Monad.IO.Class (liftIO)+import Effectful (runEff)+import Effectful.Error.Static (runError)+import Kafka.Consumer.Types (OffsetReset (Earliest))+import Kafka.Effectful.Consumer qualified as Kafka+import Shibuya.Adapter.Kafka+  ( BrokerAddress (..), ConsumerGroupId (..), KafkaError, TopicName (..)+  , defaultConfig, kafkaAdapter+  )+import Shibuya.App (ProcessorId (..), defaultAppConfig, mkProcessor, runApp, waitApp)+import Shibuya.Core.Ack (AckDecision (AckOk))+import Shibuya.Core.Ingested (Message (..))+import Shibuya.Core.Types (Envelope (..))+import Shibuya.Telemetry.Effect (runTracingNoop) -## Packages+main :: IO ()+main = do+  result <- runEff . runError @KafkaError . runTracingNoop $ do+    let props = Kafka.brokersList [BrokerAddress "localhost:9092"]+             <> Kafka.groupId (ConsumerGroupId "orders-consumer")+             <> Kafka.noAutoOffsetStore+        sub = Kafka.topics [TopicName "orders"] <> Kafka.offsetReset Earliest+    Kafka.runKafkaConsumer props sub $ do+      adapter <- kafkaAdapter (defaultConfig [TopicName "orders"])+      let handler message = do+            liftIO (print message.envelope.payload)+            pure AckOk+      app <- runApp defaultAppConfig+        [(ProcessorId "orders", mkProcessor adapter handler)]+      case app of+        Left err -> liftIO (fail (show err))+        Right handle -> waitApp handle+  case result of+    Left err -> fail (show err)+    Right () -> pure ()+``` -- `shibuya-kafka-adapter` — the adapter library (`Shibuya.Adapter.Kafka`, `.Config`, `.Convert`).-- `shibuya-kafka-adapter-bench` — micro-benchmarks for the conversion hot path (`ConsumerRecord` → `Envelope`, W3C header extraction, timestamps).-- `shibuya-kafka-adapter-jitsurei` — runnable examples: `BasicConsumer`, `MultiTopic`, `MultiPartition`, `OffsetManagement`.+`defaultConfig` polls with a 1,000 ms timeout and a maximum batch size of 100.+Its `topics` field supplies adapter metadata and is checked against the live+subscription. Subscription topics and offset-reset policy belong to+`runKafkaConsumer`. -## Tracing+## Acknowledgements and ordering -Tracing is handled by `shibuya-core`'s supervised runner. `Shibuya.Adapter.Kafka.Convert.consumerRecordToEnvelope` populates `Envelope.attributes` with `messaging.system=kafka` and the Kafka-specific typed attributes `messaging.kafka.destination.partition` (Int64) and `messaging.kafka.message.offset` (Int64). The framework merges those onto its single Consumer-kind per-message span alongside `messaging.destination.name`, `messaging.operation.type=process`, and `messaging.message.id`.+**Use serial processing.** The adapter does not support Shibuya's `Async` or+`Ahead` policies. librdkafka stores the highest offset per partition without+tracking gaps; concurrent finalization can commit past unresolved work. The+adapter cannot inspect or enforce the processor's concurrency policy. -The old opt-in `Shibuya.Adapter.Kafka.Tracing` module was removed in `0.5.0.0`; callers should run the Kafka adapter through Shibuya's normal `runApp` or `runWithMetrics` path instead of wrapping the stream.+| Decision | Adapter behavior |+| --- | --- |+| `AckOk` | Store the record's offset locally; auto-commit or consumer close flushes it to the broker. |+| `AckRetry` | Keep the earliest unresolved delivery as the retry boundary, seek back to it, and leave its offset unstored. |+| `AckDeadLetter` | Store the offset and log a `dead-lettered message DROPPED` warning. The adapter has no DLQ producer. |+| `AckHalt` | Pause the originating partition without storing its offset; the processor stops. | -## Building+With serial single-message processing, a retry blocks later messages in the+same partition. Pre-seek records are discarded, already-buffered successors+are skipped before handler entry, and replayed successors wait until the+retried record resolves. Later callbacks cannot replace the earliest retry+boundary or store offsets past it. A retry cannot undo messages already+admitted together in the same handler batch. -The repo ships a Nix flake and `direnv` config for a reproducible toolchain.+If retention or compaction removed the retry offset, the adapter warns and+resumes at the next available record. Kafka provides no delivery-attempt+counter here: `Envelope.attempt` is always `Nothing`. Use an external store+for bounded retry counts, or halt processing. -```sh-direnv allow        # or: nix develop-cabal build all-cabal test shibuya-kafka-adapter-```+A dead-letter acknowledgement advances the group's position without publishing+the message elsewhere. Arrange any required DLQ publication before returning+`AckDeadLetter`. -Benchmarks and examples:+## Rebalances and shutdown -```sh-cabal bench shibuya-kafka-adapter-bench-cabal run BasicConsumer-```+For assignment fencing, create state with `newKafkaAdapterState`, add+`Kafka.Consumer.setCallback (Kafka.Consumer.rebalanceCallback (kafkaRebalanceHandler state))`+to the consumer properties before consumer creation, and construct the adapter+with `kafkaAdapterWith state config`. -## Layout+The callback logs assignment changes, clears retry state on revocation, and+fences old delivery callbacks so they cannot store, seek, or pause a later+assignment. Deliveries revoked before their admission check are skipped. A+revocation after that check cannot cancel a handler already admitted. The+helper is optional; `kafkaAdapter` alone does not provide this ownership fence. -```-shibuya-kafka-adapter/            library sources and tests-shibuya-kafka-adapter-bench/      tasty-bench micro-benchmarks-shibuya-kafka-adapter-jitsurei/   runnable usage examples-docs/plans/                       execution plans-mori.dhall                        project manifest (mori registry)+Keep the consumer scope open while stopping the Shibuya application. Adapter+shutdown signals the source to stop and commits offsets already stored. Let+`runKafkaConsumer` close normally afterward so it can flush offsets stored+during draining.++After `AckHalt`, polling stops. The consumer can lose its assignment when+`max.poll.interval.ms` expires; a single-member group needs a restart to resume.+Partition pause state belongs to the consumer session.++## Errors and tracing++Non-fatal poll errors, including timeouts and partition EOFs, are filtered out.+Fatal poll errors terminate the source through `Error KafkaError`. Exhausted+acknowledgement operations throw `KafkaAcknowledgementException` at the+finalizer boundary so Shibuya can retain the processor failure.++The adapter preserves Kafka headers, extracts W3C `traceparent` and `tracestate`,+and adds `messaging.system=kafka`, `messaging.kafka.destination.partition`, and+`messaging.kafka.message.offset` to the envelope. Shibuya's runner uses these+attributes on its per-message span. The example above disables tracing with+`runTracingNoop`; use the configured tracing interpreter to export spans.++## Trace correlation++When a record carries a W3C `traceparent` header, Shibuya opens a Consumer span+named `process <processor>` around the handler and always adds a *link* to the+producer's span. The `telemetry` field of `KafkaAdapterConfig` decides whether+that span is also the producer's child:++```haskell+import Shibuya.Adapter.Kafka++-- Default (ParentAndLink): the handler runs inside the producer's trace.+config = defaultConfig [TopicName "orders"]++-- LinkOnly: each record's processing starts its own trace, linked to the producer.+replaying =+  (defaultConfig [TopicName "orders"])+    { telemetry = defaultAdapterTelemetry {traceCorrelation = LinkOnly}+    } ``` -## License+`ParentAndLink` suits records consumed close to real time: producer and+consumer appear in one connected trace. Choose `LinkOnly` when the producer's+trace has usually ended before processing starts, so the consumer span would+only stretch it: -MIT. See package cabal files for details.+- replay from an old offset, or a reprocessing job over retained data;+- sustained consumer lag;+- many consumer groups on one topic, each of which would otherwise add its+  processing to the same producer trace. +A `LinkOnly` trace is sampled on its own, independently of the producer's+sampling decision, and a tail-sampling collector sees it as a new trace. The+link still lets a trace backend navigate to the producer. -The adapter requires the published `shibuya-core ^>=0.12.0.0`, which provides-the delivery-status API used to skip superseded deliveries before handler entry.+OpenTelemetry's messaging conventions allow creation-context parenting for a+single message and recommend against it by default only when processing runs+inside another span; Shibuya records its choice in its ADR 0014+(`mori://shinzui/shibuya`, `docs/adr/0014-define-consumer-correlation-and-bounded-batch-link-accounting.md`;+artifact URI pending). Without a usable `traceparent` neither setting adds a+link, and the span keeps whatever span is active on the processing thread as+its parent.++Batch processors ignore the setting: a batch span is always a root with one+link per record (the first 128 records with a valid context), each link+carrying the record's `messaging.message.id`, and+`messaging.batch.message_count` on the span.++## Examples and development++See the [repository README](https://github.com/shinzui/shibuya-kafka-adapter#development)+for runnable examples, the Nix toolchain, and live Kafka integration tests. Only+the adapter library is published to Hackage.++## License++MIT. See [LICENSE](LICENSE).

@@ -1,6 +1,6 @@ cabal-version: 3.12 name: shibuya-kafka-adapter-version: 0.12.0.0+version: 0.13.0.0 synopsis: Kafka adapter for the Shibuya queue processing framework description:   A Shibuya adapter that integrates with Apache Kafka via kafka-effectful@@ -67,7 +67,7 @@     hw-kafka-client >=5.3 && <6,     hw-kafka-streamly ^>=0.2,     kafka-effectful ^>=0.3.1.0,-    shibuya-core ^>=0.12.0.0,+    shibuya-core ^>=0.13.0.0,     stm ^>=2.5,     streamly ^>=0.11,     streamly-core ^>=0.3,@@ -106,6 +106,7 @@     Shibuya.Adapter.Kafka.AckHandleTest     Shibuya.Adapter.Kafka.AdapterTest     Shibuya.Adapter.Kafka.ConvertTest+    Shibuya.Adapter.Kafka.CorrelationTest     Shibuya.Adapter.Kafka.DeliveryStatusTest     Shibuya.Adapter.Kafka.IntegrationTest     Shibuya.Adapter.Kafka.SourceGateTest@@ -117,12 +118,13 @@     containers,     effectful-core (>=2.6.1 && <2.7) || (>=2.7.1.1 && <2.8),     hs-opentelemetry-api ^>=1.0,+    hs-opentelemetry-exporter-in-memory ^>=1.0,     hs-opentelemetry-semantic-conventions ^>=1.40,     hw-kafka-client,     kafka-effectful,     process,     random,-    shibuya-core ^>=0.12.0.0,+    shibuya-core ^>=0.13.0.0,     shibuya-kafka-adapter,     stm,     streamly,

@@ -114,6 +114,11 @@     -- * Defaults     defaultConfig, +    -- * Telemetry+    AdapterTelemetry (traceCorrelation),+    defaultAdapterTelemetry,+    TraceCorrelation (..),+     -- * Re-exports     TopicName (..),     BrokerAddress (..),@@ -137,7 +142,7 @@ import Kafka.Consumer.Types qualified as KC import Kafka.Effectful.Consumer.Effect (KafkaConsumer, commitAllOffsets, subscription) import Kafka.Types (BatchSize (..), BrokerAddress (..), KafkaError (..), Timeout (..), TopicName (..))-import Shibuya.Adapter (Adapter (..))+import Shibuya.Adapter (Adapter (..), AdapterTelemetry (traceCorrelation), TraceCorrelation (..), defaultAdapterTelemetry) import Shibuya.Adapter.Kafka.Config (KafkaAdapterConfig (..), defaultConfig) import Shibuya.Adapter.Kafka.Internal (KafkaAcknowledgementException (..), KafkaAdapterState (..), gateRecords, ingestedStream, kafkaSource, markPartitionsAssigned, markPartitionsRevoked, mkIngested, newKafkaAdapterState, withConsumerLock) import System.IO (hPutStrLn, stderr)@@ -193,7 +198,8 @@           withConsumerLock state (commitAllOffsets OffsetCommit)             `catchError` \_ err -> case err of               KafkaResponseError RdKafkaRespErrNoOffset -> pure ()-              _ -> throwError err+              _ -> throwError err,+        telemetry = config.telemetry       }  warnOnSubscriptionMismatch ::

@@ -10,6 +10,7 @@  import GHC.Generics (Generic) import Kafka.Types (BatchSize (..), Timeout (..), TopicName)+import Shibuya.Adapter (AdapterTelemetry, defaultAdapterTelemetry)  -- | Configuration for the Kafka adapter. --@@ -25,7 +26,18 @@     -- | Timeout for each poll call (default: 1000ms)     pollTimeout :: !Timeout,     -- | Maximum messages per poll batch (default: 100)-    batchSize :: !BatchSize+    batchSize :: !BatchSize,+    -- | Telemetry options copied unchanged into the 'Shibuya.Adapter.Adapter'.+    --+    -- The default is 'defaultAdapterTelemetry', whose @traceCorrelation@ is+    -- 'Shibuya.Adapter.ParentAndLink': each record's processing span is a+    -- child of the producer's span and also links to it, giving one connected+    -- trace. Choose 'Shibuya.Adapter.LinkOnly' with+    -- @defaultAdapterTelemetry {traceCorrelation = LinkOnly}@ when records are+    -- replayed from old offsets, when the consumer runs with sustained lag, or+    -- when many consumer groups read the same topic: each record's processing+    -- then starts its own trace, related to the producer only by the link.+    telemetry :: !AdapterTelemetry   }   deriving stock (Show, Eq, Generic) @@ -35,10 +47,12 @@ -- -- * @pollTimeout@: 1000ms -- * @batchSize@: 100+-- * @telemetry@: 'defaultAdapterTelemetry' (@traceCorrelation = ParentAndLink@) defaultConfig :: [TopicName] -> KafkaAdapterConfig defaultConfig ts =   KafkaAdapterConfig     { topics = ts,       pollTimeout = Timeout 1000,-      batchSize = BatchSize 100+      batchSize = BatchSize 100,+      telemetry = defaultAdapterTelemetry     }

@@ -54,7 +54,7 @@     TopicName (..),   ) import Shibuya.Adapter (Adapter (..))-import Shibuya.Adapter.Kafka (KafkaAdapterConfig (..), kafkaAdapter)+import Shibuya.Adapter.Kafka (KafkaAdapterConfig (..), defaultAdapterTelemetry, kafkaAdapter) import Shibuya.Core.Ack (AckDecision (..)) import Shibuya.Core.AckHandle (AckHandle (..)) import Shibuya.Core.Ingested (Ingested (..))@@ -207,7 +207,8 @@             KafkaAdapterConfig               { topics = [env.testTopic],                 pollTimeout = Timeout 5000,-                batchSize = BatchSize 100+                batchSize = BatchSize 100,+                telemetry = defaultAdapterTelemetry               }       Adapter {source} <- kafkaAdapter config       Stream.fold Fold.drain

@@ -3,6 +3,7 @@ import Shibuya.Adapter.Kafka.AckHandleTest qualified as AckHandleTest import Shibuya.Adapter.Kafka.AdapterTest qualified as AdapterTest import Shibuya.Adapter.Kafka.ConvertTest qualified as ConvertTest+import Shibuya.Adapter.Kafka.CorrelationTest qualified as CorrelationTest import Shibuya.Adapter.Kafka.DeliveryStatusTest qualified as DeliveryStatusTest import Shibuya.Adapter.Kafka.IntegrationTest qualified as IntegrationTest import Shibuya.Adapter.Kafka.SourceGateTest qualified as SourceGateTest@@ -16,6 +17,7 @@       [ AckHandleTest.tests,         AdapterTest.tests,         ConvertTest.tests,+        CorrelationTest.tests,         DeliveryStatusTest.tests,         SourceGateTest.tests,         IntegrationTest.tests

@@ -14,7 +14,7 @@ import Kafka.Consumer.Types (ConsumerRecord (..), Offset (..), PartitionOffset (..), RebalanceEvent (..), Timestamp (..), TopicPartition (..)) import Kafka.Effectful.Consumer.Effect (KafkaConsumer (..)) import Kafka.Types (BatchSize (..), KafkaError (..), PartitionId (..), Timeout (..), TopicName (..))-import Shibuya.Adapter (Adapter (..))+import Shibuya.Adapter (Adapter (..), defaultAdapterTelemetry) import Shibuya.Adapter.Kafka (kafkaRebalanceHandler) import Shibuya.Adapter.Kafka.Config (KafkaAdapterConfig (..)) import Shibuya.Adapter.Kafka.Internal (KafkaAcknowledgementException (..), KafkaAdapterState (..), gateRecords, ingestedStream, kafkaSource, mkAckHandle, mkIngested, newKafkaAdapterState)@@ -294,7 +294,8 @@               Adapter                 { adapterName = "kafka:test-terminal-ack-failure",                   source = Stream.fromList [ingested],-                  shutdown = pure ()+                  shutdown = pure (),+                  telemetry = defaultAdapterTelemetry                 }             processor = mkProcessor adapter (\_ -> pure AckOk)         appResult <- runApp defaultAppConfig [(processorId, processor)]@@ -438,7 +439,8 @@   KafkaAdapterConfig     { topics = [TopicName "orders"],       pollTimeout = Timeout 100,-      batchSize = BatchSize 100+      batchSize = BatchSize 100,+      telemetry = defaultAdapterTelemetry     }  defaultMockState :: MockState

@@ -0,0 +1,158 @@+-- | Trace correlation of Kafka processing spans, broker-free.+--+-- A scripted single-partition log feeds one record carrying a known producer+-- @traceparent@ through 'kafkaAdapter' and Shibuya's runner, with a tracer+-- that exports into memory. The cases inspect the exported Consumer span:+-- its trace, its parent and its links.+module Shibuya.Adapter.Kafka.CorrelationTest (tests) where++import Control.Exception qualified as Exception+import Data.ByteString (ByteString)+import Data.ByteString.Char8 qualified as BS8+import Data.Foldable (toList)+import Data.IORef (newIORef, readIORef)+import Data.Maybe (isNothing)+import Data.Text (Text)+import Effectful (liftIO, runEff)+import Effectful.Error.Static (runErrorNoCallStack)+import Kafka.Consumer.Types (ConsumerRecord (..), Offset (..), Timestamp (..))+import Kafka.ScriptedLog (ScriptedLog (..), runScriptedConsumer, scriptedTopic)+import Kafka.Types (BatchSize (..), KafkaError, PartitionId (..), Timeout (..), headersFromList)+import OpenTelemetry.Attributes (emptyAttributes)+import OpenTelemetry.Exporter.InMemory (inMemoryListExporter)+import OpenTelemetry.Trace.Core qualified as OTel+import OpenTelemetry.Trace.Id (Base (..))+import OpenTelemetry.Trace.Id qualified as OTel.Id+import OpenTelemetry.Trace.Id.Generator (IdGenerator (..))+import OpenTelemetry.Util (appendOnlyBoundedCollectionValues)+import Shibuya.Adapter (Adapter (..))+import Shibuya.Adapter.Kafka+  ( AdapterTelemetry (traceCorrelation),+    KafkaAdapterConfig (..),+    TraceCorrelation (..),+    defaultAdapterTelemetry,+    defaultConfig,+    kafkaAdapter,+  )+import Shibuya.Adapter.Kafka.SourceGateTest (waitUntil)+import Shibuya.App (ProcessorId (..), defaultAppConfig, defaultShutdownConfig, mkProcessor, runApp, stopApp, stopAppGracefully)+import Shibuya.Core.Ack (AckDecision (..))+import Shibuya.Telemetry.Effect (runTracing, runTracingNoop)+import Shibuya.Telemetry.Semantic (processSpanName)+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit (assertBool, assertEqual, assertFailure, testCase)++tests :: TestTree+tests =+  testGroup+    "Trace correlation"+    [ testCase "defaultConfig uses ParentAndLink" $+        assertEqual "default policy" ParentAndLink (defaultConfig [scriptedTopic]).telemetry.traceCorrelation,+      testCase "the adapter exposes a LinkOnly override" $ do+        scripted <- scriptedLog []+        result <-+          runEff . runErrorNoCallStack @KafkaError . runScriptedConsumer scripted . runTracingNoop $ do+            adapter <- kafkaAdapter (configWith linkOnly)+            pure adapter.telemetry.traceCorrelation+        assertEqual "adapter policy" (Right LinkOnly) result,+      testCase "ParentAndLink: process span is a child of the producer with one link" $ do+        [sp] <- processRecord defaultAdapterTelemetry+        assertEqual "span name" (processSpanName processorName) sp.name+        assertEqual "trace" producerTraceHex (hexTrace sp.context)+        assertEqual "parent" (Just producerSpanHex) (hexSpan <$> sp.parent)+        assertEqual "links" [(producerTraceHex, producerSpanHex)] (linkIds sp),+      testCase "LinkOnly: process span is a linked root" $ do+        [sp] <- processRecord linkOnly+        assertBool "new trace" (hexTrace sp.context /= producerTraceHex)+        assertBool "no parent" (isNothing sp.parent)+        assertEqual "links" [(producerTraceHex, producerSpanHex)] (linkIds sp)+    ]++producerParent :: ByteString+producerParent = "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01"++producerTraceHex :: Text+producerTraceHex = "0af7651916cd43dd8448eb211c80319c"++producerSpanHex :: Text+producerSpanHex = "b7ad6b7169203331"++processorName :: Text+processorName = "kafka-correlation"++linkOnly :: AdapterTelemetry+linkOnly = defaultAdapterTelemetry {traceCorrelation = LinkOnly}++configWith :: AdapterTelemetry -> KafkaAdapterConfig+configWith telemetry =+  (defaultConfig [scriptedTopic]) {pollTimeout = Timeout 5, batchSize = BatchSize 100, telemetry = telemetry}++scriptedLog :: [ConsumerRecord (Maybe ByteString) (Maybe ByteString)] -> IO ScriptedLog+scriptedLog records =+  ScriptedLog records <$> newIORef 0 <*> newIORef [] <*> newIORef [] <*> newIORef []++producedRecord :: ConsumerRecord (Maybe ByteString) (Maybe ByteString)+producedRecord =+  ConsumerRecord+    { crTopic = scriptedTopic,+      crPartition = PartitionId 0,+      crOffset = Offset 0,+      crTimestamp = NoTimestamp,+      crHeaders = headersFromList [("traceparent", producerParent)],+      crKey = Nothing,+      crValue = Just (BS8.pack "correlated")+    }++-- | Process the one producer record with the given telemetry and return the+-- exported Consumer spans.+processRecord :: AdapterTelemetry -> IO [ConsumerSpan]+processRecord telemetry = do+  scripted <- scriptedLog [producedRecord]+  (spanProcessor, spansRef) <- inMemoryListExporter+  outcome <-+    Exception.bracket+      (OTel.createTracerProvider [spanProcessor] OTel.emptyTracerProviderOptions {OTel.tracerProviderOptionsIdGenerator = DefaultIdGenerator})+      (\provider -> OTel.shutdownTracerProvider provider (Just 5_000_000))+      $ \provider -> do+        let tracer = OTel.makeTracer provider (OTel.InstrumentationLibrary "kafka-correlation" "" "" emptyAttributes) OTel.tracerOptions+        runEff . runErrorNoCallStack @KafkaError . runScriptedConsumer scripted . runTracing tracer $ do+          adapter <- kafkaAdapter (configWith telemetry)+          app <- runApp defaultAppConfig [(ProcessorId processorName, mkProcessor adapter (\_ -> pure AckOk))]+          case app of+            Left err -> liftIO $ assertFailure (show err)+            Right handle -> do+              liftIO $ waitUntil "record stored" ((== [0]) <$> readIORef scripted.stored)+              drained <- stopAppGracefully defaultShutdownConfig handle+              if drained then pure () else stopApp handle+  assertEqual "app outcome" (Right ()) outcome+  consumerSpans =<< readIORef spansRef++data ConsumerSpan = ConsumerSpan+  { name :: Text,+    context :: OTel.SpanContext,+    parent :: Maybe OTel.SpanContext,+    links :: [OTel.SpanContext]+  }++consumerSpans :: [OTel.ImmutableSpan] -> IO [ConsumerSpan]+consumerSpans exported = traverse describeSpan [sp | sp <- exported, OTel.spanKind sp == OTel.Consumer]+  where+    describeSpan sp = do+      hot <- readIORef (OTel.spanHot sp)+      parent <- traverse OTel.getSpanContext (OTel.spanParent sp)+      pure+        ConsumerSpan+          { name = OTel.hotName hot,+            context = OTel.spanContext sp,+            parent,+            links = map OTel.frozenLinkContext (toList (appendOnlyBoundedCollectionValues (OTel.hotLinks hot)))+          }++linkIds :: ConsumerSpan -> [(Text, Text)]+linkIds sp = [(hexTrace l, hexSpan l) | l <- sp.links]++hexTrace :: OTel.SpanContext -> Text+hexTrace ctx = OTel.Id.traceIdBaseEncodedText Base16 ctx.traceId++hexSpan :: OTel.SpanContext -> Text+hexSpan ctx = OTel.Id.spanIdBaseEncodedText Base16 ctx.spanId

@@ -48,7 +48,7 @@     TopicName (..),   ) import Shibuya.Adapter (Adapter (..))-import Shibuya.Adapter.Kafka (KafkaAdapterConfig (..), kafkaAdapter, kafkaAdapterWith, kafkaRebalanceHandler, newKafkaAdapterState)+import Shibuya.Adapter.Kafka (KafkaAdapterConfig (..), defaultAdapterTelemetry, kafkaAdapter, kafkaAdapterWith, kafkaRebalanceHandler, newKafkaAdapterState) import Shibuya.Adapter.Kafka.Internal (withConsumerLock) import Shibuya.Adapter.Kafka.SourceGateTest (afterYields, await, waitUntil) import Shibuya.App (ProcessorId (..), defaultAppConfig, defaultShutdownConfig, mkProcessor, runApp, stopApp, stopAppGracefully)@@ -185,7 +185,8 @@             KafkaAdapterConfig               { topics = [env.testTopic],                 pollTimeout = Timeout 5000,-                batchSize = BatchSize 100+                batchSize = BatchSize 100,+                telemetry = defaultAdapterTelemetry               }       Adapter {source, shutdown} <- kafkaAdapter config       Stream.fold Fold.drain@@ -214,7 +215,8 @@             KafkaAdapterConfig               { topics = [env.testTopic],                 pollTimeout = Timeout 250,-                batchSize = BatchSize 100+                batchSize = BatchSize 100,+                telemetry = defaultAdapterTelemetry               }       Adapter {source, shutdown} <- kafkaAdapter config       shutdown@@ -481,7 +483,8 @@   KafkaAdapterConfig     { topics = [env.testTopic],       pollTimeout = Timeout 500,-      batchSize = BatchSize 100+      batchSize = BatchSize 100,+      telemetry = defaultAdapterTelemetry     }  countPayload :: ByteString -> [ByteString] -> Int

@@ -16,7 +16,7 @@ import Kafka.ScriptedLog import Kafka.Types (BatchSize (..), KafkaError, PartitionId (..), Timeout (..)) import Shibuya.Adapter (Adapter (..))-import Shibuya.Adapter.Kafka (KafkaAdapterConfig (..), kafkaAdapter, kafkaRebalanceHandler)+import Shibuya.Adapter.Kafka (KafkaAdapterConfig (..), defaultAdapterTelemetry, kafkaAdapter, kafkaRebalanceHandler) import Shibuya.Adapter.Kafka.AckHandleTest hiding (tests) import Shibuya.Adapter.Kafka.Internal (KafkaAdapterState (..), Polled (..), gateRecords, kafkaSource, mkAckHandle, newKafkaAdapterState) import Shibuya.App (ProcessorId (..), defaultAppConfig, defaultShutdownConfig, mkBatchProcessor, mkProcessor, runApp, stopApp, stopAppGracefully)@@ -83,7 +83,7 @@   queued <- newEmptyMVar   outcome <- timeout 30000000 $     runEff . runErrorNoCallStack @KafkaError . runScriptedConsumer scripted . runTracingNoop $ do-      upstream <- kafkaAdapter KafkaAdapterConfig {topics = [scriptedTopic], pollTimeout = Timeout 5, batchSize = BatchSize 100}+      upstream <- kafkaAdapter KafkaAdapterConfig {topics = [scriptedTopic], pollTimeout = Timeout 5, batchSize = BatchSize 100, telemetry = defaultAdapterTelemetry}       let adapter = upstream {source = afterYields 12 queued upstream.source}           handler Message {envelope} = do             let offset = case envelope.cursor of@@ -247,7 +247,7 @@   batches <- newIORef []   outcome <- timeout 10000000 $     runEff . runErrorNoCallStack @KafkaError . runScriptedConsumer scripted . runTracingNoop $ do-      adapter <- kafkaAdapter KafkaAdapterConfig {topics = [scriptedTopic], pollTimeout = Timeout 5, batchSize = BatchSize 100}+      adapter <- kafkaAdapter KafkaAdapterConfig {topics = [scriptedTopic], pollTimeout = Timeout 5, batchSize = BatchSize 100, telemetry = defaultAdapterTelemetry}       let config = defaultBatchConfig {batchSize = 12, batchTimeout = 0.05, tickInterval = Just 0.01}           handler info messages = do             isFirst <- liftIO $ atomicModifyIORef' first (\x -> (False, x))