Release v1.2.0 (ADR-0038, ADR-0039)

CHANGELOG.md

@@ -9,6 +9,14 @@ The envelope wire format is versioned separately by `meta.schema_version`
 
 ## [Unreleased]
 
+## [1.2.0] - 2026-11-03
+
+### Added
+
+- **Dead-letter observer (ADR-0039 section 5.1).** `AsbConsumer.Builder#deadLetterObserver(DeadLetterObserver)` takes the core's optional `com.babelqueue.DeadLetterObserver` (needs a `babelqueue-core` that ships it). It is called once per dead-letter this consumer carried out, after it is confirmed: a record sent to the BabelQueue DLQ sender (poison body with `reason` `poison`, raw version-gated body with an empty `reason`), and the SDK-initiated native `deadLetter` call made for a version-gated delivery when no DLQ sender is set (`deadLetterQueue` and `reason` empty, as the SDK writes no `dead_letter` block; no extra attribute). `DeadLetterEvent.queue` is the receiver's entity. It is not called when the send or the native call fails, for a broker-initiated dead-letter (`MaxDeliveryCount`), or when the message is only abandoned; an observer failure is logged and never changes the settle outcome. Without an observer nothing changes.
+- **W3C `traceparent` out-of-band propagation (ADR-0028).** `AsbPublisher`: `publishWithHeaders(Envelope, Map<String,String>)` (the produce-side seam for the core's `com.babelqueue.otel.HeaderSender`) and a `publish(..., Map<String,String> headers)` overload carry out-of-band headers (e.g. `traceparent`, `tracestate`) as string `ApplicationProperties` beside the contract `bq-` entries. A contract entry always wins a collision; blank keys/values are skipped; with no/empty headers the message is identical to a plain `publish`. The envelope body is untouched (GR-1, `schema_version` stays 1).
+- `BabelHandler.headersOf(ServiceBusReceivedMessage)` surfaces the delivered string `ApplicationProperties` as a `Map<String,String>` — the `Supplier` source for `Tracing.wrapHandler(tracer, handler, headers)`, so the consumer span becomes a true child of the producer span. No OpenTelemetry dependency is added (the core's OTel API stays optional).
+
 ## [1.1.1] - 2026-10-26
 
 ### Changed

README.md

@@ -15,7 +15,7 @@ propagation) as the .NET, Python, Go and Node SDKs. Implements
 <dependency>
   <groupId>com.babelqueue</groupId>
   <artifactId>babelqueue-azure-servicebus</artifactId>
-  <version>1.1.1</version>
+  <version>1.2.0</version>
 </dependency>
 ```
 

pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>com.babelqueue</groupId>
     <artifactId>babelqueue-azure-servicebus</artifactId>
-    <version>1.1.1</version>
+    <version>1.2.0</version>
     <packaging>jar</packaging>
 
     <name>BabelQueue Azure Service Bus</name>
@@ -49,7 +49,7 @@
         <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.14.0</babelqueue-core.version>
+        <babelqueue-core.version>1.15.0</babelqueue-core.version>
         <azure-servicebus.version>7.17.5</azure-servicebus.version>
         <junit.version>5.10.3</junit.version>
         <mockito.version>5.12.0</mockito.version>

src/main/java/com/babelqueue/azureservicebus/AsbConsumer.java

@@ -7,6 +7,8 @@
 import com.azure.messaging.servicebus.ServiceBusSenderClient;
 import com.azure.messaging.servicebus.models.DeadLetterOptions;
 import com.babelqueue.BabelQueueException;
+import com.babelqueue.DeadLetterEvent;
+import com.babelqueue.DeadLetterObserver;
 import com.babelqueue.Envelope;
 import com.babelqueue.EnvelopeCodec;
 import com.babelqueue.PoisonRecord;
@@ -79,6 +81,7 @@ public interface UnknownUrnHandler {
     private final ErrorHandler onError;
     private final UnknownUrnHandler onUnknownUrn;
     private final ServiceBusSenderClient deadLetterSender;
+    private final DeadLetterObserver deadLetterObserver;
 
     private AsbConsumer(Builder builder) {
         this.receiver = builder.receiver;
@@ -88,6 +91,7 @@ private AsbConsumer(Builder builder) {
         this.onError = builder.onError;
         this.onUnknownUrn = builder.onUnknownUrn;
         this.deadLetterSender = builder.deadLetterSender;
+        this.deadLetterObserver = builder.deadLetterObserver;
     }
 
     public static Builder builder(ServiceBusReceiverClient receiver) {
@@ -131,6 +135,8 @@ private void handle(ServiceBusReceivedMessage message) {
                 receiver.deadLetter(message, new DeadLetterOptions()
                     .setDeadLetterReason(SCHEMA_VERSION_DEAD_LETTER_REASON)
                     .setDeadLetterErrorDescription(description));
+                // The SDK itself dead-lettered it natively (ADR-0039 section 5.1 (2)): counted, with no BabelQueue DLQ.
+                notifyDeadLetter("", "");
                 report(new BabelQueueException(description), null, message);
                 return;
             }
@@ -144,6 +150,7 @@ private void handle(ServiceBusReceivedMessage message) {
                 report(sendFailure, null, message);
                 return;
             }
+            notifyDeadLetter(deadLetterSender.getEntityPath(), "");
             receiver.complete(message);
             report(new BabelQueueException(description), null, message);
             return;
@@ -216,6 +223,7 @@ private void settlePoison(ServiceBusReceivedMessage message, Envelope envelope,
             report(sendFailure, envelope, message);
             return;
         }
+        notifyDeadLetter(deadLetterSender.getEntityPath(), "poison");
         receiver.complete(message);
     }
 
@@ -266,6 +274,19 @@ private static ServiceBusMessage rawMessage(ServiceBusReceivedMessage message) {
         return forwarded;
     }
 
+    /**
+     * Reports a dead-letter the SDK just carried out (ADR-0039 section 5.1): a confirmed send to the BabelQueue DLQ, or
+     * a native {@code deadLetter} call that returned. {@code queue} is the receiver's entity (a version-gated message is
+     * never decoded, so there is no {@code meta.queue}); {@code deadLetterQueue} is empty for the native dead-letter.
+     * An observer failure never escapes.
+     */
+    private void notifyDeadLetter(String deadLetterQueue, String reason) {
+        if (deadLetterObserver == null) {
+            return;
+        }
+        DeadLetterObserver.notifySafely(deadLetterObserver, new DeadLetterEvent(entityName(), deadLetterQueue, reason));
+    }
+
     private void send(ServiceBusMessage message) {
         deadLetterSender.sendMessage(message);
     }
@@ -356,6 +377,7 @@ public static final class Builder {
         private ErrorHandler onError;
         private UnknownUrnHandler onUnknownUrn;
         private ServiceBusSenderClient deadLetterSender;
+        private DeadLetterObserver deadLetterObserver;
 
         private Builder(ServiceBusReceiverClient receiver) {
             this.receiver = Objects.requireNonNull(receiver, "receiver");
@@ -406,6 +428,19 @@ public Builder deadLetterSender(ServiceBusSenderClient sender) {
             return this;
         }
 
+        /**
+         * Optional observer called once per dead-letter this consumer carried out (ADR-0039 section 5.1), after it is
+         * confirmed: a record sent to the BabelQueue DLQ (poison body, raw version-gated body), or the native
+         * {@code deadLetter} call made for a version-gated delivery when no {@link #deadLetterSender sender} is set
+         * ({@code deadLetterQueue} and {@code reason} are then empty: the SDK writes no {@code dead_letter} block). Not called when
+         * the send or the native call fails, or for a broker-initiated dead-letter ({@code MaxDeliveryCount}). A failure
+         * of the observer is logged and never changes the settle outcome.
+         */
+        public Builder deadLetterObserver(DeadLetterObserver deadLetterObserver) {
+            this.deadLetterObserver = deadLetterObserver;
+            return this;
+        }
+
         public AsbConsumer build() {
             return new AsbConsumer(this);
         }

src/main/java/com/babelqueue/azureservicebus/AsbMessages.java

@@ -2,8 +2,13 @@
 
 import com.azure.core.util.BinaryData;
 import com.azure.messaging.servicebus.ServiceBusMessage;
+import com.azure.messaging.servicebus.ServiceBusReceivedMessage;
 import com.babelqueue.Envelope;
 import com.babelqueue.EnvelopeCodec;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
 
 /**
  * Projects the envelope's contract fields onto native Service Bus message fields —
@@ -34,4 +39,47 @@ static ServiceBusMessage project(Envelope envelope) {
         }
         return message;
     }
+
+    /**
+     * Adds the out-of-band {@code extra} headers (e.g. {@code traceparent}, ADR-0028) to {@code message} as string
+     * {@code ApplicationProperties} beside the contract {@code bq-} properties. Blank keys/values are skipped, a
+     * property already present always wins (the contract is never clobbered), and keys are added in sorted order
+     * so the result is deterministic. A {@code null}/empty map leaves the message untouched.
+     */
+    static void addOutOfBand(ServiceBusMessage message, Map<String, String> extra) {
+        if (extra == null || extra.isEmpty()) {
+            return;
+        }
+        Map<String, Object> properties = message.getApplicationProperties();
+        List<String> keys = new ArrayList<>(extra.keySet());
+        keys.removeIf(k -> k == null || k.isEmpty());
+        keys.sort(String::compareTo);
+        for (String key : keys) {
+            String value = extra.get(key);
+            if (value != null && !value.isEmpty() && !properties.containsKey(key)) {
+                properties.put(key, value);
+            }
+        }
+    }
+
+    /**
+     * A delivered message's {@code ApplicationProperties} as a flat {@code Map<String, String>} (string-typed,
+     * non-blank values only) — the consume-side counterpart of {@link #addOutOfBand}. Both the contract
+     * {@code bq-} properties that are strings and out-of-band ones (e.g. {@code traceparent}) surface; numeric
+     * contract properties ({@code bq-schema-version}, {@code bq-created-at}) are not strings and are left out.
+     * Empty when the message is {@code null} or has none.
+     */
+    static Map<String, String> toMap(ServiceBusReceivedMessage message) {
+        Map<String, String> out = new LinkedHashMap<>();
+        Map<String, Object> properties = message == null ? null : message.getApplicationProperties();
+        if (properties == null) {
+            return out;
+        }
+        for (Map.Entry<String, Object> e : properties.entrySet()) {
+            if (e.getKey() != null && e.getValue() instanceof CharSequence value && value.length() > 0) {
+                out.put(e.getKey(), value.toString());
+            }
+        }
+        return out;
+    }
 }

src/main/java/com/babelqueue/azureservicebus/AsbPublisher.java

@@ -22,6 +22,10 @@
  *     .connectionString(cs).sender().queueOrTopicName("orders").buildClient();
  * String id = AsbPublisher.create(sender).publish("urn:babel:orders:created", Map.of("order_id", 1042));
  * }</pre>
+ *
+ * <p>Out-of-band transport headers (e.g. the W3C {@code traceparent}, ADR-0028) ride as extra string
+ * {@code ApplicationProperties} <b>beside</b> the contract {@code bq-} properties — never inside the envelope
+ * (GR-1); see {@link #publishWithHeaders(Envelope, Map)}, the seam for {@code com.babelqueue.otel.HeaderSender}.
  */
 public final class AsbPublisher {
 
@@ -51,13 +55,40 @@ public String publish(String urn, Map<String, Object> data, String traceId) {
      * {@code ScheduledEnqueueTime} (the broker withholds the message until {@code now + delay}).
      */
     public String publish(String urn, Map<String, Object> data, String traceId, Duration delay) {
+        return publish(urn, data, traceId, delay, null);
+    }
+
+    /**
+     * Publish as {@link #publish(String, Map, String, Duration)}, additionally carrying the out-of-band
+     * {@code headers} (e.g. {@code traceparent}) as string {@code ApplicationProperties} beside the contract
+     * {@code bq-} properties. A contract property already present always wins a collision; blank keys or values
+     * are skipped; {@code null}/empty {@code headers} is byte-identical to the plain overload.
+     */
+    public String publish(String urn, Map<String, Object> data, String traceId, Duration delay,
+                          Map<String, String> headers) {
         Envelope envelope = EnvelopeCodec.make(urn, data, sender.getEntityPath(), traceId);
+        return send(envelope, delay, headers);
+    }
+
+    /**
+     * Publish an already-built {@code envelope} together with out-of-band transport {@code headers}
+     * (e.g. a W3C {@code traceparent}, ADR-0028). This is the produce-side seam the optional
+     * {@code com.babelqueue.otel.HeaderSender} wires to. The body stays the byte-identical envelope JSON.
+     * Returns {@code meta.id}.
+     */
+    public String publishWithHeaders(Envelope envelope, Map<String, String> headers) {
+        return send(envelope, null, headers);
+    }
+
+    private String send(Envelope envelope, Duration delay, Map<String, String> headers) {
         ServiceBusMessage message = AsbMessages.project(envelope);
 
         if (delay != null && !delay.isZero() && !delay.isNegative()) {
             message.getApplicationProperties().put("bq-delay", delay.toMillis());
+            AsbMessages.addOutOfBand(message, headers);
             sender.scheduleMessage(message, OffsetDateTime.now(ZoneOffset.UTC).plus(delay));
         } else {
+            AsbMessages.addOutOfBand(message, headers);
             sender.sendMessage(message);
         }
         return envelope.meta().id();

src/main/java/com/babelqueue/azureservicebus/BabelHandler.java

@@ -2,8 +2,23 @@
 
 import com.azure.messaging.servicebus.ServiceBusReceivedMessage;
 import com.babelqueue.Envelope;
+import java.util.Map;
 
-/** Processes one decoded, validated envelope and the raw Service Bus message it arrived on. */
+/**
+ * Processes one decoded, validated envelope and the raw Service Bus message it arrived on.
+ *
+ * <p>The raw message carries the out-of-band transport headers (ADR-0028) as {@code ApplicationProperties}.
+ * {@link #headersOf} surfaces them as a plain map, ready for the optional OpenTelemetry wrapper so the
+ * consumer span becomes a true child of the producer span:
+ *
+ * <pre>{@code
+ * AsbConsumer.builder(receiver)
+ *     .handler(urn, (env, message) -> Tracing
+ *         .wrapHandler(tracer, h, () -> BabelHandler.headersOf(message))
+ *         .handle(env))
+ *     .build();
+ * }</pre>
+ */
 @FunctionalInterface
 public interface BabelHandler {
 
@@ -13,4 +28,14 @@ public interface BabelHandler {
      * incrementing {@code DeliveryCount}).
      */
     void handle(Envelope envelope, ServiceBusReceivedMessage message) throws Exception;
+
+    /**
+     * The message's string-typed {@code ApplicationProperties} as a {@code Map<String, String>} (non-blank values
+     * only) — the delivered-headers {@code Supplier} source for {@code com.babelqueue.otel.Tracing#wrapHandler}.
+     * Contains any out-of-band header (e.g. {@code traceparent}, {@code tracestate}). Empty when {@code message}
+     * is {@code null}. Needs no OpenTelemetry dependency.
+     */
+    static Map<String, String> headersOf(ServiceBusReceivedMessage message) {
+        return AsbMessages.toMap(message);
+    }
 }