feat(otel): carry W3C traceparent via SQS MessageAttributes (ADR-0028)

Implements the babelqueue-core 1.5.0 header seam so traceparent rides the broker
beside the frozen envelope (GR-1), merge-not-clobber + (Redis) bare-value back-
compat; no header falls back to the v0.1 trace_id mapping. Core dep bumped to
1.5.0; package bumped to 1.1.0.

Author: muhammetsafak

CHANGELOG.md

@@ -7,6 +7,26 @@ this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 The envelope wire format is versioned separately by `meta.schema_version`
 (currently **1**) — see the contract at [babelqueue.com](https://babelqueue.com).
 
+## [Unreleased]
+
+### Added
+- **OpenTelemetry `traceparent` transport wiring (ADR-0028, v0.2).** `SqsPublisher`
+  gains `publishWithHeaders(Envelope, Map<String,String>)` — the produce-side seam the
+  optional core `com.babelqueue.otel.HeaderSender` wires to: out-of-band headers (e.g. a
+  W3C `traceparent`) ride on the SQS `MessageAttributes` channel **beside** the contract
+  `bq-*` attributes (`SqsAttributes.projectWithHeaders`: the contract projection wins a
+  key collision, merge in sorted order, bounded by the SQS 10-attribute cap), never inside
+  the frozen envelope (GR-1). New `SqsHeaders.of(Message)` surfaces a delivered message's
+  attributes as a `Map<String,String>`, the consume-side seam for
+  `Tracing.wrapHandler(tracer, handler, Supplier)`, so a carried `traceparent` makes the
+  consumer span a true child of the producer span. A header-less publish is byte-identical
+  to before; `trace_id` is preserved (GR-4); `schema_version` stays **1**. No new runtime
+  dependency — the header seam is a plain `Map<String,String>`; OpenTelemetry is needed
+  only by callers who opt in.
+
+### Changed
+- Require `com.babelqueue:babelqueue-core 1.5.0` (the out-of-band header-carrier seam).
+
 ## [1.0.0] - 2026-06-12
 
 ### Added

README.md

@@ -63,6 +63,32 @@ redelivers it after the visibility timeout (at-least-once). The poll loop never
 a bad message — observe via `onError` / `onUnknownUrn`. The envelope is unchanged
 (`schema_version` stays `1`); SQS is purely additive.
 
+## Trace propagation (OpenTelemetry `traceparent`, ADR-0028)
+
+The optional core `com.babelqueue.otel` module can carry a W3C `traceparent` so a
+consumer span becomes a true child of the producer span — propagated **out of band** on
+the SQS `MessageAttributes` channel, beside the contract `bq-*` attributes (a contract
+attribute always wins a key collision; bounded by the SQS 10-attribute cap), never inside
+the frozen envelope (GR-1).
+
+```java
+// produce: HeaderSender -> SqsPublisher.publishWithHeaders
+SqsPublisher publisher = SqsPublisher.create(sqs, url);
+Tracing.publish(tracer, "urn:babel:orders:created", Map.of("order_id", 1042), "orders",
+    (envelope, headers) -> publisher.publishWithHeaders(envelope, headers));
+
+// consume: surface the delivered attributes for wrapHandler's Supplier
+SqsConsumer.builder(sqs, url)
+    .handler("urn:babel:orders:created", (env, message) ->
+        Tracing.wrapHandler(tracer, h, () -> SqsHeaders.of(message)).handle(env))
+    .build();
+```
+
+A header-less `publish(...)` is byte-identical to before; with no `traceparent` the
+consumer falls back to the v0.1 `trace_id`-derived parent. Requires `babelqueue-core`
+≥ 1.5.0. No OpenTelemetry dependency is needed unless you opt in — the seam is a plain
+`Map<String,String>`.
+
 ## Build & test
 
 ```bash

pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>com.babelqueue</groupId>
     <artifactId>babelqueue-sqs</artifactId>
-    <version>1.0.0</version>
+    <version>1.1.0</version>
     <packaging>jar</packaging>
 
     <name>BabelQueue Amazon SQS</name>
@@ -49,8 +49,9 @@
         <maven.compiler.release>17</maven.compiler.release>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
-        <babelqueue-core.version>1.0.0</babelqueue-core.version>
+        <babelqueue-core.version>1.5.0</babelqueue-core.version>
         <aws-sdk.version>2.25.0</aws-sdk.version>
+        <opentelemetry.version>1.45.0</opentelemetry.version>
         <junit.version>5.10.3</junit.version>
     </properties>
 
@@ -63,6 +64,13 @@
                 <type>pom</type>
                 <scope>import</scope>
             </dependency>
+            <dependency>
+                <groupId>io.opentelemetry</groupId>
+                <artifactId>opentelemetry-bom</artifactId>
+                <version>${opentelemetry.version}</version>
+                <type>pom</type>
+                <scope>import</scope>
+            </dependency>
         </dependencies>
     </dependencyManagement>
 
@@ -92,6 +100,23 @@
             <version>20240303</version>
             <scope>test</scope>
         </dependency>
+        <!--
+          The core's OTel module is optional (not transitive), so the end-to-end
+          traceparent cross-hop test (Tracing.publish HeaderSender -> wrapHandler
+          Supplier over this transport) pulls the SDK + in-memory exporter in test scope.
+          Runtime use of this transport needs no OTel dependency at all — the
+          header seam is a plain Map<String,String>.
+        -->
+        <dependency>
+            <groupId>io.opentelemetry</groupId>
+            <artifactId>opentelemetry-sdk</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.opentelemetry</groupId>
+            <artifactId>opentelemetry-sdk-testing</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>

src/main/java/com/babelqueue/sqs/SqsAttributes.java

@@ -1,16 +1,28 @@
 package com.babelqueue.sqs;
 
 import com.babelqueue.Envelope;
+import java.util.ArrayList;
 import java.util.LinkedHashMap;
+import java.util.List;
 import java.util.Map;
 import software.amazon.awssdk.services.sqs.model.MessageAttributeValue;
 
 /**
  * Projects the envelope's contract fields onto native SQS {@code MessageAttributes} —
  * a redundant, routable view of the body (the body stays authoritative). Contract §3.2.
+ *
+ * <p>It also carries out-of-band transport headers (e.g. the W3C {@code traceparent},
+ * ADR-0028) <b>beside</b> the contract {@code bq-*} attributes on the same channel
+ * {@code bq-trace-id} already rides — never inside the frozen envelope (GR-1). The
+ * contract projection always wins a key collision, and the merge stops at the SQS
+ * 10-attribute ceiling so unbounded out-of-band headers can never crowd out the
+ * contract attributes.
  */
 final class SqsAttributes {
 
+    /** The SQS per-message cap on user {@code MessageAttributes}. */
+    static final int MAX_ATTRIBUTES = 10;
+
     private SqsAttributes() {}
 
     static Map<String, MessageAttributeValue> project(Envelope envelope) {
@@ -26,6 +38,61 @@ static Map<String, MessageAttributeValue> project(Envelope envelope) {
         return attrs;
     }
 
+    /**
+     * Projects the contract attributes and overlays the out-of-band string {@code headers}
+     * beside them — for {@link SqsPublisher#publishWithHeaders}. A blank key or value is
+     * skipped; a contract {@code bq-*} key already present always wins (the header is
+     * dropped, never clobbering it); and once the message reaches {@link #MAX_ATTRIBUTES}
+     * no further header is added. Keys are merged in sorted order so the bounded subset is
+     * deterministic. With a {@code null}/empty map this is byte-identical to {@link #project}.
+     */
+    static Map<String, MessageAttributeValue> projectWithHeaders(
+        Envelope envelope, Map<String, String> headers) {
+        Map<String, MessageAttributeValue> attrs = project(envelope);
+        if (headers == null || headers.isEmpty()) {
+            return attrs;
+        }
+        List<String> keys = new ArrayList<>(headers.keySet());
+        keys.sort(String::compareTo);
+        for (String key : keys) {
+            if (attrs.size() >= MAX_ATTRIBUTES) {
+                break; // respect the SQS 10-attribute ceiling
+            }
+            if (key == null || key.isEmpty() || attrs.containsKey(key)) {
+                continue; // contract attribute wins a collision; skip blank keys
+            }
+            String value = headers.get(key);
+            if (value == null || value.isEmpty()) {
+                continue;
+            }
+            attrs.put(key, MessageAttributeValue.builder().dataType("String").stringValue(value).build());
+        }
+        return attrs;
+    }
+
+    /**
+     * Surfaces a delivered message's {@code MessageAttributes} as a flat
+     * {@code Map<String, String>} (the consume-side counterpart of
+     * {@link #projectWithHeaders}), reading each attribute's {@code stringValue}. Both the
+     * contract {@code bq-*} attributes and out-of-band headers (e.g. {@code traceparent})
+     * surface; the consumer picks the keys it needs. Blank values are dropped; an empty or
+     * {@code null} input yields an empty map.
+     */
+    static Map<String, String> extract(Map<String, MessageAttributeValue> attrs) {
+        Map<String, String> out = new LinkedHashMap<>();
+        if (attrs == null) {
+            return out;
+        }
+        for (Map.Entry<String, MessageAttributeValue> e : attrs.entrySet()) {
+            MessageAttributeValue v = e.getValue();
+            String value = v == null ? null : v.stringValue();
+            if (value != null && !value.isEmpty()) {
+                out.put(e.getKey(), value);
+            }
+        }
+        return out;
+    }
+
     private static void putString(Map<String, MessageAttributeValue> attrs, String key, String value) {
         if (value != null && !value.isEmpty()) {
             attrs.put(key, MessageAttributeValue.builder().dataType("String").stringValue(value).build());

src/main/java/com/babelqueue/sqs/SqsHeaders.java

@@ -0,0 +1,47 @@
+package com.babelqueue.sqs;
+
+import java.util.Map;
+import software.amazon.awssdk.services.sqs.model.Message;
+import software.amazon.awssdk.services.sqs.model.MessageAttributeValue;
+
+/**
+ * Surfaces a delivered SQS message's {@code MessageAttributes} as a flat
+ * {@code Map<String, String>} — the consume-side counterpart of
+ * {@link SqsPublisher#publishWithHeaders}.
+ *
+ * <p>It is the seam a consumer wires to the optional
+ * {@code com.babelqueue.otel.Tracing#wrapHandler(io.opentelemetry.api.trace.Tracer,
+ * com.babelqueue.idempotency.Handler, java.util.function.Supplier)} headers
+ * {@code Supplier}, so a carried W3C {@code traceparent} (ADR-0028) makes the consumer
+ * span a true child of the producer span:
+ *
+ * <pre>{@code
+ * SqsConsumer.builder(sqs, url)
+ *     .handler(urn, (env, message) -> Tracing
+ *         .wrapHandler(tracer, h, () -> SqsHeaders.of(message))
+ *         .handle(env))
+ *     .build();
+ * }</pre>
+ *
+ * <p>The consumer already requests {@code messageAttributeNames("All")}, so every
+ * attribute — both the contract {@code bq-*} projection and any out-of-band header — is
+ * delivered. Reading the headers requires no OpenTelemetry dependency; the map is a plain
+ * {@code Map<String, String>}.
+ */
+public final class SqsHeaders {
+
+    private SqsHeaders() {}
+
+    /**
+     * Returns the message's {@code MessageAttributes} as a {@code Map<String, String>}
+     * (reading each attribute's {@code stringValue}; blank values dropped). An empty map
+     * when the message carries no attributes.
+     */
+    public static Map<String, String> of(Message message) {
+        if (message == null) {
+            return Map.of();
+        }
+        Map<String, MessageAttributeValue> attrs = message.messageAttributes();
+        return SqsAttributes.extract(attrs);
+    }
+}

src/main/java/com/babelqueue/sqs/SqsPublisher.java

@@ -5,6 +5,7 @@
 import java.util.Map;
 import java.util.Objects;
 import software.amazon.awssdk.services.sqs.SqsClient;
+import software.amazon.awssdk.services.sqs.model.MessageAttributeValue;
 import software.amazon.awssdk.services.sqs.model.SendMessageRequest;
 
 /**
@@ -50,18 +51,37 @@ public String publish(String urn, Map<String, Object> data) {
     /** Publish, continuing an existing {@code traceId} (or {@code null} to mint a fresh one). */
     public String publish(String urn, Map<String, Object> data, String traceId) {
         Envelope envelope = EnvelopeCodec.make(urn, data, SqsQueues.nameFromUrl(queueUrl), traceId);
+        send(envelope, SqsAttributes.project(envelope));
+        return envelope.meta().id();
+    }
+
+    /**
+     * Publish an already-built {@code envelope} together with out-of-band transport
+     * {@code headers} (e.g. a W3C {@code traceparent}, ADR-0028). The headers ride on the
+     * SQS {@code MessageAttributes} channel <b>beside</b> the contract {@code bq-*}
+     * attributes ({@link SqsAttributes#projectWithHeaders}: contract wins a collision,
+     * bounded by the 10-attribute cap), never inside the frozen envelope (GR-1). This is
+     * the produce-side seam the optional {@code com.babelqueue.otel.HeaderSender} wires to;
+     * with no/empty headers it is byte-identical to {@link #publish}. Returns
+     * {@code meta.id}.
+     */
+    public String publishWithHeaders(Envelope envelope, Map<String, String> headers) {
+        send(envelope, SqsAttributes.projectWithHeaders(envelope, headers));
+        return envelope.meta().id();
+    }
+
+    private void send(Envelope envelope, Map<String, MessageAttributeValue> attributes) {
         SendMessageRequest.Builder request = SendMessageRequest.builder()
             .queueUrl(queueUrl)
             .messageBody(EnvelopeCodec.encode(envelope))
-            .messageAttributes(SqsAttributes.project(envelope));
+            .messageAttributes(attributes);
         if (fifo) {
             request.messageGroupId(messageGroupId != null ? messageGroupId : SqsQueues.nameFromUrl(queueUrl));
             if (!contentDedup) {
                 request.messageDeduplicationId(envelope.meta().id());
             }
         }
         client.sendMessage(request.build());
-        return envelope.meta().id();
     }
 
     /** Fluent builder for {@link SqsPublisher}. */