kafka-effectful 0.2.0.0 → 0.3.0.0

raw patch · 12 files changed

+461/−187 lines, 12 filesdep ~hs-opentelemetry-apidep ~hs-opentelemetry-exporter-in-memorydep ~hs-opentelemetry-exporter-otlpPVP ok

version bump matches the API change (PVP)

Dependency ranges changed: hs-opentelemetry-api, hs-opentelemetry-exporter-in-memory, hs-opentelemetry-exporter-otlp, hs-opentelemetry-sdk, hs-opentelemetry-semantic-conventions

API changes (from Hackage documentation)

Files

• CHANGELOG.md +29/−0
• README.md +40/−4
• examples/OtelTracing.hs +3/−3
• kafka-effectful.cabal +19/−19
• src/Kafka/Effectful/OpenTelemetry.hs +8/−0
• src/Kafka/Effectful/OpenTelemetry/Consumer/Interpreter.hs +7/−19
• src/Kafka/Effectful/OpenTelemetry/Producer/Interpreter.hs +47/−16
• src/Kafka/Effectful/OpenTelemetry/Propagation.hs +54/−21
• src/Kafka/Effectful/OpenTelemetry/Semantic.hs +124/−23
• test/Kafka/Effectful/OpenTelemetry/PropagationTest.hs +23/−1
• test/Kafka/Effectful/OpenTelemetry/SemanticTest.hs +101/−76
• test/Kafka/Effectful/OpenTelemetry/ShibuyaCompatibilityTest.hs +6/−5

@@ -4,6 +4,35 @@  This package follows the [Haskell Package Versioning Policy](https://pvp.haskell.org/). +## 0.3.0.0 — 2026-05-31++### Breaking Changes++- Upgrade OpenTelemetry support to the `hs-opentelemetry` 1.0 package+  family. The library now requires `hs-opentelemetry-api ^>=1.0`,+  `hs-opentelemetry-sdk ^>=1.0`, 1.0 exporters, and+  `hs-opentelemetry-semantic-conventions >=1.40 && <2`. Downstream+  build plans pinned to the 0.x package family must upgrade.++### New Features++- Add `producerRecordAttributesWith` and `consumerRecordAttributesWith`+  attribute builders that honor the semantic-convention stability mode.+- Add `kafkaHeadersToTextMap` and `textMapToKafkaHeaders` propagation+  helpers built on the OpenTelemetry 1.0 `TextMap` carrier.++### Other Changes++- Align Kafka messaging attributes with the v1.40 semantic-convention+  behavior used by `hs-opentelemetry-instrumentation-hw-kafka-client`+  1.0. Legacy messaging keys remain the default; set+  `OTEL_SEMCONV_STABILITY_OPT_IN=messaging` for stable names or+  `OTEL_SEMCONV_STABILITY_OPT_IN=messaging/dup` to emit both during+  migration.+- Propagation now uses the OpenTelemetry 1.0 `TextMap` carrier+  internally while preserving the existing request-header bridge+  helpers for callers that imported them directly.+ ## 0.2.0.0 — 2026-05-06  Additive release. No breaking changes to existing modules.

@@ -13,6 +13,35 @@ - Resource-safe interpreters that acquire and release Kafka handles via `bracket` - Errors surfaced through `Effectful.Error.Static` (`Error KafkaError`) +## Local Kafka and Jaeger++This repo includes the same Redpanda + Jaeger `process-compose`+setup used by `shibuya-kafka-adapter`. It uses `rpk container`, so+Docker or a compatible Docker socket provider such as Colima must be+running. From a dev shell, start the services in one terminal:++```bash+just process-up+```++Then create the example topics and run the OpenTelemetry tracing+demo from another terminal:++```bash+just create-topics+just otel-example+```++`just otel-example` produces one record to Redpanda, consumes it back,+and prints matching producer and consumer trace IDs. Jaeger receives+OTLP on `localhost:4318`, and its UI is available at+`http://localhost:16686` while services are running. Stop everything+with:++```bash+just process-down+```+ ## Usage  ### Producer@@ -238,11 +267,18 @@ 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`,+populated with the OpenTelemetry messaging semantic conventions.+By default the attributes use the legacy-compatible names emitted by+`hs-opentelemetry`'s Kafka instrumentation (`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+`messaging.kafka.consumer.group`). Set+`OTEL_SEMCONV_STABILITY_OPT_IN=messaging` to emit the stable v1.40+operation and consumer-group names (`messaging.operation.name`,+`messaging.operation.type`, `messaging.consumer.group.name`) instead,+or `OTEL_SEMCONV_STABILITY_OPT_IN=messaging/dup` to emit both old and+stable names while migrating dashboards. 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.

@@ -182,7 +182,7 @@     case producerResult of         Left (_cs, err) -> do             hPutStrLn stderr $ "producer error: " <> show err-            shutdownTracerProvider tp+            _ <- shutdownTracerProvider tp Nothing             exitFailure         Right () -> pure () @@ -196,13 +196,13 @@     case consumerResult of         Left (_cs, err) -> do             hPutStrLn stderr $ "consumer error: " <> show err-            shutdownTracerProvider tp+            _ <- shutdownTracerProvider tp Nothing             exitFailure         Right () -> pure ()      pid <- readIORef producerTraceRef     cid <- readIORef consumerTraceRef-    shutdownTracerProvider tp+    _ <- shutdownTracerProvider tp Nothing     case (pid, cid) of         (Just p, Just c) -> do             putStrLn $ "[otel-tracing] producer trace id: " <> Text.unpack p

@@ -1,6 +1,6 @@ cabal-version:   3.4 name:            kafka-effectful-version:         0.2.0.0+version:         0.3.0.0 synopsis:        Effectful effects for hw-kafka-client description:   Effectful effects and interpreters for hw-kafka-client, a Haskell@@ -56,8 +56,8 @@     , 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+    , hs-opentelemetry-api                   ^>=1.0+    , hs-opentelemetry-semantic-conventions  >=1.40 && <2     , http-types                             >=0.12 && <0.13     , hw-kafka-client                        >=5.3  && <6     , text                                   >=2.0  && <2.2@@ -148,9 +148,9 @@     , 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+    , hs-opentelemetry-api            ^>=1.0+    , hs-opentelemetry-exporter-otlp  ^>=1.0+    , hs-opentelemetry-sdk            ^>=1.0     , hw-kafka-client                 >=5.3  && <6     , kafka-effectful     , text                            >=2.0  && <2.2@@ -184,19 +184,19 @@     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+    , 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                   ^>=1.0+    , hs-opentelemetry-exporter-in-memory    ^>=1.0+    , hs-opentelemetry-sdk                   ^>=1.0+    , hs-opentelemetry-semantic-conventions  >=1.40  && <2+    , 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+    , text                                   >=2.0   && <2.2+    , unordered-containers                   >=0.2   && <0.3

@@ -28,13 +28,17 @@      -- * Attribute helpers     producerRecordAttributes,+    producerRecordAttributesWith,     consumerRecordAttributes,+    consumerRecordAttributesWith,     producerSpanName,     consumerSpanName,      -- * Trace-context propagation     extractTraceContextFromRecord,     injectTraceContextIntoRecord,+    kafkaHeadersToTextMap,+    textMapToKafkaHeaders,     kafkaHeadersToRequestHeaders,     requestHeadersToKafkaHeaders, )@@ -46,11 +50,15 @@     extractTraceContextFromRecord,     injectTraceContextIntoRecord,     kafkaHeadersToRequestHeaders,+    kafkaHeadersToTextMap,     requestHeadersToKafkaHeaders,+    textMapToKafkaHeaders,  ) import Kafka.Effectful.OpenTelemetry.Semantic (     consumerRecordAttributes,+    consumerRecordAttributesWith,     consumerSpanName,     producerRecordAttributes,+    producerRecordAttributesWith,     producerSpanName,  )

@@ -32,7 +32,6 @@ 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)@@ -41,7 +40,7 @@ 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.ConsumerProperties (ConsumerProperties) import Kafka.Consumer.Subscription (Subscription) import Kafka.Consumer.Types (ConsumerRecord (crTopic)) import Kafka.Effectful.Consumer.Effect (KafkaConsumer (..))@@ -49,14 +48,12 @@     extractTraceContextFromRecord,  ) import Kafka.Effectful.OpenTelemetry.Semantic (-    consumerRecordAttributes,+    consumerRecordAttributesWith,     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.SemanticsConfig (getSemanticsOptions, lookupStability) import OpenTelemetry.Trace.Core (     SpanArguments (kind),     SpanKind (Consumer),@@ -189,23 +186,14 @@     Eff es a ->     Eff es a withConsumerSpan tracer props cr action = do+    semOpts <- Effectful.liftIO $ lookupStability "messaging" <$> getSemanticsOptions     inboundCtx <- Effectful.liftIO $ do         currentCtx <- getContext         extractTraceContextFromRecord cr currentCtx     void $ attachContext inboundCtx-    inSpan'' tracer (consumerSpanName (crTopic cr)) spanArgs $ \_span -> action+    inSpan'' tracer (consumerSpanName (crTopic cr)) (spanArgs semOpts) $ \_span -> action   where-    spanArgs =+    spanArgs semOpts =         addAttributesToSpanArguments-            (withConsumerGroup (consumerRecordAttributes cr))+            (consumerRecordAttributesWith semOpts props 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

@@ -32,6 +32,7 @@  import Control.Concurrent.MVar qualified as Concurrent import Data.Foldable (for_)+import Data.Text qualified as Text import Effectful (Eff, IOE, (:>)) import Effectful qualified import Effectful.Dispatch.Dynamic (EffectHandler, interpret)@@ -41,7 +42,7 @@     injectTraceContextIntoRecord,  ) import Kafka.Effectful.OpenTelemetry.Semantic (-    producerRecordAttributes,+    producerRecordAttributesWith,     producerSpanName,  ) import Kafka.Effectful.Producer.Effect (KafkaProducer (..))@@ -50,15 +51,22 @@ import Kafka.Producer.ProducerProperties (ProducerProperties) import Kafka.Transaction qualified as K import Kafka.Types (KafkaError)+import OpenTelemetry.Attributes.Key (unkey) import OpenTelemetry.Context qualified as Context import OpenTelemetry.Context.ThreadLocal (getContext)+import OpenTelemetry.SemanticConventions (error_type)+import OpenTelemetry.SemanticsConfig (getSemanticsOptions, lookupStability) import OpenTelemetry.Trace.Core (+    Span,     SpanArguments (kind),     SpanKind (Producer),+    SpanStatus (Error),     Tracer,+    addAttribute,     addAttributesToSpanArguments,     defaultSpanArguments,     inSpan'',+    setStatus,  )  {- | Run the 'KafkaProducer' effect with OpenTelemetry tracing.@@ -102,32 +110,42 @@     EffectHandler KafkaProducer es handleTracedProducer tracer producer _env = \case     ProduceMessage record ->-        withProducerSpan tracer record $ \instrumentedRecord -> do+        withProducerSpan tracer record $ \span_ instrumentedRecord -> do             mbErr <- Effectful.liftIO $ K.produceMessage producer instrumentedRecord-            for_ mbErr throwError+            for_ mbErr $ \err -> do+                recordKafkaError span_ err+                throwError err     ProduceMessage' record cb ->-        withProducerSpan tracer record $ \instrumentedRecord -> do+        withProducerSpan tracer record $ \span_ instrumentedRecord -> do             res <-                 Effectful.liftIO $                     K.produceMessage' producer instrumentedRecord cb             case res of-                Left (K.ImmediateError err) -> throwError err+                Left (K.ImmediateError err) -> do+                    recordKafkaError span_ err+                    throwError err                 Right () -> pure ()     ProduceMessageSync record ->-        withProducerSpan tracer record $ \instrumentedRecord -> do+        withProducerSpan tracer record $ \span_ 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+                Left (K.ImmediateError err) -> do+                    recordKafkaError span_ 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+                        K.DeliveryFailure _ err -> do+                            recordKafkaError span_ err+                            throwError err+                        K.NoMessageError err -> do+                            recordKafkaError span_ err+                            throwError err     ProduceMessageBatch records -> do         -- One span per record so the Producer-kind attributes         -- (partition, key) are per-record, matching the upstream@@ -136,10 +154,11 @@         results <-             traverse                 ( \r ->-                    withProducerSpan tracer r $ \instrumentedRecord -> do+                    withProducerSpan tracer r $ \span_ instrumentedRecord -> do                         mbErr <-                             Effectful.liftIO $                                 K.produceMessage producer instrumentedRecord+                        for_ mbErr (recordKafkaError span_)                         pure (r, mbErr)                 )                 records@@ -175,17 +194,29 @@     (IOE :> es) =>     Tracer ->     ProducerRecord ->-    (ProducerRecord -> Eff es a) ->+    (Span -> ProducerRecord -> Eff es a) ->     Eff es a-withProducerSpan tracer record action =-    inSpan'' tracer (producerSpanName (prTopic record)) spanArgs $ \newSpan -> do+withProducerSpan tracer record action = do+    semOpts <- Effectful.liftIO $ lookupStability "messaging" <$> getSemanticsOptions+    inSpan'' tracer (producerSpanName (prTopic record)) (spanArgs semOpts) $ \newSpan -> do         ctx <- getContext         instrumentedRecord <-             Effectful.liftIO $                 injectTraceContextIntoRecord (Context.insertSpan newSpan ctx) record-        action instrumentedRecord+        action newSpan instrumentedRecord   where-    spanArgs =+    spanArgs semOpts =         addAttributesToSpanArguments-            (producerRecordAttributes record)+            (producerRecordAttributesWith semOpts record)             defaultSpanArguments{kind = Producer}++recordKafkaError ::+    (IOE :> es) =>+    Span ->+    KafkaError ->+    Eff es ()+recordKafkaError span_ err = do+    let errText = Text.pack (show err)+    Effectful.liftIO $ do+        addAttribute span_ (unkey error_type) errText+        setStatus span_ (Error errText)

@@ -1,17 +1,18 @@-{- | 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.+{- | Bridges between @hw-kafka-client@\'s 'Kafka.Types.Headers' and+OpenTelemetry propagation carriers, plus convenience helpers that+fetch the global propagator and inject\/extract 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).+The traced interpreters use the @hs-opentelemetry-api@ 1.0+'TextMap' carrier. The older @http-types@ 'RequestHeaders' helpers+remain available for users who imported them directly.  @since 0.2.0.0 -} module Kafka.Effectful.OpenTelemetry.Propagation (     -- * Header bridges+    kafkaHeadersToTextMap,+    textMapToKafkaHeaders,     kafkaHeadersToRequestHeaders,     requestHeadersToKafkaHeaders, @@ -23,17 +24,49 @@  import Data.Bifunctor (first) import Data.CaseInsensitive qualified as CI+import Data.Text.Encoding qualified as Text 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,+import OpenTelemetry.Propagator (+    TextMap,+    emptyTextMap,+    extract,+    getGlobalTextMapPropagator,+    inject,+    textMapFromList,+    textMapToList,  ) +{- | Convert @hw-kafka-client@ 'Headers' to the OpenTelemetry 1.0+'TextMap' propagation carrier.++Header names and values are decoded as UTF-8, matching upstream+@hs-opentelemetry-instrumentation-hw-kafka-client@ 1.0.+-}+kafkaHeadersToTextMap :: Headers -> TextMap+kafkaHeadersToTextMap =+    textMapFromList+        . map+            ( \(k, v) ->+                (Text.decodeUtf8 k, Text.decodeUtf8 v)+            )+        . headersToList++{- | Convert the OpenTelemetry 1.0 'TextMap' propagation carrier back+to @hw-kafka-client@ 'Headers'.+-}+textMapToKafkaHeaders :: TextMap -> Headers+textMapToKafkaHeaders =+    headersFromList+        . map+            ( \(k, v) ->+                (Text.encodeUtf8 k, Text.encodeUtf8 v)+            )+        . textMapToList+ {- | Convert @hw-kafka-client@ 'Headers' (case-sensitive) to @http-types@ 'RequestHeaders' (case-insensitive). Each @(bsKey, bsVal)@ becomes @(CI.mk bsKey, bsVal)@.@@ -52,10 +85,10 @@ {- | 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 fetches the global text-map propagator, hands it the record\'s+headers, 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@@ -66,8 +99,8 @@     Context ->     IO Context extractTraceContextFromRecord record ctx = do-    propagator <- getTracerProviderPropagators <$> getGlobalTracerProvider-    extract propagator (kafkaHeadersToRequestHeaders (crHeaders record)) ctx+    propagator <- getGlobalTextMapPropagator+    extract propagator (kafkaHeadersToTextMap (crHeaders record)) ctx  {- | Inject the supplied 'Context'\'s W3C trace context into a 'ProducerRecord'\'s headers, returning the augmented record.@@ -86,7 +119,7 @@     ProducerRecord ->     IO ProducerRecord injectTraceContextIntoRecord ctx record = do-    propagator <- getTracerProviderPropagators <$> getGlobalTracerProvider-    extraHeaders <- inject propagator ctx []-    let merged = prHeaders record <> requestHeadersToKafkaHeaders extraHeaders+    propagator <- getGlobalTextMapPropagator+    extraHeaders <- inject propagator ctx emptyTextMap+    let merged = prHeaders record <> textMapToKafkaHeaders extraHeaders     pure record{prHeaders = merged}

@@ -11,12 +11,11 @@ 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.+semantic conventions v1.40 as exposed by+@hs-opentelemetry-semantic-conventions@. Operation and consumer-group+keys follow @hs-opentelemetry-api@ 1.0\'s stability opt-in policy:+legacy by default, stable with @OTEL_SEMCONV_STABILITY_OPT_IN=messaging@,+and both with @OTEL_SEMCONV_STABILITY_OPT_IN=messaging\/dup@.  The design parallels the upstream @hs-opentelemetry-instrumentation-hw-kafka-client@\'s@@ -36,33 +35,45 @@      -- * Attribute builders     producerRecordAttributes,+    producerRecordAttributesWith,     consumerRecordAttributes,+    consumerRecordAttributesWith, ) where  import Data.ByteString (ByteString)+import Data.ByteString qualified as BS import Data.Int (Int64)+import Data.Map.Strict qualified as Map import Data.Text (Text) import Data.Text.Encoding (decodeUtf8')+import Kafka.Consumer.ConsumerProperties (ConsumerProperties (cpProps)) import Kafka.Consumer.Types (-    ConsumerRecord (crKey, crOffset, crPartition, crTopic),+    ConsumerRecord (crKey, crOffset, crPartition, crTopic, crValue),     Offset (unOffset),  ) import Kafka.Producer.Types (     ProducePartition (SpecifiedPartition, UnassignedPartition),-    ProducerRecord (prKey, prPartition, prTopic),+    ProducerRecord (prKey, prPartition, prTopic, prValue),  ) import Kafka.Types (PartitionId (unPartitionId), TopicName (..)) import OpenTelemetry.Attributes.Attribute (toAttribute) import OpenTelemetry.Attributes.Map (AttributeMap, insertAttributeByKey) import OpenTelemetry.SemanticConventions (+    messaging_client_id,+    messaging_consumer_group_name,     messaging_destination_name,+    messaging_kafka_consumer_group,     messaging_kafka_destination_partition,     messaging_kafka_message_key,     messaging_kafka_message_offset,+    messaging_message_body_size,     messaging_operation,+    messaging_operation_name,+    messaging_operation_type,     messaging_system,  )+import OpenTelemetry.SemanticsConfig (StabilityOpt (..))  {- | The constant @"kafka"@ used as the value of the @messaging.system@ attribute.@@ -111,20 +122,38 @@ is present and decodes as UTF-8. -} producerRecordAttributes :: ProducerRecord -> AttributeMap-producerRecordAttributes record =-    addOperation+producerRecordAttributes = producerRecordAttributesWith Old++{- | Build the OpenTelemetry attribute map for a Producer-kind span+using the supplied messaging semantic-convention stability mode.+-}+producerRecordAttributesWith ::+    StabilityOpt ->+    ProducerRecord ->+    AttributeMap+producerRecordAttributesWith semOpts record =+    addOperation semOpts         . addDestination         . addPartition         . addKey+        . addBodySize         . addSystem         $ mempty   where     addSystem =         insertAttributeByKey messaging_system $             toAttribute kafkaMessagingSystem-    addOperation =-        insertAttributeByKey messaging_operation $-            toAttribute producerOperationName+    addOperation = \case+        Old ->+            insertAttributeByKey messaging_operation $+                toAttribute producerOperationName+        Stable ->+            insertAttributeByKey messaging_operation_name (toAttribute producerOperationName)+                . insertAttributeByKey messaging_operation_type (toAttribute producerOperationName)+        StableAndOld ->+            insertAttributeByKey messaging_operation (toAttribute producerOperationName)+                . insertAttributeByKey messaging_operation_name (toAttribute producerOperationName)+                . insertAttributeByKey messaging_operation_type (toAttribute producerOperationName)     addDestination =         insertAttributeByKey messaging_destination_name $             toAttribute (unTopicName (prTopic record))@@ -138,6 +167,11 @@             insertAttributeByKey messaging_kafka_message_key $                 toAttribute k         Nothing -> id+    addBodySize = case prValue record of+        Just v ->+            insertAttributeByKey messaging_message_body_size $+                toAttribute (fromIntegral (BS.length v) :: Int64)+        Nothing -> id  {- | Build the OpenTelemetry attribute map for a Consumer-kind span describing a single 'ConsumerRecord'.@@ -150,30 +184,92 @@ @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.+Note: consumer-group and client-id attributes need+@Kafka.Consumer.ConsumerProperties@, not just the record. Use+'consumerRecordAttributesWith' when those properties are available. -} consumerRecordAttributes ::     ConsumerRecord (Maybe ByteString) (Maybe ByteString) ->     AttributeMap-consumerRecordAttributes record =-    addOperation+consumerRecordAttributes = consumerRecordAttributesLegacy++consumerRecordAttributesLegacy ::+    ConsumerRecord (Maybe ByteString) (Maybe ByteString) ->+    AttributeMap+consumerRecordAttributesLegacy record =+    addConsumerCommon Old record++{- | Build the OpenTelemetry attribute map for a Consumer-kind span+using the supplied messaging semantic-convention stability mode and+consumer properties.+-}+consumerRecordAttributesWith ::+    StabilityOpt ->+    ConsumerProperties ->+    ConsumerRecord (Maybe ByteString) (Maybe ByteString) ->+    AttributeMap+consumerRecordAttributesWith semOpts props record =+    addConsumerGroup semOpts+        . addClientId+        $ addConsumerCommon semOpts record+  where+    addConsumerGroup = \case+        Old -> addOldConsumerGroup+        Stable -> addStableConsumerGroup+        StableAndOld -> addOldConsumerGroup . addStableConsumerGroup+    addOldConsumerGroup attrs =+        case Map.lookup "group.id" (cpProps props) of+            Just groupId ->+                insertAttributeByKey+                    messaging_kafka_consumer_group+                    (toAttribute groupId)+                    attrs+            Nothing -> attrs+    addStableConsumerGroup attrs =+        case Map.lookup "group.id" (cpProps props) of+            Just groupId ->+                insertAttributeByKey+                    messaging_consumer_group_name+                    (toAttribute groupId)+                    attrs+            Nothing -> attrs+    addClientId attrs =+        case Map.lookup "client.id" (cpProps props) of+            Just clientId ->+                insertAttributeByKey+                    messaging_client_id+                    (toAttribute clientId)+                    attrs+            Nothing -> attrs++addConsumerCommon ::+    StabilityOpt ->+    ConsumerRecord (Maybe ByteString) (Maybe ByteString) ->+    AttributeMap+addConsumerCommon semOpts record =+    addOperation semOpts         . addDestination         . addPartition         . addOffset         . addKey+        . addBodySize         . addSystem         $ mempty   where     addSystem =         insertAttributeByKey messaging_system $             toAttribute kafkaMessagingSystem-    addOperation =-        insertAttributeByKey messaging_operation $-            toAttribute consumerOperationName+    addOperation = \case+        Old ->+            insertAttributeByKey messaging_operation $+                toAttribute consumerOperationName+        Stable ->+            insertAttributeByKey messaging_operation_name (toAttribute consumerOperationName)+                . insertAttributeByKey messaging_operation_type (toAttribute consumerOperationName)+        StableAndOld ->+            insertAttributeByKey messaging_operation (toAttribute consumerOperationName)+                . insertAttributeByKey messaging_operation_name (toAttribute consumerOperationName)+                . insertAttributeByKey messaging_operation_type (toAttribute consumerOperationName)     addDestination =         insertAttributeByKey messaging_destination_name $             toAttribute (unTopicName (crTopic record))@@ -187,6 +283,11 @@         Just k ->             insertAttributeByKey messaging_kafka_message_key $                 toAttribute k+        Nothing -> id+    addBodySize = case crValue record of+        Just v ->+            insertAttributeByKey messaging_message_body_size $+                toAttribute (fromIntegral (BS.length v) :: Int64)         Nothing -> id  {- | Decode message-key bytes as UTF-8, returning 'Nothing' if the

@@ -5,6 +5,7 @@ import Data.CaseInsensitive qualified as CI import Data.Text (Text) import Data.Text qualified as Text+import Data.Text.Encoding qualified as Text.Encoding import Kafka.Consumer.Types (     ConsumerRecord (..),     Offset (..),@@ -14,7 +15,9 @@     extractTraceContextFromRecord,     injectTraceContextIntoRecord,     kafkaHeadersToRequestHeaders,+    kafkaHeadersToTextMap,     requestHeadersToKafkaHeaders,+    textMapToKafkaHeaders,  ) import Kafka.Producer.Types (     ProducePartition (UnassignedPartition),@@ -27,6 +30,7 @@     headersToList,  ) import OpenTelemetry.Context qualified as Context+import OpenTelemetry.Propagator (textMapLookup) import OpenTelemetry.Trace (initializeGlobalTracerProvider) import OpenTelemetry.Trace.Core (     SpanContext (..),@@ -71,7 +75,25 @@     withResource initializeGlobalTracerProvider (\_ -> pure ()) $ \_ ->         testGroup             "Propagation"-            [ testCase "round-trip kafka headers <-> request headers (case-fold)" $ do+            [ testCase "round-trip kafka headers <-> TextMap" $ do+                let original =+                        headersFromList+                            [ ("traceparent", sampleTraceparent)+                            , ("Custom-Header", "value")+                            ]+                    roundTripped =+                        textMapToKafkaHeaders+                            (kafkaHeadersToTextMap original)+                lookup "traceparent" (headersToList roundTripped)+                    `shouldBeJust` sampleTraceparent+                lookup "Custom-Header" (headersToList roundTripped)+                    `shouldBeJust` "value"+            , testCase "kafkaHeadersToTextMap provides case-insensitive lookup" $ do+                let h = headersFromList [("TraceParent", sampleTraceparent)]+                    tm = kafkaHeadersToTextMap h+                textMapLookup "traceparent" tm+                    `shouldBeJust` Text.Encoding.decodeUtf8 sampleTraceparent+            , testCase "request header compatibility helpers still case-fold" $ do                 let original =                         headersFromList                             [ ("traceparent", sampleTraceparent)

@@ -2,23 +2,28 @@  import Data.ByteString (ByteString) import Data.ByteString qualified as BS-import Data.HashMap.Strict qualified as HashMap+import Data.HashMap.Lazy qualified as HashMap import Data.Int (Int64) import Data.Text (Text)+import Kafka.Consumer.ConsumerProperties qualified as ConsumerProperties import Kafka.Consumer.Types (+    ConsumerGroupId (..),     ConsumerRecord (..),     Offset (..),     Timestamp (NoTimestamp),  ) import Kafka.Effectful.OpenTelemetry.Semantic (     consumerRecordAttributes,+    consumerRecordAttributesWith,     producerRecordAttributes,+    producerRecordAttributesWith,  ) import Kafka.Producer.Types (     ProducePartition (SpecifiedPartition, UnassignedPartition),     ProducerRecord (..),  ) import Kafka.Types (+    ClientId (..),     PartitionId (..),     TopicName (..),     headersFromList,@@ -27,8 +32,9 @@     Attribute (AttributeValue),     PrimitiveAttribute (IntAttribute, TextAttribute),  )+import OpenTelemetry.SemanticsConfig (StabilityOpt (..)) import Test.Tasty (TestTree, testGroup)-import Test.Tasty.HUnit (assertEqual, assertFailure, testCase, (@?=))+import Test.Tasty.HUnit (assertFailure, testCase, (@?=))  tests :: TestTree tests =@@ -38,108 +44,111 @@         , 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+        [ testCase "legacy helper keeps old operation key" $ do             let attrs = producerRecordAttributes (sampleProducerRecord (Just "k1"))+            HashMap.lookup "messaging.system" attrs @?= Just (textAttr "kafka")+            HashMap.lookup "messaging.destination.name" attrs @?= Just (textAttr "orders")+            HashMap.lookup "messaging.operation" attrs @?= Just (textAttr "send")+            assertAbsent "messaging.operation.name" attrs+            assertAbsent "messaging.operation.type" attrs+        , testCase "stable mode uses stable operation keys" $ do+            let attrs = producerRecordAttributesWith Stable (sampleProducerRecord (Just "k1"))+            HashMap.lookup "messaging.operation.name" attrs @?= Just (textAttr "send")+            HashMap.lookup "messaging.operation.type" attrs @?= Just (textAttr "send")+            assertAbsent "messaging.operation" attrs+        , testCase "duplicate mode emits old and stable operation keys" $ do+            let attrs = producerRecordAttributesWith StableAndOld (sampleProducerRecord (Just "k1"))+            HashMap.lookup "messaging.operation" attrs @?= Just (textAttr "send")+            HashMap.lookup "messaging.operation.name" attrs @?= Just (textAttr "send")+            HashMap.lookup "messaging.operation.type" attrs @?= Just (textAttr "send")+        , testCase "adds partition, key, and body size when available" $ do+            let attrs = producerRecordAttributesWith Old (sampleProducerRecord (Just "hello"))             HashMap.lookup "messaging.kafka.destination.partition" attrs                 @?= Just (intAttr 42)-        , testCase "omits messaging.kafka.destination.partition for UnassignedPartition" $ do+            HashMap.lookup "messaging.kafka.message.key" attrs+                @?= Just (textAttr "hello")+            HashMap.lookup "messaging.message.body.size" attrs+                @?= Just (intAttr 5)+        , testCase "omits partition for UnassignedPartition" $ do             let attrs =-                    producerRecordAttributes+                    producerRecordAttributesWith+                        Old                         (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+            assertAbsent "messaging.kafka.destination.partition" attrs+        , testCase "omits invalid or absent keys" $ do+            let invalidAttrs =+                    producerRecordAttributesWith+                        Old                         (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)+                missingAttrs =+                    producerRecordAttributesWith Old (sampleProducerRecord Nothing)+            assertAbsent "messaging.kafka.message.key" invalidAttrs+            assertAbsent "messaging.kafka.message.key" missingAttrs         ] --- 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+        [ testCase "legacy helper keeps old operation key" $ do             let attrs = consumerRecordAttributes (sampleConsumerRecord (Just "k1"))+            HashMap.lookup "messaging.system" attrs @?= Just (textAttr "kafka")+            HashMap.lookup "messaging.destination.name" attrs @?= Just (textAttr "orders")+            HashMap.lookup "messaging.operation" attrs @?= Just (textAttr "process")+            assertAbsent "messaging.operation.name" attrs+            assertAbsent "messaging.operation.type" attrs+        , testCase "old mode uses legacy consumer group key" $ do+            let attrs = consumerRecordAttributesWith Old sampleConsumerProps (sampleConsumerRecord (Just "k1"))+            HashMap.lookup "messaging.kafka.consumer.group" attrs+                @?= Just (textAttr "orders-group")+            assertAbsent "messaging.consumer.group.name" attrs+        , testCase "stable mode uses stable operation and consumer group keys" $ do+            let attrs = consumerRecordAttributesWith Stable sampleConsumerProps (sampleConsumerRecord (Just "k1"))+            HashMap.lookup "messaging.operation.name" attrs @?= Just (textAttr "process")+            HashMap.lookup "messaging.operation.type" attrs @?= Just (textAttr "process")+            HashMap.lookup "messaging.consumer.group.name" attrs+                @?= Just (textAttr "orders-group")+            assertAbsent "messaging.operation" attrs+            assertAbsent "messaging.kafka.consumer.group" attrs+        , testCase "duplicate mode emits old and stable operation and group keys" $ do+            let attrs = consumerRecordAttributesWith StableAndOld sampleConsumerProps (sampleConsumerRecord (Just "k1"))+            HashMap.lookup "messaging.operation" attrs @?= Just (textAttr "process")+            HashMap.lookup "messaging.operation.name" attrs @?= Just (textAttr "process")+            HashMap.lookup "messaging.operation.type" attrs @?= Just (textAttr "process")+            HashMap.lookup "messaging.kafka.consumer.group" attrs+                @?= Just (textAttr "orders-group")+            HashMap.lookup "messaging.consumer.group.name" attrs+                @?= Just (textAttr "orders-group")+        , testCase "adds client id, partition, offset, key, and body size" $ do+            let attrs = consumerRecordAttributesWith Old sampleConsumerProps (sampleConsumerRecord (Just "hello"))+            HashMap.lookup "messaging.client.id" attrs+                @?= Just (textAttr "orders-client")             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+            HashMap.lookup "messaging.message.body.size" attrs+                @?= Just (intAttr 5)+        , testCase "omits invalid or absent keys" $ do+            let invalidAttrs =+                    consumerRecordAttributesWith+                        Old+                        sampleConsumerProps                         (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)+                missingAttrs =+                    consumerRecordAttributesWith Old sampleConsumerProps (sampleConsumerRecord Nothing)+            assertAbsent "messaging.kafka.message.key" invalidAttrs+            assertAbsent "messaging.kafka.message.key" missingAttrs         ] --- Helpers- sampleProducerRecord :: Maybe ByteString -> ProducerRecord sampleProducerRecord key =     ProducerRecord@@ -164,8 +173,24 @@         , crValue = Just "value"         } +sampleConsumerProps :: ConsumerProperties.ConsumerProperties+sampleConsumerProps =+    ConsumerProperties.groupId (ConsumerGroupId "orders-group")+        <> ConsumerProperties.clientId (ClientId "orders-client")+ textAttr :: Text -> Attribute textAttr = AttributeValue . TextAttribute  intAttr :: Int64 -> Attribute intAttr = AttributeValue . IntAttribute++assertAbsent :: Text -> HashMap.HashMap Text Attribute -> IO ()+assertAbsent key attrs =+    case HashMap.lookup key attrs of+        Nothing -> pure ()+        Just attr ->+            assertFailure $+                "expected "+                    <> show key+                    <> " to be absent, got "+                    <> show attr

@@ -1,6 +1,6 @@ module Kafka.Effectful.OpenTelemetry.ShibuyaCompatibilityTest (tests) where -import Data.HashMap.Strict qualified as HashMap+import Data.HashMap.Lazy qualified as HashMap import Data.Int (Int64) import Data.Text (Text) import Kafka.Consumer.Types (@@ -22,15 +22,16 @@ import Test.Tasty (TestTree, testGroup) import Test.Tasty.HUnit (assertEqual, testCase) -{- | Tests pinning @Kafka.Effectful.OpenTelemetry.Semantic@\'s-'consumerRecordAttributes' against the attribute set that+{- | Tests pinning @Kafka.Effectful.OpenTelemetry.Semantic@\'s legacy+'consumerRecordAttributes' helper 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.+together with shibuya for the legacy-mode compatibility pin. Stable+OpenTelemetry messaging keys are covered separately in+@SemanticTest@. -} tests :: TestTree tests =