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).** `ArtemisConsumer.Builder#deadLetterObserver(DeadLetterObserver)` takes the core's optional `com.babelqueue.DeadLetterObserver` (needs a `babelqueue-core` that ships it). It is called once per record written to the DLQ, right after the send succeeded: an exhausted retry (`failed`), an unmapped URN under `dead_letter` (`unknown_urn`), a poison body (`poison`) and a raw version-gated body (empty `reason`). `DeadLetterEvent.queue` is `meta.queue` when the envelope carries one, else the consumed queue. It is not called when the write fails, when no DLQ is configured, under `delete`, or for a broker-side dead-letter; an observer failure is logged and never changes the settle outcome. Without an observer nothing changes.
+- **W3C `traceparent` out-of-band propagation (ADR-0028).** `ArtemisPublisher`: `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 JMS string properties beside the contract `bq_` entries. A contract entry always wins a collision; blank keys/values and JMS-illegal property names (e.g. hyphenated, `JMS*`) 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(jakarta.jms.Message)` surfaces the delivered JMS string properties 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

@@ -32,7 +32,7 @@ them instead of re-implementing them (the envelope stays `schema_version: 1`):
 <dependency>
   <groupId>com.babelqueue</groupId>
   <artifactId>babelqueue-artemis</artifactId>
-  <version>1.1.1</version>
+  <version>1.2.0</version>
 </dependency>
 ```
 

pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>com.babelqueue</groupId>
     <artifactId>babelqueue-artemis</artifactId>
-    <version>1.1.1</version>
+    <version>1.2.0</version>
     <packaging>jar</packaging>
 
     <name>BabelQueue Apache ActiveMQ Artemis</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>
         <jakarta.jms.version>3.1.0</jakarta.jms.version>
         <junit.version>5.10.3</junit.version>
         <mockito.version>5.12.0</mockito.version>

src/main/java/com/babelqueue/artemis/ArtemisConsumer.java

@@ -1,6 +1,8 @@
 package com.babelqueue.artemis;
 
 import com.babelqueue.BabelQueueException;
+import com.babelqueue.DeadLetterEvent;
+import com.babelqueue.DeadLetterObserver;
 import com.babelqueue.DeadLetters;
 import com.babelqueue.Envelope;
 import com.babelqueue.EnvelopeCodec;
@@ -54,6 +56,11 @@
  * {@code dead_letter}, a handler that exhausted max-tries): otherwise the session is {@code recover()}ed and the
  * failure goes to {@code onError}.
  *
+ * <p>An optional {@link Builder#deadLetterObserver DeadLetterObserver} (ADR-0039 section 5.1) is called once per record
+ * this consumer wrote to the DLQ, right after the send succeeded (exhausted retry, unmapped URN under
+ * {@code dead_letter}, poison body, raw version-gated body). It is not called when the write fails, when no DLQ is
+ * configured, or for a broker-side dead-letter; an observer failure never changes the settle outcome.
+ *
  * <p>Process one message per {@link #poll()} so the per-message {@code acknowledge()} and
  * {@code recover()} stay correct under the session-wide ack model. Not thread-safe — a JMS
  * session is single-threaded; run one per thread.
@@ -80,6 +87,7 @@ public interface ErrorHandler {
     private final String unknownUrn;
     private final ErrorHandler onError;
     private final long receiveTimeoutMillis;
+    private final DeadLetterObserver deadLetterObserver;
 
     private ArtemisConsumer(Builder builder) {
         this.consumer = builder.consumer;
@@ -91,6 +99,7 @@ private ArtemisConsumer(Builder builder) {
         this.unknownUrn = builder.unknownUrn;
         this.onError = builder.onError;
         this.receiveTimeoutMillis = builder.receiveTimeoutMillis;
+        this.deadLetterObserver = builder.deadLetterObserver;
     }
 
     public static Builder builder(MessageConsumer consumer, Session session) {
@@ -277,6 +286,7 @@ envelope, reason, originalQueue(envelope), envelope.attempts(),
             error == null ? null : error.getMessage(),
             error == null ? null : error.getClass().getName());
         dlqProducer.send(session.createQueue(dlqQueue), JmsMessages.project(session, annotated));
+        notifyDeadLetter(() -> sourceQueue(envelope, message), reason);
     }
 
     /** Whether dead-lettering is on: a DLQ producer and queue name are configured. */
@@ -289,6 +299,7 @@ private void deadLetterPoison(Message message, Throwable cause) throws JMSExcept
         String record = poisonRecord(JmsMessages.bodyBytes(message),
             consumedQueue(message), Math.max(JmsMessages.deliveryCount(message) - 1, 0), cause);
         dlqProducer.send(session.createQueue(dlqQueue), session.createTextMessage(record));
+        notifyDeadLetter(() -> consumedQueue(message), "poison");
     }
 
     /**
@@ -329,6 +340,7 @@ private void deadLetterRaw(Message message) throws JMSException {
             forwarded = session.createTextMessage(new String(body, StandardCharsets.UTF_8));
         }
         dlqProducer.send(session.createQueue(dlqQueue), forwarded);
+        notifyDeadLetter(() -> consumedQueue(message), "");
     }
 
     /**
@@ -351,6 +363,20 @@ private static String urn(Message message, Envelope envelope) {
         return type != null && !type.isEmpty() ? type : EnvelopeCodec.urn(envelope);
     }
 
+    /** {@code meta.queue} when the envelope carries one, else the queue the message was consumed from. */
+    private String sourceQueue(Envelope envelope, Message message) {
+        String queue = originalQueue(envelope);
+        return EnvelopeCodec.isBlankAscii(queue) ? consumedQueue(message) : queue;
+    }
+
+    /** Reports a DLQ write the broker just accepted (ADR-0039 section 5.1); an observer failure never escapes. */
+    private void notifyDeadLetter(java.util.function.Supplier<String> queue, String reason) {
+        if (deadLetterObserver == null) {
+            return;
+        }
+        DeadLetterObserver.notifySafely(deadLetterObserver, new DeadLetterEvent(queue.get(), dlqQueue, reason));
+    }
+
     private static String originalQueue(Envelope envelope) {
         return envelope.meta() != null && envelope.meta().queue() != null ? envelope.meta().queue() : "";
     }
@@ -372,6 +398,7 @@ public static final class Builder {
         private String unknownUrn = UnknownUrnStrategy.FAIL;
         private ErrorHandler onError;
         private long receiveTimeoutMillis = 1000;
+        private DeadLetterObserver deadLetterObserver;
 
         private Builder(MessageConsumer consumer, Session session) {
             this.consumer = Objects.requireNonNull(consumer, "consumer");
@@ -417,6 +444,15 @@ public Builder receiveTimeout(long millis) {
             return this;
         }
 
+        /**
+         * Optional observer called once per record written to the DLQ, after the send succeeded (ADR-0039 section
+         * 5.1). Failures of the observer are logged and never change the settle outcome.
+         */
+        public Builder deadLetterObserver(DeadLetterObserver deadLetterObserver) {
+            this.deadLetterObserver = deadLetterObserver;
+            return this;
+        }
+
         public ArtemisConsumer build() {
             return new ArtemisConsumer(this);
         }

src/main/java/com/babelqueue/artemis/ArtemisPublisher.java

@@ -20,7 +20,13 @@
  * ({@code schema_version} stays 1); Artemis is purely additive.
  *
  * <p>A positive delay uses native JMS&nbsp;2.0 scheduled delivery ({@code setDeliveryDelay});
- * {@code bq-delay} is set for observability.
+ * {@code bq_delay} is set for observability.
+ *
+ * <p>Out-of-band transport headers (e.g. the W3C {@code traceparent}, ADR-0028) ride as extra JMS string
+ * properties <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}. Only
+ * JMS-legal property names are carried ({@code traceparent} and {@code tracestate} are); an illegal name
+ * (e.g. one with a hyphen) is skipped, degrading to {@code trace_id} correlation with no error.
  */
 public final class ArtemisPublisher {
 
@@ -49,18 +55,45 @@ public String publish(String urn, Map<String, Object> data, String traceId) thro
 
     /** Publish with an optional relative {@code delay} (native JMS 2.0 {@code setDeliveryDelay}). */
     public String publish(String urn, Map<String, Object> data, String traceId, Duration delay) throws JMSException {
+        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 JMS string properties beside the contract {@code bq_}
+     * properties. A contract property already present always wins a collision; blank or JMS-illegal names and
+     * blank 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) throws JMSException {
         Envelope envelope = EnvelopeCodec.make(urn, data, queueName(), 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) throws JMSException {
+        return send(envelope, null, headers);
+    }
+
+    private String send(Envelope envelope, Duration delay, Map<String, String> headers) throws JMSException {
         TextMessage message = JmsMessages.project(session, envelope);
 
         if (delay != null && !delay.isZero() && !delay.isNegative()) {
             message.setStringProperty(JmsMessages.DELAY, Long.toString(delay.toMillis()));
+            JmsMessages.addOutOfBand(message, headers);
             producer.setDeliveryDelay(delay.toMillis());
             try {
                 producer.send(message);
             } finally {
                 producer.setDeliveryDelay(0);
             }
         } else {
+            JmsMessages.addOutOfBand(message, headers);
             producer.send(message);
         }
         return envelope.meta() != null ? envelope.meta().id() : "";

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

@@ -2,14 +2,38 @@
 
 import com.babelqueue.Envelope;
 import jakarta.jms.Message;
+import java.util.Map;
 
 /**
  * Processes one decoded, validated envelope and the raw JMS message it arrived on. Returning
  * normally acknowledges it (the consumer {@code acknowledge}s it); throwing leaves it
  * unacknowledged so the broker redelivers it (incrementing {@code JMSXDeliveryCount}), or
  * routes it to the DLQ once max-tries is reached.
+ *
+ * <p>The raw message carries the out-of-band transport headers (ADR-0028) as JMS string properties.
+ * {@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
+ * ArtemisConsumer.builder(consumer, session)
+ *     .handler(urn, (env, message) -> Tracing
+ *         .wrapHandler(tracer, h, () -> BabelHandler.headersOf(message))
+ *         .handle(env))
+ *     .build();
+ * }</pre>
  */
 @FunctionalInterface
 public interface BabelHandler {
     void handle(Envelope envelope, Message message) throws Exception;
+
+    /**
+     * The message's string properties as a {@code Map<String, String>} (blank and non-string values dropped) —
+     * the delivered-headers {@code Supplier} source for {@code com.babelqueue.otel.Tracing#wrapHandler}.
+     * Contains the contract {@code bq_} properties and any out-of-band header (e.g. {@code traceparent},
+     * {@code tracestate}). Empty when {@code message} is {@code null} or unreadable. Needs no OpenTelemetry
+     * dependency.
+     */
+    static Map<String, String> headersOf(Message message) {
+        return JmsMessages.toMap(message);
+    }
 }

src/main/java/com/babelqueue/artemis/JmsMessages.java

@@ -11,6 +11,12 @@
 import java.nio.charset.CharacterCodingException;
 import java.nio.charset.CodingErrorAction;
 import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Enumeration;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
 
 /**
  * Projects the envelope's contract fields onto a native JMS message ({@code JMSType} = URN,
@@ -127,6 +133,79 @@ static int deliveryCount(Message message) {
         return 0;
     }
 
+    /** JMS 3.1 §3.5.1: identifiers that may not be used as property names. */
+    private static final Set<String> RESERVED_NAMES = Set.of(
+        "NULL", "TRUE", "FALSE", "NOT", "AND", "OR", "BETWEEN", "LIKE", "IN", "IS", "ESCAPE");
+
+    /**
+     * Whether {@code name} is a legal JMS property name for an out-of-band header: a non-empty Java identifier
+     * that is not a reserved selector word and does not use the {@code JMS}/{@code JMSX} provider namespace.
+     * {@code traceparent} and {@code tracestate} are legal (ADR-0017: hyphenated names are not).
+     */
+    static boolean isLegalPropertyName(String name) {
+        if (name == null || name.isEmpty() || name.startsWith("JMS")
+            || RESERVED_NAMES.contains(name.toUpperCase(java.util.Locale.ROOT))) {
+            return false;
+        }
+        int[] codePoints = name.codePoints().toArray();
+        if (!Character.isJavaIdentifierStart(codePoints[0])) {
+            return false;
+        }
+        for (int cp : codePoints) {
+            if (!Character.isJavaIdentifierPart(cp)) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    /**
+     * Adds the out-of-band {@code extra} headers (e.g. {@code traceparent}, ADR-0028) to {@code message} as
+     * string properties beside the contract {@code bq_} properties. Blank keys/values and JMS-illegal names are
+     * skipped, a property already set 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(Message message, Map<String, String> extra) throws JMSException {
+        if (extra == null || extra.isEmpty()) {
+            return;
+        }
+        List<String> keys = new ArrayList<>(extra.keySet());
+        keys.removeIf(k -> !isLegalPropertyName(k));
+        keys.sort(String::compareTo);
+        for (String key : keys) {
+            String value = extra.get(key);
+            if (value == null || value.isEmpty() || message.propertyExists(key)) {
+                continue;
+            }
+            message.setStringProperty(key, value);
+        }
+    }
+
+    /**
+     * A delivered message's string properties as a flat {@code Map<String, String>} (blank and non-string
+     * values dropped) — the consume-side counterpart of {@link #addOutOfBand}. Both the contract {@code bq_}
+     * properties and out-of-band ones (e.g. {@code traceparent}) surface. Empty when the message is {@code null}
+     * or its properties are unreadable.
+     */
+    static Map<String, String> toMap(Message message) {
+        Map<String, String> out = new LinkedHashMap<>();
+        if (message == null) {
+            return out;
+        }
+        try {
+            Enumeration<?> names = message.getPropertyNames();
+            while (names != null && names.hasMoreElements()) {
+                String name = String.valueOf(names.nextElement());
+                if (message.getObjectProperty(name) instanceof String value && !value.isEmpty()) {
+                    out.put(name, value);
+                }
+            }
+        } catch (JMSException e) {
+            return out;
+        }
+        return out;
+    }
+
     private static boolean notBlank(String s) {
         return s != null && !s.isEmpty();
     }