feat(otel): carry W3C traceparent via Spring AMQP MessageProperties (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

@@ -9,6 +9,24 @@ The envelope wire format is versioned separately by `meta.schema_version`
 
 ## [Unreleased]
 
+### Added
+- **OpenTelemetry `traceparent` transport wiring (ADR-0028, v0.2).**
+  `BabelQueuePublisher` gains `publishWithHeaders(Envelope, Map<String,String>)` — the
+  produce-side seam the optional core `com.babelqueue.otel.HeaderSender` wires to: it sends
+  through a `MessagePostProcessor` that writes out-of-band headers (e.g. a W3C
+  `traceparent`) onto the AMQP `MessageProperties` headers **beside** the contract `x-*`
+  headers (`SpringHeaders.apply`: a contract header always wins a key collision), never
+  inside the frozen envelope (GR-1). New `SpringHeaders.of(Message)` /
+  `of(MessageProperties)` surface a delivered message's headers as a `Map<String,String>`,
+  the consume-side seam a `@RabbitListener` wires to `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 equivalent 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-07
 
 **1.0.0 — the public API is now SemVer-stable**: breaking changes require a MAJOR,

README.md

@@ -108,6 +108,30 @@ Non-conformant messages (missing URN, unsupported `meta.schema_version`, blank
 `trace_id`, missing `data`) raise a `MessageConversionException`, so Spring rejects
 them — route them to a dead-letter exchange the usual Spring AMQP way.
 
+## 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 AMQP `MessageProperties` headers, beside the contract `x-*` headers (a contract header
+always wins a key collision), never inside the frozen envelope (GR-1).
+
+```java
+// produce: HeaderSender -> BabelQueuePublisher.publishWithHeaders
+Tracing.publish(tracer, "urn:babel:orders:created", Map.of("order_id", 1042L), "orders",
+    (envelope, headers) -> babelQueue.publishWithHeaders(envelope, headers));
+
+// consume: take the raw Message too and surface its headers for wrapHandler's Supplier
+@RabbitListener(queues = "orders")
+void onMessage(Envelope envelope, Message message) throws Exception {
+    Tracing.wrapHandler(tracer, h, () -> SpringHeaders.of(message)).handle(envelope);
+}
+```
+
+A header-less `publish(...)` is unchanged; 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>`.
+
 ## Configuration
 
 ```yaml

pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>com.babelqueue</groupId>
     <artifactId>babelqueue-spring</artifactId>
-    <version>1.0.0</version>
+    <version>1.1.0</version>
     <packaging>jar</packaging>
 
     <name>BabelQueue Spring</name>
@@ -50,8 +50,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>
         <spring-boot.version>3.3.5</spring-boot.version>
+        <opentelemetry.version>1.45.0</opentelemetry.version>
     </properties>
 
     <dependencyManagement>
@@ -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>
 
@@ -97,6 +105,23 @@
             <artifactId>spring-boot-starter-test</artifactId>
             <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 Spring AMQP MessageProperties headers) pulls the SDK + in-memory exporter in
+          test scope. Runtime use of this adapter 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/spring/BabelQueuePublisher.java

@@ -4,6 +4,7 @@
 import com.babelqueue.EnvelopeCodec;
 import com.babelqueue.PolyglotMessage;
 import java.util.Map;
+import org.springframework.amqp.core.MessagePostProcessor;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 
 /**
@@ -15,6 +16,11 @@
  * <pre>
  * babelQueue.publish("urn:babel:orders:created", Map.of("order_id", 1042L), "orders");
  * </pre>
+ *
+ * <p>{@link #publishWithHeaders} additionally carries out-of-band transport headers (e.g.
+ * a W3C {@code traceparent}, ADR-0028) on the AMQP {@code MessageProperties} headers,
+ * <b>beside</b> the contract {@code x-*} headers — the produce-side seam the optional core
+ * {@code com.babelqueue.otel.HeaderSender} wires to.
  */
 public class BabelQueuePublisher {
 
@@ -47,6 +53,26 @@ public String publish(String urn, Map<String, Object> data, String queue, String
         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 are written
+     * onto the AMQP {@code MessageProperties} headers <b>beside</b> the contract {@code x-*}
+     * headers ({@link SpringHeaders#apply}: a contract header always wins a key collision),
+     * never inside the frozen envelope (GR-1). This is the produce-side seam the optional
+     * core {@code com.babelqueue.otel.HeaderSender} wires to; with no/empty headers it is
+     * equivalent to a plain publish. Returns {@code meta.id}.
+     */
+    public String publishWithHeaders(Envelope envelope, Map<String, String> headers) {
+        String target = envelope.meta() != null && envelope.meta().queue() != null
+            && !envelope.meta().queue().isBlank() ? envelope.meta().queue() : defaultQueue;
+        MessagePostProcessor addHeaders = message -> {
+            SpringHeaders.apply(message.getMessageProperties(), headers);
+            return message;
+        };
+        rabbitTemplate.convertAndSend(target, (Object) envelope, addHeaders);
+        return envelope.meta().id();
+    }
+
     /** Publish a {@link PolyglotMessage} to the default queue. Returns meta.id. */
     public String publish(PolyglotMessage message) {
         return publish(message, defaultQueue);

src/main/java/com/babelqueue/spring/SpringHeaders.java

@@ -0,0 +1,94 @@
+package com.babelqueue.spring;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageProperties;
+
+/**
+ * The out-of-band transport-header seam for the Spring AMQP adapter (ADR-0028).
+ *
+ * <p>Out-of-band headers (e.g. a W3C {@code traceparent}) ride on the AMQP
+ * {@code MessageProperties} headers <b>beside</b> the contract {@code x-*} headers
+ * ({@code x-schema-version}/{@code x-source-lang}/{@code x-attempts}) that
+ * {@link BabelQueueMessageConverter} stamps — never inside the frozen envelope (GR-1).
+ *
+ * <ul>
+ *   <li><b>Produce</b> ({@link #apply}): {@link BabelQueuePublisher#publishWithHeaders}
+ *       merges the headers onto the message's properties; a contract header already present
+ *       always wins a key collision, and blank keys/values are skipped.</li>
+ *   <li><b>Consume</b> ({@link #of(Message)} / {@link #of(MessageProperties)}): surfaces a
+ *       delivered message's headers as a flat {@code Map<String, String>}, the seam a
+ *       {@code @RabbitListener} wires to the optional core
+ *       {@code com.babelqueue.otel.Tracing#wrapHandler(io.opentelemetry.api.trace.Tracer,
+ *       com.babelqueue.idempotency.Handler, java.util.function.Supplier)} so a carried
+ *       {@code traceparent} makes the consumer span a true child of the producer span.</li>
+ * </ul>
+ *
+ * <pre>{@code
+ * @RabbitListener(queues = "orders")
+ * void onOrder(Envelope env, Message message) throws Exception {
+ *     Tracing.wrapHandler(tracer, h, () -> SpringHeaders.of(message)).handle(env);
+ * }
+ * }</pre>
+ *
+ * <p>Reading or writing the headers requires no OpenTelemetry dependency; the map is a
+ * plain {@code Map<String, String>}.
+ */
+public final class SpringHeaders {
+
+    private SpringHeaders() {}
+
+    /**
+     * Writes the out-of-band {@code headers} onto {@code props} beside the contract
+     * {@code x-*} headers. A blank key or value is skipped, and a header is never written
+     * over a property already set (the contract projection wins). A {@code null}/empty map
+     * is a no-op.
+     */
+    static void apply(MessageProperties props, Map<String, String> headers) {
+        if (props == null || headers == null || headers.isEmpty()) {
+            return;
+        }
+        for (Map.Entry<String, String> e : headers.entrySet()) {
+            String key = e.getKey();
+            String value = e.getValue();
+            if (key == null || key.isEmpty() || value == null || value.isEmpty()) {
+                continue;
+            }
+            if (props.getHeader(key) != null) {
+                continue; // contract / pre-existing header wins
+            }
+            props.setHeader(key, value);
+        }
+    }
+
+    /**
+     * Returns the delivered message's {@code MessageProperties} headers as a
+     * {@code Map<String, String>}. An empty map for a {@code null} message.
+     */
+    public static Map<String, String> of(Message message) {
+        return message == null ? Map.of() : of(message.getMessageProperties());
+    }
+
+    /**
+     * Returns {@code props}' headers as a flat {@code Map<String, String>}
+     * (each value rendered via {@code toString}; blank values dropped). An empty map for
+     * {@code null} props or no headers.
+     */
+    public static Map<String, String> of(MessageProperties props) {
+        Map<String, String> out = new LinkedHashMap<>();
+        if (props == null) {
+            return out;
+        }
+        for (Map.Entry<String, Object> e : props.getHeaders().entrySet()) {
+            Object value = e.getValue();
+            if (value != null) {
+                String str = value.toString();
+                if (!str.isEmpty()) {
+                    out.put(e.getKey(), str);
+                }
+            }
+        }
+        return out;
+    }
+}

src/test/java/com/babelqueue/spring/SpringTraceparentTest.java

@@ -0,0 +1,143 @@
+package com.babelqueue.spring;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+import com.babelqueue.Envelope;
+import com.babelqueue.EnvelopeCodec;
+import io.opentelemetry.api.trace.Tracer;
+import io.opentelemetry.sdk.OpenTelemetrySdk;
+import io.opentelemetry.sdk.testing.exporter.InMemorySpanExporter;
+import io.opentelemetry.sdk.trace.SdkTracerProvider;
+import io.opentelemetry.sdk.trace.data.SpanData;
+import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Map;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessagePostProcessor;
+import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+
+/**
+ * The Spring AMQP wiring of the out-of-band {@code traceparent} header (ADR-0028): produce
+ * writes it onto {@code MessageProperties} headers beside the contract {@code x-*} headers
+ * (contract wins), consume surfaces them as a {@code Map<String,String>}, and a
+ * publish→consume round-trip (converter + post-processor) carries the header end to end so
+ * the consumer span is a true child of the producer span. No broker (the
+ * {@code RabbitTemplate} is mocked; the converter runs in-process).
+ */
+class SpringTraceparentTest {
+
+    private static final String TRACEPARENT = "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01";
+    private final BabelQueueMessageConverter converter = new BabelQueueMessageConverter("default");
+
+    @Test
+    void applyWritesHeadersBesideContractHeadersAndContractWins() {
+        Envelope env = EnvelopeCodec.make("urn:babel:orders:created", Map.of("x", 1L), "orders", "trace-1");
+        Message message = converter.toMessage(env, new MessageProperties()); // stamps x-*
+        MessageProperties props = message.getMessageProperties();
+
+        SpringHeaders.apply(props, Map.of(
+            "traceparent", TRACEPARENT,
+            "x-schema-version", "999")); // collides with a contract header -> must not win
+
+        assertThat((Object) props.getHeader("traceparent")).isEqualTo(TRACEPARENT);
+        assertThat((Object) props.getHeader("x-schema-version")).isEqualTo(1); // contract value preserved
+    }
+
+    @Test
+    void applyIsANoOpForBlankAndEmptyInputs() {
+        MessageProperties props = new MessageProperties();
+        SpringHeaders.apply(props, Map.of());
+        SpringHeaders.apply(props, null);
+        SpringHeaders.apply(null, Map.of("traceparent", TRACEPARENT));
+        SpringHeaders.apply(props, Map.of("", "x")); // blank key skipped
+        assertThat(props.getHeaders()).isEmpty();
+    }
+
+    @Test
+    void ofSurfacesDeliveredHeadersAsAMap() {
+        MessageProperties props = new MessageProperties();
+        props.setHeader("traceparent", TRACEPARENT);
+        props.setHeader("x-schema-version", 1);
+        Message message = new Message("{}".getBytes(StandardCharsets.UTF_8), props);
+
+        Map<String, String> headers = SpringHeaders.of(message);
+        assertThat(headers).containsEntry("traceparent", TRACEPARENT).containsEntry("x-schema-version", "1");
+        assertThat(SpringHeaders.of((Message) null)).isEmpty();
+        assertThat(SpringHeaders.of((MessageProperties) null)).isEmpty();
+    }
+
+    @Test
+    void publishWithHeadersSendsThroughAPostProcessorThatStampsTheHeader() {
+        RabbitTemplate template = mock(RabbitTemplate.class);
+        BabelQueuePublisher publisher = new BabelQueuePublisher(template, "default");
+        Envelope env = EnvelopeCodec.make("urn:babel:orders:created", Map.of("x", 1L), "orders", "trace-1");
+
+        String id = publisher.publishWithHeaders(env, Map.of("traceparent", TRACEPARENT));
+        assertThat(id).isEqualTo(env.meta().id());
+
+        ArgumentCaptor<Object> payload = ArgumentCaptor.forClass(Object.class);
+        ArgumentCaptor<MessagePostProcessor> post = ArgumentCaptor.forClass(MessagePostProcessor.class);
+        verify(template).convertAndSend(eq("orders"), payload.capture(), post.capture());
+
+        // Drive the converter + the captured post-processor to get the final wire message.
+        Message converted = converter.toMessage(payload.getValue(), new MessageProperties());
+        Message finalMsg = post.getValue().postProcessMessage(converted);
+        assertThat((Object) finalMsg.getMessageProperties().getHeader("traceparent")).isEqualTo(TRACEPARENT);
+        // GR-1: traceparent is not in the body; GR-4: trace_id preserved.
+        String body = new String(finalMsg.getBody(), StandardCharsets.UTF_8);
+        assertThat(body).doesNotContain("traceparent");
+        assertThat(EnvelopeCodec.decode(body).traceId()).isEqualTo("trace-1");
+    }
+
+    @Test
+    void endToEndTraceparentMakesConsumerSpanAChildOfTheProducerSpan() throws Exception {
+        InMemorySpanExporter exporter = InMemorySpanExporter.create();
+        SdkTracerProvider provider = SdkTracerProvider.builder()
+            .addSpanProcessor(SimpleSpanProcessor.create(exporter))
+            .build();
+        OpenTelemetrySdk sdk = OpenTelemetrySdk.builder().setTracerProvider(provider).build();
+        Tracer tracer = sdk.getTracer("test");
+
+        RabbitTemplate template = mock(RabbitTemplate.class);
+        BabelQueuePublisher publisher = new BabelQueuePublisher(template, "default");
+
+        // Producer: a PRODUCER span injects traceparent onto the headers via the otel
+        // HeaderSender seam -> BabelQueuePublisher.publishWithHeaders.
+        com.babelqueue.otel.Tracing.publish(
+            tracer, "urn:babel:orders:created", Map.of("order_id", 7), "orders",
+            (envelope, headers) -> publisher.publishWithHeaders(envelope, headers));
+
+        // Reconstruct the delivered AMQP message (converter + post-processor).
+        ArgumentCaptor<Object> payload = ArgumentCaptor.forClass(Object.class);
+        ArgumentCaptor<MessagePostProcessor> post = ArgumentCaptor.forClass(MessagePostProcessor.class);
+        verify(template).convertAndSend(eq("orders"), payload.capture(), post.capture());
+        Message delivered = post.getValue().postProcessMessage(
+            converter.toMessage(payload.getValue(), new MessageProperties()));
+        Envelope env = (Envelope) converter.fromMessage(delivered);
+
+        // Consumer: wrapHandler reads the delivered headers and parents the CONSUMER span on
+        // the carried traceparent.
+        com.babelqueue.otel.Tracing.wrapHandler(tracer, e -> { }, () -> SpringHeaders.of(delivered))
+            .handle(env);
+
+        List<SpanData> spans = exporter.getFinishedSpanItems();
+        SpanData producer = spanByName(spans, "publish urn:babel:orders:created");
+        SpanData consumer = spanByName(spans, "process urn:babel:orders:created");
+        assertThat(consumer.getParentSpanContext().getSpanId()).isEqualTo(producer.getSpanContext().getSpanId());
+        assertThat(consumer.getTraceId()).isEqualTo(producer.getSpanContext().getTraceId());
+        assertThat(consumer.getParentSpanContext().isRemote()).isTrue();
+        provider.close();
+    }
+
+    private static SpanData spanByName(List<SpanData> spans, String name) {
+        return spans.stream().filter(s -> s.getName().equals(name)).findFirst()
+            .orElseThrow(() -> new AssertionError("span not found: " + name));
+    }
+}