diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/shibuya-kafka-adapter.cabal b/shibuya-kafka-adapter.cabal --- a/shibuya-kafka-adapter.cabal +++ b/shibuya-kafka-adapter.cabal @@ -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 diff --git a/src/Shibuya/Adapter/Kafka/Convert.hs b/src/Shibuya/Adapter/Kafka/Convert.hs --- a/src/Shibuya/Adapter/Kafka/Convert.hs +++ b/src/Shibuya/Adapter/Kafka/Convert.hs @@ -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 diff --git a/src/Shibuya/Adapter/Kafka/Tracing.hs b/src/Shibuya/Adapter/Kafka/Tracing.hs deleted file mode 100644 --- a/src/Shibuya/Adapter/Kafka/Tracing.hs +++ /dev/null @@ -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 @" "@, -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 @" 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 () diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -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 ] diff --git a/test/Shibuya/Adapter/Kafka/ConvertTest.hs b/test/Shibuya/Adapter/Kafka/ConvertTest.hs --- a/test/Shibuya/Adapter/Kafka/ConvertTest.hs +++ b/test/Shibuya/Adapter/Kafka/ConvertTest.hs @@ -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] diff --git a/test/Shibuya/Adapter/Kafka/TracingTest.hs b/test/Shibuya/Adapter/Kafka/TracingTest.hs deleted file mode 100644 --- a/test/Shibuya/Adapter/Kafka/TracingTest.hs +++ /dev/null @@ -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