From 7bbaf891de08473a8c47af30a58307cd9d2578b1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Muhammet=20=C5=9Eafak?= Date: Sat, 31 Oct 2026 10:21:28 +0300 Subject: [PATCH 1/4] Carry traceparent out of band on the artemis publisher and consumer Let the publisher accept transport headers and surface non-bq headers to the handler through BabelHandler.headersOf, so the OTel wrapper can link send and process spans across the hop (ADR-0028). --- CHANGELOG.md | 5 + .../babelqueue/artemis/ArtemisPublisher.java | 35 +++++- .../com/babelqueue/artemis/BabelHandler.java | 24 +++++ .../com/babelqueue/artemis/JmsMessages.java | 79 ++++++++++++++ .../artemis/ArtemisConsumerTest.java | 29 +++++ .../artemis/ArtemisPublisherTest.java | 102 ++++++++++++++++++ .../babelqueue/artemis/JmsMessagesTest.java | 34 ++++++ 7 files changed, 307 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 55a9318..9a26e18 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,11 @@ The envelope wire format is versioned separately by `meta.schema_version` ## [Unreleased] +### Added + +- **W3C `traceparent` out-of-band propagation (ADR-0028).** `ArtemisPublisher`: `publishWithHeaders(Envelope, Map)` (the produce-side seam for the core's `com.babelqueue.otel.HeaderSender`) and a `publish(..., Map 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` — 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 diff --git a/src/main/java/com/babelqueue/artemis/ArtemisPublisher.java b/src/main/java/com/babelqueue/artemis/ArtemisPublisher.java index 43d4cdf..8c97ddf 100644 --- a/src/main/java/com/babelqueue/artemis/ArtemisPublisher.java +++ b/src/main/java/com/babelqueue/artemis/ArtemisPublisher.java @@ -20,7 +20,13 @@ * ({@code schema_version} stays 1); Artemis is purely additive. * *

A positive delay uses native JMS 2.0 scheduled delivery ({@code setDeliveryDelay}); - * {@code bq-delay} is set for observability. + * {@code bq_delay} is set for observability. + * + *

Out-of-band transport headers (e.g. the W3C {@code traceparent}, ADR-0028) ride as extra JMS string + * properties beside 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,11 +55,37 @@ public String publish(String urn, Map data, String traceId) thro /** Publish with an optional relative {@code delay} (native JMS 2.0 {@code setDeliveryDelay}). */ public String publish(String urn, Map 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 data, String traceId, Duration delay, + Map 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 headers) throws JMSException { + return send(envelope, null, headers); + } + + private String send(Envelope envelope, Duration delay, Map 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); @@ -61,6 +93,7 @@ public String publish(String urn, Map data, String traceId, Dura producer.setDeliveryDelay(0); } } else { + JmsMessages.addOutOfBand(message, headers); producer.send(message); } return envelope.meta() != null ? envelope.meta().id() : ""; diff --git a/src/main/java/com/babelqueue/artemis/BabelHandler.java b/src/main/java/com/babelqueue/artemis/BabelHandler.java index 71f2943..a816f4f 100644 --- a/src/main/java/com/babelqueue/artemis/BabelHandler.java +++ b/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. + * + *

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: + * + * 

{@code
+ * ArtemisConsumer.builder(consumer, session)
+ *     .handler(urn, (env, message) -> Tracing
+ *         .wrapHandler(tracer, h, () -> BabelHandler.headersOf(message))
+ *         .handle(env))
+ *     .build();
+ * }
*/ @FunctionalInterface public interface BabelHandler { void handle(Envelope envelope, Message message) throws Exception; + + /** + * The message's string properties as a {@code Map} (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 headersOf(Message message) { + return JmsMessages.toMap(message); + } } diff --git a/src/main/java/com/babelqueue/artemis/JmsMessages.java b/src/main/java/com/babelqueue/artemis/JmsMessages.java index ba854eb..031c960 100644 --- a/src/main/java/com/babelqueue/artemis/JmsMessages.java +++ b/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 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 extra) throws JMSException { + if (extra == null || extra.isEmpty()) { + return; + } + List 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} (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 toMap(Message message) { + Map 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(); } diff --git a/src/test/java/com/babelqueue/artemis/ArtemisConsumerTest.java b/src/test/java/com/babelqueue/artemis/ArtemisConsumerTest.java index 3a6d9e6..4950d5e 100644 --- a/src/test/java/com/babelqueue/artemis/ArtemisConsumerTest.java +++ b/src/test/java/com/babelqueue/artemis/ArtemisConsumerTest.java @@ -862,4 +862,33 @@ void versionGatedBytesMessageIsForwardedAsBytesExactly() throws Exception { verify(dlqProducer).send(any(Queue.class), org.mockito.ArgumentMatchers.eq(forwarded)); verify(message).acknowledge(); } + + @Test + @SuppressWarnings("unchecked") + void publishedTraceparentReachesTheHandlerViaHeadersOfAndNotTheBody() throws Exception { + String traceparent = "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01"; + Map store = new java.util.LinkedHashMap<>(); + TextMessage sentMessage = ArtemisPublisherTest.storingMessage(store); + Session session = mock(Session.class); + when(session.createTextMessage(org.mockito.ArgumentMatchers.anyString())).thenReturn(sentMessage); + MessageProducer producer = mock(MessageProducer.class); + ArtemisPublisher.create(session, producer).publish(URN, Map.of("order_id", 1), "trace-1", null, + Map.of("traceparent", traceparent)); + org.mockito.ArgumentCaptor sentBody = org.mockito.ArgumentCaptor.forClass(String.class); + org.mockito.Mockito.verify(session).createTextMessage(sentBody.capture()); + + // the broker hands the same message back, with the body the publisher wrote + when(sentMessage.getText()).thenReturn(sentBody.getValue()); + when(sentMessage.getJMSType()).thenReturn(URN); + Map[] seen = new Map[]{null}; + + ArtemisConsumer.builder(consumerWith(sentMessage), mock(Session.class)) + .handler(URN, (e, m) -> seen[0] = BabelHandler.headersOf(m)) + .build() + .poll(); + + assertEquals(traceparent, seen[0].get("traceparent")); + assertEquals("0", seen[0].get("bq_attempts")); + assertTrue(!sentBody.getValue().contains("traceparent")); + } } diff --git a/src/test/java/com/babelqueue/artemis/ArtemisPublisherTest.java b/src/test/java/com/babelqueue/artemis/ArtemisPublisherTest.java index 80cb804..4e7caf3 100644 --- a/src/test/java/com/babelqueue/artemis/ArtemisPublisherTest.java +++ b/src/test/java/com/babelqueue/artemis/ArtemisPublisherTest.java @@ -1,7 +1,10 @@ package com.babelqueue.artemis; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -13,6 +16,8 @@ import jakarta.jms.Session; import jakarta.jms.TextMessage; import java.time.Duration; +import java.util.Collections; +import java.util.LinkedHashMap; import java.util.Map; import org.junit.jupiter.api.Test; import org.mockito.ArgumentCaptor; @@ -101,4 +106,101 @@ void publishWithoutTraceMintsAFreshTrace() throws Exception { org.junit.jupiter.api.Assertions.assertFalse( EnvelopeCodec.decode(body.getValue()).traceId().isEmpty()); } + + private static final String TRACEPARENT = "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01"; + + /** A TextMessage mock backed by a real property store, so what is set can be read back (round-trip). */ + static TextMessage storingMessage(Map store) throws Exception { + TextMessage message = mock(TextMessage.class); + doAnswer(inv -> store.put(inv.getArgument(0), inv.getArgument(1))) + .when(message).setStringProperty(anyString(), anyString()); + when(message.propertyExists(anyString())).thenAnswer(inv -> store.containsKey((String) inv.getArgument(0))); + when(message.getObjectProperty(anyString())).thenAnswer(inv -> store.get((String) inv.getArgument(0))); + when(message.getPropertyNames()).thenAnswer(inv -> Collections.enumeration(store.keySet())); + return message; + } + + @Test + void publishWithHeadersCarriesTraceparentBesideContractPropertiesAndNotInTheBody() throws Exception { + Map store = new LinkedHashMap<>(); + TextMessage message = storingMessage(store); + Session session = sessionReturning(message); + Envelope env = EnvelopeCodec.make(URN, Map.of("order_id", 7), "orders", "trace-1"); + + String id = ArtemisPublisher.create(session, producerOn("orders")) + .publishWithHeaders(env, Map.of("traceparent", TRACEPARENT, "tracestate", "vendor=1")); + + assertEquals(env.meta().id(), id); + assertEquals(TRACEPARENT, store.get("traceparent")); + assertEquals("vendor=1", store.get("tracestate")); + assertEquals("1", store.get("bq_schema_version")); // contract projection intact + ArgumentCaptor body = ArgumentCaptor.forClass(String.class); + verify(session).createTextMessage(body.capture()); + assertFalse(body.getValue().contains("traceparent")); + assertFalse(body.getValue().contains(TRACEPARENT)); + assertEquals(EnvelopeCodec.encode(env), body.getValue()); + } + + @Test + void headerlessPublishWithHeadersSetsExactlyTheContractProperties() throws Exception { + Map plain = new LinkedHashMap<>(); + Map withNull = new LinkedHashMap<>(); + Map withEmpty = new LinkedHashMap<>(); + Envelope env = EnvelopeCodec.make(URN, Map.of(), "orders", "trace-1"); + + ArtemisPublisher.create(sessionReturning(storingMessage(plain)), producerOn("orders")) + .publishWithHeaders(env, null); + ArtemisPublisher.create(sessionReturning(storingMessage(withNull)), producerOn("orders")) + .publishWithHeaders(env, Map.of()); + ArtemisPublisher.create(sessionReturning(storingMessage(withEmpty)), producerOn("orders")) + .publishWithHeaders(env, Map.of("traceparent", "")); + + assertEquals(plain, withNull); + assertEquals(plain, withEmpty); + assertFalse(plain.containsKey("traceparent")); + } + + @Test + void contractPropertyWinsACollisionAndIllegalJmsNamesAreSkipped() throws Exception { + Map store = new LinkedHashMap<>(); + Envelope env = EnvelopeCodec.make(URN, Map.of(), "orders", "trace-1"); + + ArtemisPublisher.create(sessionReturning(storingMessage(store)), producerOn("orders")) + .publishWithHeaders(env, Map.of( + "bq_attempts", "99", "x-hyphen", "v", "JMSXGroupID", "g", "JMSType", "t", "1bad", "v", + "traceparent", TRACEPARENT)); + + assertEquals("0", store.get("bq_attempts")); + assertFalse(store.containsKey("x-hyphen")); + assertFalse(store.containsKey("JMSXGroupID")); + assertFalse(store.containsKey("JMSType")); + assertFalse(store.containsKey("1bad")); + assertEquals(TRACEPARENT, store.get("traceparent")); + } + + @Test + void publishOverloadWithHeadersRidesTheDelayedMessageToo() throws Exception { + Map store = new LinkedHashMap<>(); + MessageProducer producer = producerOn("orders"); + + ArtemisPublisher.create(sessionReturning(storingMessage(store)), producer) + .publish(URN, Map.of(), null, Duration.ofSeconds(30), Map.of("traceparent", TRACEPARENT)); + + assertEquals(TRACEPARENT, store.get("traceparent")); + assertEquals("30000", store.get("bq_delay")); + verify(producer).setDeliveryDelay(30000L); + } + + @Test + void publishWithHeadersMatchesTheOtelHeaderSenderContract() throws Exception { + Map store = new LinkedHashMap<>(); + ArtemisPublisher publisher = ArtemisPublisher.create( + sessionReturning(storingMessage(store)), producerOn("orders")); + com.babelqueue.otel.HeaderSender sender = publisher::publishWithHeaders; + + assertDoesNotThrow(() -> sender.send(EnvelopeCodec.make(URN, Map.of(), "orders", "t"), + Map.of("traceparent", TRACEPARENT))); + + assertEquals(TRACEPARENT, store.get("traceparent")); + } } diff --git a/src/test/java/com/babelqueue/artemis/JmsMessagesTest.java b/src/test/java/com/babelqueue/artemis/JmsMessagesTest.java index dd29dca..a2d0fcd 100644 --- a/src/test/java/com/babelqueue/artemis/JmsMessagesTest.java +++ b/src/test/java/com/babelqueue/artemis/JmsMessagesTest.java @@ -1,7 +1,9 @@ package com.babelqueue.artemis; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -74,4 +76,36 @@ void deliveryCountIsZeroWhenAbsentOrOnError() throws Exception { when(throwing.propertyExists("JMSXDeliveryCount")).thenThrow(new JMSException("boom")); assertEquals(0, JmsMessages.deliveryCount(throwing)); } + + @Test + void propertyNameLegalityFollowsJmsIdentifierRules() { + assertTrue(JmsMessages.isLegalPropertyName("traceparent")); + assertTrue(JmsMessages.isLegalPropertyName("tracestate")); + assertTrue(JmsMessages.isLegalPropertyName("bq_replay_bypass")); + assertFalse(JmsMessages.isLegalPropertyName("bq-replay-bypass")); + assertFalse(JmsMessages.isLegalPropertyName("JMSXGroupID")); + assertFalse(JmsMessages.isLegalPropertyName("NULL")); + assertFalse(JmsMessages.isLegalPropertyName("1abc")); + assertTrue(JmsMessages.isLegalPropertyName("trace\uD835\uDC1Aparent")); // supplementary-plane letter + assertFalse(JmsMessages.isLegalPropertyName("")); + assertFalse(JmsMessages.isLegalPropertyName(null)); + } + + @Test + void toMapSurfacesStringPropertiesAndToleratesNullUnreadableAndNonStrings() throws Exception { + Message message = mock(Message.class); + when(message.getPropertyNames()).thenReturn( + java.util.Collections.enumeration(java.util.List.of("traceparent", "blank", "JMSXDeliveryCount"))); + when(message.getObjectProperty("traceparent")).thenReturn("00-x"); + when(message.getObjectProperty("blank")).thenReturn(""); + when(message.getObjectProperty("JMSXDeliveryCount")).thenReturn(2); + + assertEquals(java.util.Map.of("traceparent", "00-x"), JmsMessages.toMap(message)); + assertTrue(JmsMessages.toMap(null).isEmpty()); + assertTrue(JmsMessages.toMap(mock(Message.class)).isEmpty()); // null enumeration + Message throwing = mock(Message.class); + when(throwing.getPropertyNames()).thenThrow(new JMSException("boom")); + assertTrue(JmsMessages.toMap(throwing).isEmpty()); + assertTrue(BabelHandler.headersOf(null).isEmpty()); + } } From 21eb496575ef9bde57572c3d3022256b17871e22 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Muhammet=20=C5=9Eafak?= Date: Sat, 31 Oct 2026 17:35:01 +0300 Subject: [PATCH 2/4] Wire the dead-letter observer into the artemis consumer Accept an optional DeadLetterObserver and call it exactly once after each confirmed DLQ write (and, on ASB, after an SDK-initiated native dead-letter); observer failures never change the settle outcome (ADR-0039). --- CHANGELOG.md | 1 + .../babelqueue/artemis/ArtemisConsumer.java | 36 ++++ .../artemis/ArtemisConsumerTest.java | 181 ++++++++++++++++++ 3 files changed, 218 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9a26e18..c19d7e1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,7 @@ The envelope wire format is versioned separately by `meta.schema_version` ### 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)` (the produce-side seam for the core's `com.babelqueue.otel.HeaderSender`) and a `publish(..., Map 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` — 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). diff --git a/src/main/java/com/babelqueue/artemis/ArtemisConsumer.java b/src/main/java/com/babelqueue/artemis/ArtemisConsumer.java index 3244463..d900610 100644 --- a/src/main/java/com/babelqueue/artemis/ArtemisConsumer.java +++ b/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}. * + *

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. + * *

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 queue.isBlank() ? 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 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); } diff --git a/src/test/java/com/babelqueue/artemis/ArtemisConsumerTest.java b/src/test/java/com/babelqueue/artemis/ArtemisConsumerTest.java index 4950d5e..7721a81 100644 --- a/src/test/java/com/babelqueue/artemis/ArtemisConsumerTest.java +++ b/src/test/java/com/babelqueue/artemis/ArtemisConsumerTest.java @@ -15,6 +15,8 @@ import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import com.babelqueue.DeadLetterEvent; +import com.babelqueue.DeadLetterObserver; import com.babelqueue.Envelope; import com.babelqueue.EnvelopeCodec; import com.babelqueue.UnknownUrnException; @@ -891,4 +893,183 @@ void publishedTraceparentReachesTheHandlerViaHeadersOfAndNotTheBody() throws Exc assertEquals("0", seen[0].get("bq_attempts")); assertTrue(!sentBody.getValue().contains("traceparent")); } + + // ---- ADR-0039 section 5.1 dead-letter observer ---------------------------------------------- + + private static Session dlqSession() throws Exception { + Session session = mock(Session.class); + when(session.createTextMessage(anyString())).thenReturn(mock(TextMessage.class)); + when(session.createQueue("orders.dlq")).thenReturn(mock(Queue.class)); + return session; + } + + @Test + void observerSeesAnExhaustedRetryOnceWithQueueDlqAndReason() throws Exception { + TextMessage message = incoming(body(2), URN, 3); + List events = new ArrayList<>(); + MessageProducer dlqProducer = mock(MessageProducer.class); + + ArtemisConsumer.builder(consumerWith(message), dlqSession()) + .handler(URN, (env, m) -> { throw new IllegalStateException("boom"); }) + .deadLetterQueue(dlqProducer, "orders.dlq") + .maxTries(3) + .deadLetterObserver(events::add) + .build() + .poll(); + + assertEquals(List.of(new DeadLetterEvent("orders", "orders.dlq", "failed")), events); + verify(message).acknowledge(); + } + + @Test + void observerIsNotCalledWhileTheRetryCapIsNotReached() throws Exception { + List events = new ArrayList<>(); + + ArtemisConsumer.builder(consumerWith(incoming(body(0), URN, 1)), dlqSession()) + .handler(URN, (env, m) -> { throw new IllegalStateException("boom"); }) + .deadLetterQueue(mock(MessageProducer.class), "orders.dlq") + .maxTries(3) + .deadLetterObserver(events::add) + .build() + .poll(); + + assertTrue(events.isEmpty()); + } + + @Test + void observerSeesAnUnknownUrnDeadLetter() throws Exception { + List events = new ArrayList<>(); + + ArtemisConsumer.builder(consumerWith(incoming(body(0), URN, 1)), dlqSession()) + .deadLetterQueue(mock(MessageProducer.class), "orders.dlq") + .unknownUrn(UnknownUrnStrategy.DEAD_LETTER) + .deadLetterObserver(events::add) + .build() + .poll(); + + assertEquals(List.of(new DeadLetterEvent("orders", "orders.dlq", "unknown_urn")), events); + } + + @Test + void observerSeesAPoisonRecord() throws Exception { + List events = new ArrayList<>(); + + ArtemisConsumer.builder(consumerWith(incoming("not-json", null, 1)), dlqSession()) + .deadLetterQueue(mock(MessageProducer.class), "orders.dlq") + .deadLetterObserver(events::add) + .build() + .poll(); + + assertEquals(List.of(new DeadLetterEvent("orders", "orders.dlq", "poison")), events); + } + + @Test + void observerSeesAVersionGatedRawBodyWithAnEmptyReason() throws Exception { + List events = new ArrayList<>(); + TextMessage message = incoming("not-json", URN, 1); + when(message.getStringProperty("bq_schema_version")).thenReturn("2"); + + ArtemisConsumer.builder(consumerWith(message), dlqSession()) + .deadLetterQueue(mock(MessageProducer.class), "orders.dlq") + .deadLetterObserver(events::add) + .build() + .poll(); + + assertEquals(List.of(new DeadLetterEvent("orders", "orders.dlq", "")), events); + verify(message).acknowledge(); + } + + @Test + void observerIsNotCalledWhenTheDlqWriteFails() throws Exception { + List events = new ArrayList<>(); + TextMessage gated = incoming("not-json", URN, 1); + when(gated.getStringProperty("bq_schema_version")).thenReturn("2"); + for (TextMessage message : List.of(gated, incoming("not-json", null, 1), incoming(body(2), URN, 3))) { + MessageProducer failing = mock(MessageProducer.class); + Session session = sessionWithDlq(failing); + + ArtemisConsumer.builder(consumerWith(message), session) + .handler(URN, (env, m) -> { throw new IllegalStateException("boom"); }) + .deadLetterQueue(failing, "orders.dlq") + .maxTries(3) + .deadLetterObserver(events::add) + .build() + .poll(); + + verify(session).recover(); + verify(message, never()).acknowledge(); + } + + assertTrue(events.isEmpty()); + } + + @Test + void observerIsNotCalledWhenNoDlqIsConfiguredOrTheStrategyIsDelete() throws Exception { + List events = new ArrayList<>(); + + ArtemisConsumer.builder(consumerWith(incoming("not-json", null, 1)), mock(Session.class)) + .deadLetterObserver(events::add).build().poll(); + ArtemisConsumer.builder(consumerWith(incoming(body(2), URN, 3)), mock(Session.class)) + .handler(URN, (env, m) -> { throw new IllegalStateException("boom"); }) + .maxTries(3).deadLetterObserver(events::add).build().poll(); + ArtemisConsumer.builder(consumerWith(incoming("not-json", null, 1)), dlqSession()) + .deadLetterQueue(mock(MessageProducer.class), "orders.dlq") + .unknownUrn(UnknownUrnStrategy.DELETE) + .deadLetterObserver(events::add).build().poll(); + + assertTrue(events.isEmpty()); + } + + @Test + void aThrowingObserverDoesNotChangeTheSettleOutcome() throws Exception { + DeadLetterObserver boom = event -> { throw new IllegalStateException("observer down"); }; + DeadLetterObserver fatal = event -> { throw new AssertionError("observer asserts"); }; + + for (DeadLetterObserver observer : List.of(boom, fatal)) { + TextMessage gated = incoming("not-json", URN, 1); + when(gated.getStringProperty("bq_schema_version")).thenReturn("2"); + for (TextMessage message : List.of(gated, incoming("not-json", null, 1), incoming(body(2), URN, 3))) { + Session session = dlqSession(); + MessageProducer dlqProducer = mock(MessageProducer.class); + + ArtemisConsumer.builder(consumerWith(message), session) + .handler(URN, (env, m) -> { throw new IllegalStateException("boom"); }) + .deadLetterQueue(dlqProducer, "orders.dlq") + .maxTries(3) + .deadLetterObserver(observer) + .build() + .poll(); + + verify(dlqProducer).send(any(Queue.class), any(Message.class)); + verify(message).acknowledge(); + verify(session, never()).recover(); + } + } + } + + @Test + void eventQueueIsMetaQueueForAnEnvelopeAndTheConsumedQueueForAPoisonRecord() throws Exception { + // meta.queue ("billing") differs from the consumed queue ("orders", derived from the DLQ name). + Envelope base = EnvelopeCodec.make(URN, Map.of("order_id", 1), "billing", "trace-1"); + String billing = EnvelopeCodec.encode( + new Envelope(base.job(), base.traceId(), base.data(), base.meta(), 2, base.deadLetter())); + List events = new ArrayList<>(); + + ArtemisConsumer.builder(consumerWith(incoming(billing, URN, 3)), dlqSession()) + .handler(URN, (env, m) -> { throw new IllegalStateException("boom"); }) + .deadLetterQueue(mock(MessageProducer.class), "orders.dlq") + .maxTries(3) + .deadLetterObserver(events::add) + .build() + .poll(); + ArtemisConsumer.builder(consumerWith(incoming("not-json", null, 1)), dlqSession()) + .deadLetterQueue(mock(MessageProducer.class), "orders.dlq") + .deadLetterObserver(events::add) + .build() + .poll(); + + assertEquals(List.of( + new DeadLetterEvent("billing", "orders.dlq", "failed"), + new DeadLetterEvent("orders", "orders.dlq", "poison")), events); + } } From 7dbf82ac8a582160c969e1bb97b61ebf14bece48 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Muhammet=20=C5=9Eafak?= Date: Sat, 31 Oct 2026 20:00:43 +0300 Subject: [PATCH 3/4] Use the ADR-0031 ASCII-blank set for the DLQ event queue fallback Fall back from a blank meta.queue to the consumed queue only for the dead-letter observer event, using the core isBlankAscii check; the DLQ record's dead_letter.original_queue is unchanged, so the wire is too. --- .../babelqueue/artemis/ArtemisConsumer.java | 2 +- .../artemis/ArtemisConsumerTest.java | 24 +++++++++++++++++++ 2 files changed, 25 insertions(+), 1 deletion(-) diff --git a/src/main/java/com/babelqueue/artemis/ArtemisConsumer.java b/src/main/java/com/babelqueue/artemis/ArtemisConsumer.java index d900610..fb80906 100644 --- a/src/main/java/com/babelqueue/artemis/ArtemisConsumer.java +++ b/src/main/java/com/babelqueue/artemis/ArtemisConsumer.java @@ -366,7 +366,7 @@ private static String urn(Message message, Envelope 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 queue.isBlank() ? consumedQueue(message) : queue; + 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. */ diff --git a/src/test/java/com/babelqueue/artemis/ArtemisConsumerTest.java b/src/test/java/com/babelqueue/artemis/ArtemisConsumerTest.java index 7721a81..16fd239 100644 --- a/src/test/java/com/babelqueue/artemis/ArtemisConsumerTest.java +++ b/src/test/java/com/babelqueue/artemis/ArtemisConsumerTest.java @@ -1072,4 +1072,28 @@ void eventQueueIsMetaQueueForAnEnvelopeAndTheConsumedQueueForAPoisonRecord() thr new DeadLetterEvent("billing", "orders.dlq", "failed"), new DeadLetterEvent("orders", "orders.dlq", "poison")), events); } + + @Test + void eventQueueFallbackTreatsOnlyTheAsciiWhitespaceSetAsBlank() throws Exception { + String json = EnvelopeCodec.encode(EnvelopeCodec.make(URN, Map.of("order_id", 1), "billing", "trace-1")); + // meta.queue as JSON text -> the queue the event must report. Only the ASCII whitespace set is blank (ADR-0031): + // Unicode spaces and control characters (Character.isWhitespace) are real, non-blank names. + String[][] cases = { + {" \\t", "orders"}, {"", "orders"}, {"\\u000b\\f\\r", "orders"}, + {"\\u2003", "\u2003"}, {"\\u00a0", "\u00a0"}, {"\\u001c", "\u001c"}, {"\\u0085", "\u0085"}, + }; + for (String[] c : cases) { + List events = new ArrayList<>(); + + ArtemisConsumer.builder(consumerWith(incoming(json.replace("\"queue\":\"billing\"", "\"queue\":\"" + c[0] + "\""), null, 1)), dlqSession()) + .deadLetterQueue(mock(MessageProducer.class), "orders.dlq") + .unknownUrn(UnknownUrnStrategy.DEAD_LETTER) + .deadLetterObserver(events::add) + .build() + .poll(); + + assertEquals(1, events.size(), c[0]); + assertEquals(c[1], events.get(0).queue(), "meta.queue=" + c[0]); + } + } } From fca0e1d3964aa87dd3069b9da0164e3ac9a59e08 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Muhammet=20=C5=9Eafak?= Date: Tue, 3 Nov 2026 21:20:46 +0300 Subject: [PATCH 4/4] Release v1.2.0 (ADR-0038, ADR-0039) --- CHANGELOG.md | 2 ++ README.md | 2 +- pom.xml | 4 ++-- 3 files changed, 5 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c19d7e1..0354f7d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,8 @@ 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. diff --git a/README.md b/README.md index 6e91c4c..e751ffe 100644 --- a/README.md +++ b/README.md @@ -32,7 +32,7 @@ them instead of re-implementing them (the envelope stays `schema_version: 1`): com.babelqueue babelqueue-artemis - 1.1.1 + 1.2.0 ``` diff --git a/pom.xml b/pom.xml index a300e94..278aa5a 100644 --- a/pom.xml +++ b/pom.xml @@ -6,7 +6,7 @@ com.babelqueue babelqueue-artemis - 1.1.1 + 1.2.0 jar BabelQueue Apache ActiveMQ Artemis @@ -49,7 +49,7 @@ 17 UTF-8 UTF-8 - 1.14.0 + 1.15.0 3.1.0 5.10.3 5.12.0