shibuya-kafka-adapter 0.4.0.0 → 0.5.0.0
raw patch · 7 files changed
+112/−374 lines, 7 filesdep +unordered-containersdep ~shibuya-corePVP ok
version bump matches the API change (PVP)
Dependencies added: unordered-containers
Dependency ranges changed: shibuya-core
API changes (from Hackage documentation)
- Shibuya.Adapter.Kafka.Tracing: traced :: forall (es :: [Effect]) v. (Tracing :> es, IOE :> es) => TopicName -> Stream (Eff es) (Ingested es v) -> Stream (Eff es) (Ingested es v)
Files
- CHANGELOG.md +48/−0
- shibuya-kafka-adapter.cabal +4/−4
- src/Shibuya/Adapter/Kafka/Convert.hs +36/−0
- src/Shibuya/Adapter/Kafka/Tracing.hs +0/−139
- test/Main.hs +0/−2
- test/Shibuya/Adapter/Kafka/ConvertTest.hs +24/−0
- test/Shibuya/Adapter/Kafka/TracingTest.hs +0/−229
CHANGELOG.md view
@@ -1,5 +1,53 @@ # Changelog +## 0.5.0.0 — 2026-05-05++### Breaking Changes++- Tracks the `shibuya-core 0.5.0.0` release, which adds an+ `attributes :: !(HashMap Text Attribute)` field to the `Envelope`+ record exported from `Shibuya.Core.Types`. The adapter's+ `consumerRecordToEnvelope` now populates this field with+ Kafka-typed OpenTelemetry attributes:+ `messaging.system="kafka"` (overrides the framework default of+ `"shibuya"`), the typed `messaging.kafka.destination.partition`+ (`Int64`) and `messaging.kafka.message.offset` (`Int64`).+- The opt-in module `Shibuya.Adapter.Kafka.Tracing` is **deleted**.+ Its single export `traced :: TopicName -> Stream (Eff es)+ (Ingested es v) -> Stream (Eff es) (Ingested es v)` opened a+ duplicate Consumer-kind span per message under `runApp`, with+ the typed Kafka attributes split across the two spans (see plan+ 9 of the `shinzui/shibuya` repo, Finding F1). The same job is+ now done by the framework's `processOne`, which reads+ `Envelope.attributes` and emits exactly one span. Callers that+ imported `traced` should remove the import and the+ `traced topic source` step; the rest of the wiring (the+ `runApp` / `runWithMetrics` integration) is unchanged.+- The companion test+ `Shibuya.Adapter.Kafka.TracingTest` is deleted; its assertions+ (span shape, attribute set, ack passthrough) move into the+ framework's `Shibuya.Telemetry.SemanticSpec` upstream and into+ this repo's `Shibuya.Adapter.Kafka.ConvertTest` (the+ attribute-set assertions, against the envelope produced by+ `consumerRecordToEnvelope` in isolation).++### Other Changes++- Bumps the `shibuya-core` build-depends pin to `^>=0.5` in all+ three packages of this repo (`shibuya-kafka-adapter`,+ `shibuya-kafka-adapter-bench`, `shibuya-kafka-adapter-jitsurei`).+- The `shibuya-kafka-adapter-jitsurei/app/OtelDemo.hs` example is+ refactored to drive its message stream through Shibuya's+ `runWithMetrics`, so the framework's `processOne` opens the+ per-message span. The pre-deletion `traced` shape opened a+ sibling span; the new shape emits exactly one Consumer-kind+ span per message, parented on the producer's `traceparent`+ when present, carrying the spec-aligned messaging attributes+ plus the typed `messaging.kafka.*` attributes.+- `unordered-containers ^>=0.2` is now a direct build-depends of+ `shibuya-kafka-adapter` (library and test stanzas) since+ `Convert.hs` constructs the new attribute `HashMap`.+ ## 0.4.0.0 — 2026-04-29 ### Breaking Changes
shibuya-kafka-adapter.cabal view
@@ -1,6 +1,6 @@ cabal-version: 3.12 name: shibuya-kafka-adapter-version: 0.4.0.0+version: 0.5.0.0 synopsis: Kafka adapter for the Shibuya queue processing framework description: A Shibuya adapter that integrates with Apache Kafka via kafka-effectful@@ -39,7 +39,6 @@ Shibuya.Adapter.Kafka.Config Shibuya.Adapter.Kafka.Convert Shibuya.Adapter.Kafka.Internal- Shibuya.Adapter.Kafka.Tracing default-extensions: DeriveAnyClass@@ -62,12 +61,13 @@ , hw-kafka-client >=5.3 && <6 , hw-kafka-streamly ^>=0.1 , kafka-effectful ^>=0.1- , shibuya-core ^>=0.4+ , shibuya-core ^>=0.5 , stm ^>=2.5 , streamly ^>=0.11 , streamly-core ^>=0.3 , text ^>=2.1 , time ^>=1.14+ , unordered-containers ^>=0.2 hs-source-dirs: src default-language: GHC2024@@ -95,7 +95,6 @@ Shibuya.Adapter.Kafka.AdapterTest Shibuya.Adapter.Kafka.ConvertTest Shibuya.Adapter.Kafka.IntegrationTest- Shibuya.Adapter.Kafka.TracingTest build-depends: , async ^>=2.2@@ -118,3 +117,4 @@ , tasty-hunit ^>=0.10 , text , time+ , unordered-containers ^>=0.2
src/Shibuya/Adapter/Kafka/Convert.hs view
@@ -12,7 +12,10 @@ where import Data.ByteString (ByteString)+import Data.HashMap.Strict (HashMap)+import Data.HashMap.Strict qualified as HashMap import Data.Int (Int64)+import Data.Text (Text) import Data.Text qualified as Text import Data.Time (UTCTime) import Data.Time.Clock.POSIX (posixSecondsToUTCTime)@@ -24,6 +27,8 @@ TopicName (..), headersToList, )+import OpenTelemetry.Attributes (Attribute, toAttribute, unkey)+import OpenTelemetry.SemanticConventions qualified as Sem import Shibuya.Core.Types (Cursor (..), Envelope (..), MessageId (..), TraceHeaders) {- | Convert a Kafka 'ConsumerRecord' to a Shibuya 'Envelope'.@@ -36,6 +41,13 @@ * @enqueuedAt@: converted from Kafka timestamp if available * @traceContext@: extracted from @traceparent@/@tracestate@ headers * @attempt@: 'Nothing' (Kafka does not expose a redelivery counter)+* @attributes@: kafka-typed OTel attributes (system, partition,+ offset). The framework's @processOne@ adds these to its+ Consumer-kind span and the @messaging.system@ key overrides+ the framework default of @"shibuya"@. Other @messaging.*@+ attributes (destination.name, operation, message.id) are+ populated by the framework from the @ProcessorId@/@MessageId@+ pair and are not duplicated here. * @payload@: the @crValue@ field (@Maybe ByteString@) -} consumerRecordToEnvelope ::@@ -49,8 +61,32 @@ , enqueuedAt = timestampToUTCTime cr.crTimestamp , traceContext = extractTraceHeaders cr.crHeaders , attempt = Nothing+ , attributes = kafkaSpanAttributes cr.crPartition cr.crOffset , payload = cr.crValue }++{- | OpenTelemetry attributes that the framework's per-message span+should carry for a Kafka-sourced envelope.++* @messaging.system@ — overrides the framework default+ @"shibuya"@ to @"kafka"@.+* @messaging.kafka.destination.partition@ — Int64; from+ @PartitionId@.+* @messaging.kafka.message.offset@ — Int64; from @Offset@.+-}+kafkaSpanAttributes :: PartitionId -> Offset -> HashMap Text Attribute+kafkaSpanAttributes (PartitionId pid) (Offset off) =+ HashMap.fromList+ [ ("messaging.system", toAttribute ("kafka" :: Text))+ ,+ ( unkey Sem.messaging_kafka_destination_partition+ , toAttribute (fromIntegral pid :: Int64)+ )+ ,+ ( unkey Sem.messaging_kafka_message_offset+ , toAttribute (off :: Int64)+ )+ ] -- | Build a globally unique message ID from topic, partition, and offset. mkMessageId :: TopicName -> PartitionId -> Offset -> MessageId
− src/Shibuya/Adapter/Kafka/Tracing.hs
@@ -1,139 +0,0 @@-{- | Opt-in per-message tracing for the Shibuya Kafka adapter.--Exposes a single stream transformer, 'traced', that wraps each 'Ingested'-emitted by 'Shibuya.Adapter.Kafka.kafkaAdapter' so that when a downstream-handler eventually calls the envelope's 'AckHandle.finalize', the call is-enclosed in an OpenTelemetry Consumer-kind span following the messaging-semantic-conventions span-name pattern @"<destination> <operation>"@,-e.g. @"orders process"@ for a topic named @orders@ (see-'Shibuya.Telemetry.Semantic.processSpanName').--The span:--* Is a child of the W3C trace context carried on the envelope, when present.- The parent is recovered via 'Shibuya.Telemetry.Propagation.extractTraceContext'- from @envelope.traceContext@. When absent, the span is a fresh root span.-* Carries the spec-aligned messaging attributes @messaging.system=kafka@,- @messaging.destination.name@, @messaging.operation=process@, and- @messaging.message.id@.-* Carries the Kafka-specific typed attributes- @messaging.kafka.destination.partition@ (Int64, from @envelope.partition@- parsed as an integer) and @messaging.kafka.message.offset@ (Int64, from- @envelope.cursor@ when it is a 'CursorInt'). If the partition text fails- to parse as an integer, the shibuya-namespaced @shibuya.partition@ is- emitted as a defensive fallback.--Typical usage:--@-Adapter{source} <- kafkaAdapter (defaultConfig [TopicName \"orders\"])-Stream.fold Fold.drain- $ Stream.mapM userHandler- $ traced (TopicName \"orders\") source-@--A caller that does not import this module pays nothing: the adapter's public-surface is unchanged and no span is opened.--}-module Shibuya.Adapter.Kafka.Tracing (- traced,-)-where--import Data.Int (Int64)-import Data.Text (Text)-import Data.Text.Read qualified as TR-import Effectful (Eff, IOE, (:>))-import Kafka.Types (TopicName (..))-import OpenTelemetry.Attributes (unkey)-import OpenTelemetry.SemanticConventions qualified as Sem-import OpenTelemetry.Trace.Core (Span)-import Shibuya.Core.AckHandle (AckHandle (..))-import Shibuya.Core.Ingested (Ingested (..))-import Shibuya.Core.Types (Cursor (..), Envelope (..), MessageId (..))-import Shibuya.Telemetry.Effect (- Tracing,- addAttribute,- withExtractedContext,- withSpan',- )-import Shibuya.Telemetry.Propagation (extractTraceContext)-import Shibuya.Telemetry.Semantic (- attrMessagingDestinationName,- attrMessagingMessageId,- attrMessagingOperation,- attrMessagingSystem,- attrShibuyaPartition,- consumerSpanArgs,- processSpanName,- )-import Streamly.Data.Stream (Stream)-import Streamly.Data.Stream qualified as Stream--{- | Rewrite each 'Ingested' so that its 'AckHandle.finalize' opens a-Consumer-kind span parented on the envelope's carried trace context-(when any), named @"<topic> process"@ per-'Shibuya.Telemetry.Semantic.processSpanName', and populated with the-spec-aligned messaging attributes plus the Kafka-specific-@messaging.kafka.*@ attributes.--The topic name is supplied by the caller rather than parsed from the-envelope's 'MessageId' — see the Decision Log of-@docs/plans/9-add-shibuya-kafka-tracing-module.md@ for the rationale.--}-traced ::- forall es v.- (Tracing :> es, IOE :> es) =>- TopicName ->- Stream (Eff es) (Ingested es v) ->- Stream (Eff es) (Ingested es v)-traced (TopicName topicName) =- Stream.mapM $ \ing -> do- let envelope = ing.envelope- AckHandle finalize = ing.ack- parentCtx = envelope.traceContext >>= extractTraceContext- wrappedFinalize decision =- withExtractedContext parentCtx $- withSpan' (processSpanName topicName) consumerSpanArgs $- \sp -> do- populateAttrs sp topicName envelope- finalize decision- pure ing{ack = AckHandle wrappedFinalize}--{- | Set the spec-aligned messaging attribute set on a span, plus the-Kafka-specific typed attributes when the envelope carries them.--}-populateAttrs ::- (Tracing :> es, IOE :> es) =>- Span ->- Text ->- Envelope v ->- Eff es ()-populateAttrs sp topicName envelope = do- -- Generic messaging.* attributes (wire-names from shibuya-core's- -- aligned Semantic module, which derives them from typed AttributeKeys).- addAttribute sp attrMessagingSystem ("kafka" :: Text)- addAttribute sp attrMessagingDestinationName topicName- addAttribute sp attrMessagingOperation ("process" :: Text)- let MessageId msgIdText = envelope.messageId- addAttribute sp attrMessagingMessageId msgIdText-- -- Kafka-specific typed attributes (wire-names from AttributeKey- -- values in OpenTelemetry.SemanticConventions).- case envelope.partition of- Just p -> case TR.decimal p of- Right (n :: Int64, "") ->- addAttribute sp (unkey Sem.messaging_kafka_destination_partition) n- _ ->- -- Defensive: the partition text didn't parse as an int.- -- Emit the shibuya-namespaced fallback so the information- -- is not lost.- addAttribute sp attrShibuyaPartition p- Nothing -> pure ()- case envelope.cursor of- Just (CursorInt off) ->- addAttribute- sp- (unkey Sem.messaging_kafka_message_offset)- (fromIntegral off :: Int64)- _ -> pure ()
test/Main.hs view
@@ -3,7 +3,6 @@ import Shibuya.Adapter.Kafka.AdapterTest qualified as AdapterTest import Shibuya.Adapter.Kafka.ConvertTest qualified as ConvertTest import Shibuya.Adapter.Kafka.IntegrationTest qualified as IntegrationTest-import Shibuya.Adapter.Kafka.TracingTest qualified as TracingTest import Test.Tasty (defaultMain, testGroup) main :: IO ()@@ -14,5 +13,4 @@ [ AdapterTest.tests , ConvertTest.tests , IntegrationTest.tests- , TracingTest.tests ]
test/Shibuya/Adapter/Kafka/ConvertTest.hs view
@@ -1,6 +1,7 @@ module Shibuya.Adapter.Kafka.ConvertTest (tests) where import Data.ByteString (ByteString)+import Data.HashMap.Strict qualified as HashMap import Data.Time.Clock.POSIX (posixSecondsToUTCTime) import Kafka.Consumer.Types (ConsumerRecord (..), Offset (..), Timestamp (..)) import Kafka.Types (@@ -10,6 +11,8 @@ TopicName (..), headersFromList, )+import OpenTelemetry.Attributes (Attribute (..), PrimitiveAttribute (..), unkey)+import OpenTelemetry.SemanticConventions qualified as Sem import Shibuya.Adapter.Kafka.Convert ( consumerRecordToEnvelope, extractTraceHeaders,@@ -84,6 +87,27 @@ cr = mkRecord (TopicName "t") (PartitionId 0) (Offset 0) NoTimestamp hdrs Nothing Nothing env = consumerRecordToEnvelope cr assertEqual "traceContext" (Just [("traceparent", "00-abc-def-01")]) env.traceContext+ , testCase "attributes carry messaging.system=kafka" $ do+ let cr = mkRecord (TopicName "orders") (PartitionId 2) (Offset 42) NoTimestamp mempty Nothing Nothing+ env = consumerRecordToEnvelope cr+ assertEqual+ "messaging.system"+ (Just (AttributeValue (TextAttribute "kafka")))+ (HashMap.lookup "messaging.system" env.attributes)+ , testCase "attributes carry typed messaging.kafka.destination.partition" $ do+ let cr = mkRecord (TopicName "orders") (PartitionId 2) (Offset 42) NoTimestamp mempty Nothing Nothing+ env = consumerRecordToEnvelope cr+ assertEqual+ "messaging.kafka.destination.partition"+ (Just (AttributeValue (IntAttribute 2)))+ (HashMap.lookup (unkey Sem.messaging_kafka_destination_partition) env.attributes)+ , testCase "attributes carry typed messaging.kafka.message.offset" $ do+ let cr = mkRecord (TopicName "orders") (PartitionId 2) (Offset 42) NoTimestamp mempty Nothing Nothing+ env = consumerRecordToEnvelope cr+ assertEqual+ "messaging.kafka.message.offset"+ (Just (AttributeValue (IntAttribute 42)))+ (HashMap.lookup (unkey Sem.messaging_kafka_message_offset) env.attributes) ] traceHeaderTests :: [TestTree]
− test/Shibuya/Adapter/Kafka/TracingTest.hs
@@ -1,229 +0,0 @@-{- | Unit tests for 'Shibuya.Adapter.Kafka.Tracing.traced'.--Drives the 'traced' stream transformer over a single synthetic 'Ingested',-captures the resulting OTel span with an in-memory 'SpanProcessor', and-asserts span shape: trace parenting, attribute set, and that the original-'AckDecision' still flows through the wrapped 'AckHandle.finalize'.--No Kafka broker is required. The processor implementation mirrors-@OpenTelemetry.Exporter.InMemory.Span.inMemoryListExporter@ from-hs-opentelemetry-exporter-in-memory; the latter is not used directly-because its current Hackage release pins @hs-opentelemetry-api <0.3@,-while this repository resolves to 0.3.1.0.--}-module Shibuya.Adapter.Kafka.TracingTest (tests) where--import Control.Concurrent.Async (async)-import Control.Exception (bracket)-import Data.ByteString (ByteString)-import Data.IORef (IORef, atomicModifyIORef, newIORef, readIORef, writeIORef)-import Data.Text (Text)-import Data.Text.Encoding qualified as TE-import Effectful (Eff, IOE, liftIO, runEff, (:>))-import Kafka.Types (TopicName (..))-import OpenTelemetry.Attributes (Attribute (..), PrimitiveAttribute (..), lookupAttribute, unkey)-import OpenTelemetry.Processor.Span (ShutdownResult (..), SpanProcessor (..))-import OpenTelemetry.SemanticConventions qualified as Sem-import OpenTelemetry.Trace.Core (- ImmutableSpan (..),- SpanContext (..),- Tracer,- createTracerProvider,- emptyTracerProviderOptions,- getSpanContext,- makeTracer,- shutdownTracerProvider,- tracerOptions,- )-import OpenTelemetry.Trace.Id (Base (..), spanIdBaseEncodedText, traceIdBaseEncodedText)-import Shibuya.Adapter.Kafka.Tracing (traced)-import Shibuya.Core.Ack (AckDecision (..))-import Shibuya.Core.AckHandle (AckHandle (..))-import Shibuya.Core.Ingested (Ingested (..))-import Shibuya.Core.Types (Cursor (..), Envelope (..), MessageId (..), TraceHeaders)-import Shibuya.Telemetry.Effect (runTracing)-import Streamly.Data.Fold qualified as Fold-import Streamly.Data.Stream qualified as Stream-import Test.Tasty (TestTree, testGroup)-import Test.Tasty.HUnit (Assertion, assertEqual, assertFailure, testCase)--tests :: TestTree-tests =- testGroup- "Tracing"- [ testCase "envelope traceparent becomes span parent" testParentedSpan- , testCase "missing traceContext yields root span" testRootSpan- , testCase "messaging attributes populated from envelope" testAttributes- , testCase "span name follows spec pattern" testSpanName- , testCase "AckDecision threads through wrapped finalize" testAckPassthrough- ]---- | SpanProcessor that records ended spans to an IORef.-recordingProcessor :: IO (SpanProcessor, IORef [ImmutableSpan])-recordingProcessor = do- ref <- newIORef []- let processor =- SpanProcessor- { spanProcessorOnStart = \_ _ -> pure ()- , spanProcessorOnEnd = \spanRef -> do- s <- readIORef spanRef- atomicModifyIORef ref (\l -> (s : l, ()))- , spanProcessorShutdown = async (pure ShutdownSuccess)- , spanProcessorForceFlush = pure ()- }- pure (processor, ref)--withRecordingTracer :: (IORef [ImmutableSpan] -> Tracer -> IO a) -> IO a-withRecordingTracer k = do- (processor, spansRef) <- recordingProcessor- bracket- (createTracerProvider [processor] emptyTracerProviderOptions)- shutdownTracerProvider- $ \tp ->- k spansRef (makeTracer tp "shibuya-kafka-adapter-test" tracerOptions)---- Known-good W3C traceparent with recoverable trace and span ids.-expectedTraceIdHex :: Text-expectedTraceIdHex = "0af7651916cd43dd8448eb211c80319c"--expectedSpanIdHex :: Text-expectedSpanIdHex = "b7ad6b7169203331"--sampleTraceparent :: ByteString-sampleTraceparent =- TE.encodeUtf8 ("00-" <> expectedTraceIdHex <> "-" <> expectedSpanIdHex <> "-01")--mkEnvelope :: Maybe TraceHeaders -> Envelope ()-mkEnvelope tc =- Envelope- { messageId = MessageId "orders-2-42"- , cursor = Just (CursorInt 42)- , partition = Just "2"- , enqueuedAt = Nothing- , traceContext = tc- , attempt = Nothing- , payload = ()- }--mkIngestedFor ::- (IOE :> es) =>- IORef (Maybe AckDecision) ->- Envelope () ->- Ingested es ()-mkIngestedFor ackRef env =- Ingested- { envelope = env- , ack = AckHandle $ \decision -> liftIO $ writeIORef ackRef (Just decision)- , lease = Nothing- }---- | Drive 'traced' over a single envelope, calling finalize with 'AckOk'.-runOneThroughTraced ::- Tracer ->- Envelope () ->- IORef (Maybe AckDecision) ->- IO ()-runOneThroughTraced tracer env ackRef = runEff . runTracing tracer $ do- let ing = mkIngestedFor ackRef env- Stream.fold Fold.drain $- Stream.mapM callFinalize $- traced (TopicName "orders") $- Stream.fromList [ing]- where- callFinalize :: Ingested es () -> Eff es ()- callFinalize i =- let AckHandle fin = i.ack- in fin AckOk--testParentedSpan :: Assertion-testParentedSpan = withRecordingTracer $ \spansRef tracer -> do- ackRef <- newIORef Nothing- runOneThroughTraced tracer (mkEnvelope (Just [("traceparent", sampleTraceparent)])) ackRef- spans <- readIORef spansRef- case spans of- [s] -> do- assertEqual- "span traceId matches parent"- expectedTraceIdHex- (traceIdBaseEncodedText Base16 s.spanContext.traceId)- case s.spanParent of- Just parentSpan -> do- parentCtx <- getSpanContext parentSpan- assertEqual- "parent traceId"- expectedTraceIdHex- (traceIdBaseEncodedText Base16 parentCtx.traceId)- assertEqual- "parent spanId"- expectedSpanIdHex- (spanIdBaseEncodedText Base16 parentCtx.spanId)- Nothing -> assertFailure "expected spanParent to be set"- other -> assertFailure ("expected exactly one span, got " <> show (length other))--testRootSpan :: Assertion-testRootSpan = withRecordingTracer $ \spansRef tracer -> do- ackRef <- newIORef Nothing- runOneThroughTraced tracer (mkEnvelope Nothing) ackRef- spans <- readIORef spansRef- case spans of- [s] -> case s.spanParent of- Nothing -> pure ()- Just _ -> assertFailure "expected root span to have no parent"- other -> assertFailure ("expected exactly one span, got " <> show (length other))--testAttributes :: Assertion-testAttributes = withRecordingTracer $ \spansRef tracer -> do- ackRef <- newIORef Nothing- runOneThroughTraced tracer (mkEnvelope (Just [("traceparent", sampleTraceparent)])) ackRef- spans <- readIORef spansRef- case spans of- [s] -> do- let attrs = s.spanAttributes- -- Generic messaging.* attributes.- assertEqual- "messaging.system"- (Just (AttributeValue (TextAttribute "kafka")))- (lookupAttribute attrs "messaging.system")- assertEqual- "messaging.destination.name"- (Just (AttributeValue (TextAttribute "orders")))- (lookupAttribute attrs "messaging.destination.name")- assertEqual- "messaging.operation"- (Just (AttributeValue (TextAttribute "process")))- (lookupAttribute attrs "messaging.operation")- assertEqual- "messaging.message.id"- (Just (AttributeValue (TextAttribute "orders-2-42")))- (lookupAttribute attrs "messaging.message.id")- -- Kafka-specific typed attributes (Int64 on the wire).- assertEqual- "messaging.kafka.destination.partition"- (Just (AttributeValue (IntAttribute 2)))- (lookupAttribute attrs (unkey Sem.messaging_kafka_destination_partition))- assertEqual- "messaging.kafka.message.offset"- (Just (AttributeValue (IntAttribute 42)))- (lookupAttribute attrs (unkey Sem.messaging_kafka_message_offset))- -- The deleted pre-alignment key must NOT be present.- assertEqual- "messaging.destination.partition.id absent"- Nothing- (lookupAttribute attrs "messaging.destination.partition.id")- other -> assertFailure ("expected exactly one span, got " <> show (length other))--testSpanName :: Assertion-testSpanName = withRecordingTracer $ \spansRef tracer -> do- ackRef <- newIORef Nothing- runOneThroughTraced tracer (mkEnvelope Nothing) ackRef- spans <- readIORef spansRef- case spans of- [s] -> assertEqual "span name" "orders process" s.spanName- other -> assertFailure ("expected exactly one span, got " <> show (length other))--testAckPassthrough :: Assertion-testAckPassthrough = withRecordingTracer $ \_ tracer -> do- ackRef <- newIORef Nothing- runOneThroughTraced tracer (mkEnvelope Nothing) ackRef- decision <- readIORef ackRef- assertEqual "underlying finalize received the decision" (Just AckOk) decision