shibuya-kafka-adapter 0.2.0.0 → 0.4.0.0
raw patch · 6 files changed
+170/−36 lines, 6 filesdep +hs-opentelemetry-semantic-conventionsdep ~shibuya-corePVP ok
version bump matches the API change (PVP)
Dependencies added: hs-opentelemetry-semantic-conventions
Dependency ranges changed: shibuya-core
API changes (from Hackage documentation)
Files
- CHANGELOG.md +65/−0
- README.md +1/−1
- shibuya-kafka-adapter.cabal +22/−20
- src/Shibuya/Adapter/Kafka/Convert.hs +2/−0
- src/Shibuya/Adapter/Kafka/Tracing.hs +50/−12
- test/Shibuya/Adapter/Kafka/TracingTest.hs +30/−3
CHANGELOG.md view
@@ -1,5 +1,70 @@ # Changelog +## 0.4.0.0 — 2026-04-29++### Breaking Changes++- Tracks the `shibuya-core 0.4.0.0` release, which adds an+ `attempt :: !(Maybe Attempt)` field to the `Envelope` record+ exported from `Shibuya.Core.Types`. The adapter's+ `consumerRecordToEnvelope` now sets this field to `Nothing`,+ consistent with the upstream guidance that adapters which cannot+ observe broker-side redeliveries (e.g. Kafka) report `Nothing`.+ Downstream code that pattern-matches on `Envelope` with positional+ patterns or with non-punned record patterns that name every field+ must be updated.++### Other Changes++- Bumps the `shibuya-core` build-depends pin to `^>=0.4` in all three+ packages of this repo.+- Drops three orphan `NFData` instances (`MessageId`, `Cursor`,+ `Envelope a`) from `shibuya-kafka-adapter-bench/bench/Main.hs`;+ these instances have been provided upstream by `shibuya-core` since+ `0.2.0.0` and the orphans had become duplicate-instance hazards+ whenever the bench resolved against a `shibuya-core` newer than+ `0.1`.+- `shibuya-kafka-adapter-bench` and `shibuya-kafka-adapter-jitsurei`+ are re-released at `0.4.0.0` to track the shared version of this+ repo; neither has user-visible changes of its own.++## 0.3.0.0 — 2026-04-22++Telemetry wire-format change. No Haskell API break —+`Shibuya.Adapter.Kafka.Tracing.traced`'s signature is unchanged — but+operators with dashboards filtering on the old attribute-key strings+or span name must update their queries.++### Changed++- Per-message spans now follow the OpenTelemetry messaging+ semantic-conventions span-name pattern `"<destination> <operation>"`,+ yielding e.g. `"orders process"` in place of the previous constant+ `"shibuya.process.message"`.+- The `messaging.operation` attribute is now set to `"process"` on+ every consumer span.+- The Kafka partition is now emitted as the typed Kafka-specific key+ `messaging.kafka.destination.partition` (Int64), replacing the+ never-defined `messaging.destination.partition.id` (Text). If the+ envelope's partition text does not parse as an integer, the+ shibuya-namespaced `shibuya.partition` is emitted as a defensive+ fallback.+- The Kafka offset is now emitted as `messaging.kafka.message.offset`+ (Int64), derived from `Envelope.cursor` when it is a `CursorInt`.++### Aligned with++- `Shibuya.Telemetry.Semantic` as of sibling `shibuya` plan 2+ (`docs/plans/2-align-opentelemetry-semantic-conventions.md` in the+ shibuya repo). Attribute keys for the generic `messaging.*`+ namespace are sourced from that module, which in turn derives them+ from typed `AttributeKey` values in+ `OpenTelemetry.SemanticConventions`.+- `OpenTelemetry.SemanticConventions` (new direct `build-depends`)+ for the typed Kafka-specific keys+ `messaging_kafka_destination_partition` and+ `messaging_kafka_message_offset`.+ ## 0.2.0.0 — 2026-04-18 Additive release. Adds one new exposed module, no changes to existing
README.md view
@@ -12,7 +12,7 @@ ## Tracing (opt-in) -`Shibuya.Adapter.Kafka.Tracing.traced` is an opt-in stream transformer that wraps each emitted `Ingested` so that the downstream handler's eventual `finalize` call runs inside a Consumer-kind `shibuya.process.message` OpenTelemetry span. The span inherits the envelope's W3C `traceparent` as parent (from `Envelope.traceContext`) or opens a fresh root span when no parent is present, and is populated with the v1.27 messaging-conventions attributes (`messaging.system=kafka`, `messaging.destination.name`, `messaging.message.id`, and `messaging.destination.partition.id` when the partition is known). A caller that does not import this module pays nothing — no spans are opened and the adapter's public surface is unchanged.+`Shibuya.Adapter.Kafka.Tracing.traced` is an opt-in stream transformer that wraps each emitted `Ingested` so that the downstream handler's eventual `finalize` call runs inside a Consumer-kind OpenTelemetry span named following the messaging convention `"<destination> <operation>"` — e.g. `"orders process"` for a topic named `orders`. The span inherits the envelope's W3C `traceparent` as parent (from `Envelope.traceContext`) or opens a fresh root span when no parent is present, and is populated with the spec-aligned messaging attributes (`messaging.system=kafka`, `messaging.destination.name`, `messaging.operation=process`, `messaging.message.id`) and the Kafka-specific typed attributes `messaging.kafka.destination.partition` (Int64) and `messaging.kafka.message.offset` (Int64) when available. A caller that does not import this module pays nothing — no spans are opened and the adapter's public surface is unchanged. Typical wiring:
shibuya-kafka-adapter.cabal view
@@ -1,6 +1,6 @@ cabal-version: 3.12 name: shibuya-kafka-adapter-version: 0.2.0.0+version: 0.4.0.0 synopsis: Kafka adapter for the Shibuya queue processing framework description: A Shibuya adapter that integrates with Apache Kafka via kafka-effectful@@ -53,20 +53,21 @@ QuasiQuotes build-depends:- , base ^>=4.21.0.0- , bytestring ^>=0.12- , containers ^>=0.7- , effectful-core ^>=2.6.1.0- , hs-opentelemetry-api ^>=0.3- , hw-kafka-client >=5.3 && <6- , hw-kafka-streamly ^>=0.1- , kafka-effectful ^>=0.1- , shibuya-core ^>=0.1.0.0- , stm ^>=2.5- , streamly ^>=0.11- , streamly-core ^>=0.3- , text ^>=2.1- , time ^>=1.14+ , base ^>=4.21.0.0+ , bytestring ^>=0.12+ , containers ^>=0.7+ , effectful-core ^>=2.6.1.0+ , hs-opentelemetry-api ^>=0.3+ , hs-opentelemetry-semantic-conventions ^>=0.1+ , hw-kafka-client >=5.3 && <6+ , hw-kafka-streamly ^>=0.1+ , kafka-effectful ^>=0.1+ , shibuya-core ^>=0.4+ , stm ^>=2.5+ , streamly ^>=0.11+ , streamly-core ^>=0.3+ , text ^>=2.1+ , time ^>=1.14 hs-source-dirs: src default-language: GHC2024@@ -97,12 +98,13 @@ Shibuya.Adapter.Kafka.TracingTest build-depends:- , async ^>=2.2- , base ^>=4.21.0.0+ , async ^>=2.2+ , base ^>=4.21.0.0 , bytestring , containers , effectful-core- , hs-opentelemetry-api ^>=0.3+ , hs-opentelemetry-api ^>=0.3+ , hs-opentelemetry-semantic-conventions ^>=0.1 , hw-kafka-client , kafka-effectful , process@@ -112,7 +114,7 @@ , stm , streamly , streamly-core- , tasty ^>=1.5- , tasty-hunit ^>=0.10+ , tasty ^>=1.5+ , tasty-hunit ^>=0.10 , text , time
src/Shibuya/Adapter/Kafka/Convert.hs view
@@ -35,6 +35,7 @@ * @partition@: @Just (show partitionId)@ * @enqueuedAt@: converted from Kafka timestamp if available * @traceContext@: extracted from @traceparent@/@tracestate@ headers+* @attempt@: 'Nothing' (Kafka does not expose a redelivery counter) * @payload@: the @crValue@ field (@Maybe ByteString@) -} consumerRecordToEnvelope ::@@ -47,6 +48,7 @@ , partition = Just (Text.pack (show (unPartitionId cr.crPartition))) , enqueuedAt = timestampToUTCTime cr.crTimestamp , traceContext = extractTraceHeaders cr.crHeaders+ , attempt = Nothing , payload = cr.crValue }
src/Shibuya/Adapter/Kafka/Tracing.hs view
@@ -3,17 +3,25 @@ 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 named-@shibuya.process.message@ (see 'Shibuya.Telemetry.Semantic.processMessageSpanName').+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 v1.27 messaging-conventions attribute set:- @messaging.system@, @messaging.destination.name@, @messaging.message.id@,- and — when a partition is known — @messaging.destination.partition.id@.+* 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: @@ -32,13 +40,17 @@ ) 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 (Envelope (..), MessageId (..))+import Shibuya.Core.Types (Cursor (..), Envelope (..), MessageId (..)) import Shibuya.Telemetry.Effect ( Tracing, addAttribute,@@ -48,18 +60,22 @@ import Shibuya.Telemetry.Propagation (extractTraceContext) import Shibuya.Telemetry.Semantic ( attrMessagingDestinationName,- attrMessagingDestinationPartitionId, attrMessagingMessageId,+ attrMessagingOperation, attrMessagingSystem,+ attrShibuyaPartition, consumerSpanArgs,- processMessageSpanName,+ 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) and populated with the v1.27 messaging attributes.+(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@@ -78,13 +94,15 @@ parentCtx = envelope.traceContext >>= extractTraceContext wrappedFinalize decision = withExtractedContext parentCtx $- withSpan' processMessageSpanName consumerSpanArgs $+ withSpan' (processSpanName topicName) consumerSpanArgs $ \sp -> do populateAttrs sp topicName envelope finalize decision pure ing{ack = AckHandle wrappedFinalize} --- | Set the messaging-conventions v1.27 attribute set on a span.+{- | 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 ->@@ -92,10 +110,30 @@ 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 -> addAttribute sp attrMessagingDestinationPartitionId p+ 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/Shibuya/Adapter/Kafka/TracingTest.hs view
@@ -21,8 +21,9 @@ import Data.Text.Encoding qualified as TE import Effectful (Eff, IOE, liftIO, runEff, (:>)) import Kafka.Types (TopicName (..))-import OpenTelemetry.Attributes (Attribute (..), PrimitiveAttribute (..), lookupAttribute)+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 (..),@@ -53,6 +54,7 @@ [ 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 ] @@ -99,6 +101,7 @@ , partition = Just "2" , enqueuedAt = Nothing , traceContext = tc+ , attempt = Nothing , payload = () } @@ -176,6 +179,7 @@ case spans of [s] -> do let attrs = s.spanAttributes+ -- Generic messaging.* attributes. assertEqual "messaging.system" (Just (AttributeValue (TextAttribute "kafka")))@@ -185,13 +189,36 @@ (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.destination.partition.id"- (Just (AttributeValue (TextAttribute "2")))+ "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