Release v1.1.0 (ADR-0037)

.github/workflows/ci.yml

@@ -70,7 +70,7 @@ jobs:
     name: Conformance suite in sync
     runs-on: ubuntu-latest
     env:
-      BQ_CONFORMANCE_VERSION: 1.6.0
+      BQ_CONFORMANCE_VERSION: 1.8.0
       # Integrity pin for fetch.sh (sha256 of conformance/fetch.sh at the pinned tag).
       BQ_FETCH_SH_SHA256: 8bb6ba567b288fb0ac71a70fd17041b2503df119e01998d76fdad372570bf849
     steps:

CHANGELOG.md

@@ -9,6 +9,36 @@ The envelope wire format is versioned separately by `meta.schema_version`
 
 ## [Unreleased]
 
+## [1.1.0] - 2026-10-26
+
+MINOR (ADR-0037): poison bodies are dead-lettered as the ADR-0035 record and the version gate follows one
+settle rule. The envelope is unchanged (`schema_version` stays `1`).
+
+### Changed (behaviour)
+
+- **Poison body** (not JSON, not an envelope, no valid identity, or an unsupported `meta.schema_version`):
+  the DLQ record is now the **ADR-0035 record** (`PoisonRecord`: an object body spliced with a `dead_letter`
+  block, `reason: "poison"`, `original_queue` = the consumed queue; any other body under `raw`, base64 with
+  `raw_encoding` when it is not valid UTF-8) instead of a forward of the raw body. It is acknowledged only
+  after the write succeeded (otherwise `recover()`, as before). Under `unknownUrn(DELETE)` it is now
+  acknowledged without a record. With no dead-letter queue it is acknowledged, dropped and **logged**
+  (unchanged, now visible).
+- **Version-gated delivery** (`bq_schema_version` present, not blank, not `1`): settled on the first delivery,
+  never decoded, never retried. The raw body is still forwarded to the DLQ byte for byte (no `dead_letter`
+  block) and acknowledged after the write, but now **not** under `unknownUrn(DELETE)`: that acknowledges it
+  without a record (**dropped**). With no dead-letter queue it is acknowledged, dropped and logged.
+- **`BytesMessage` bodies are read, not dropped.** `JmsMessages.text(...)` treated every non-`TextMessage` as an
+  empty body, so a message from an AMQP 1.0 producer (its Data section arrives as a `BytesMessage`) was decoded as
+  `""`, and a poison one was recorded as `{"raw": ""}` and acknowledged, losing its bytes. The consumer now reads the
+  exact body bytes of a `TextMessage` (UTF-8; empty when unset) or a `BytesMessage` (new `JmsMessages.bodyBytes`)
+  and decodes them with a **strict** UTF-8 check: a valid envelope in a `BytesMessage` is routed normally; a body
+  that is not valid UTF-8 (never decoded leniently) is poison and its exact bytes go to the ADR-0035 record
+  (`raw` base64 with `raw_encoding: "base64"`). A version-gated `BytesMessage` is forwarded to the DLQ as a
+  `BytesMessage` with the same bytes. A `TextMessage` with no body is read as empty instead of `null`.
+- Requires `babelqueue-core` 1.13.0 (was 1.11.0) for `PoisonRecord`.
+- Vendored the cross-SDK conformance suite v1.8.0 (`version_gate_artemis`, `poison_dlq` runners added) and moved
+  the CI pin to it.
+
 ## [1.0.3] - 2026-10-17
 
 ### Fixed

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.0.3</version>
+  <version>1.1.0</version>
 </dependency>
 ```
 
@@ -115,8 +115,16 @@ additive.
 `bq_schema_version` (underscores — JMS forbids hyphens in property names, ADR-0017). Absent or blank
 (empty / ASCII whitespace only) → normal decode. Any other value that is not exactly `1` (no
 trimming — `"2"`, `"x"`, `" 1"`) → the message is **not** decoded and the handler is not called:
-`onError` is notified with a `null` envelope, the raw body is forwarded to `<queue>.dlq` (if
-configured) and the message is acknowledged.
+`onError` is notified with a `null` envelope and the message is settled on its **first** delivery
+(ADR-0037): under the `delete` strategy it is acknowledged; otherwise the raw body is forwarded to
+`<queue>.dlq` **exactly as received** (no `dead_letter` block) and acknowledged after the write succeeded (a
+failed write `recover()`s the session, so it is redelivered and rejected again); with no dead-letter queue it
+is acknowledged and the drop is logged.
+
+**Poison bodies (ADR-0037).** A body that is not a valid envelope (not JSON, no identity, unsupported
+`meta.schema_version`) never reaches a handler and is never retried. With a dead-letter queue it is written to
+`<queue>.dlq` as the ADR-0035 record (`reason: "poison"`; the original body kept, spliced or under `raw`) and
+acknowledged after the write; under `delete`, or with no dead-letter queue, it is acknowledged (and logged).
 
 ## Build & test
 

pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>com.babelqueue</groupId>
     <artifactId>babelqueue-artemis</artifactId>
-    <version>1.0.3</version>
+    <version>1.1.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.11.0</babelqueue-core.version>
+        <babelqueue-core.version>1.13.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

@@ -4,15 +4,21 @@
 import com.babelqueue.DeadLetters;
 import com.babelqueue.Envelope;
 import com.babelqueue.EnvelopeCodec;
+import com.babelqueue.JsonValues;
+import com.babelqueue.PoisonRecord;
 import com.babelqueue.UnknownUrnException;
 import com.babelqueue.UnknownUrnStrategy;
+import jakarta.jms.BytesMessage;
 import jakarta.jms.JMSException;
 import jakarta.jms.Message;
 import jakarta.jms.MessageConsumer;
 import jakarta.jms.MessageProducer;
+import jakarta.jms.Queue;
 import jakarta.jms.Session;
 import jakarta.jms.TextMessage;
+import java.nio.charset.StandardCharsets;
 import java.util.HashMap;
+import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.Objects;
 import java.util.function.BooleanSupplier;
@@ -27,26 +33,37 @@
  * {@code max(body, JMSXDeliveryCount − 1)} — {@code JMSXDeliveryCount} is the broker-maintained,
  * 1-based authoritative counter.
  *
- * <p>Per §7.7 the {@code bq_schema_version} string property is checked <b>before</b> the body is
- * decoded: when it is present, not blank (empty or ASCII-whitespace-only) and not exactly
- * {@link EnvelopeCodec#SCHEMA_VERSION} (compared as sent, no trimming), the message is never decoded and
- * the handler is never called — {@code onError} is notified (with a {@code null} envelope), the raw body
- * is forwarded to the DLQ (when configured) and the message is acknowledged — but only after the DLQ write succeeded. If it fails, the session is
- * {@code recover()}ed (the message is redelivered and rejected again, never lost, never handled — a later
- * message's session-wide {@code acknowledge()} cannot swallow it) and the failure goes to {@code onError}.
- * An absent or blank property
+ * <p><b>Version gate (§7.7, ADR-0037 decision 1).</b> The {@code bq_schema_version} string property is checked
+ * <b>before</b> the body is decoded: when it is present, not blank (empty or ASCII-whitespace-only) and not exactly
+ * {@link EnvelopeCodec#SCHEMA_VERSION} (compared as sent, no trimming), the message is never decoded, the handler is
+ * never called and nothing is retried, on its first delivery whatever {@code JMSXDeliveryCount} says. {@code onError}
+ * is notified (with a {@code null} envelope). Under the {@code delete} {@link Builder#unknownUrn strategy} it is
+ * acknowledged. Otherwise, with a {@link Builder#deadLetterQueue dead-letter queue}, the body is forwarded to the DLQ
+ * <b>exactly as received</b> (no {@code dead_letter} block) and the message acknowledged — but only after the DLQ write
+ * succeeded; if it fails, the session is {@code recover()}ed (the message is redelivered and rejected again, never
+ * lost, never handled — a later message's session-wide {@code acknowledge()} cannot swallow it) and the failure goes
+ * to {@code onError}. With no dead-letter queue it is acknowledged and the drop is logged. An absent or blank property
  * changes nothing, and the post-decode {@link EnvelopeCodec#accepts} check still applies.
  *
- * <p>The same rule holds for every other DLQ write (a non-conformant envelope, an unmapped URN under
- * {@code dead_letter}, a handler that exhausted max-tries): the message is acknowledged only after the DLQ write
- * succeeded; otherwise the session is {@code recover()}ed and the failure goes to {@code onError}.
+ * <p><b>Poison (ADR-0037 decision 2).</b> A body that is not a valid envelope (not JSON, no valid identity, an
+ * unsupported {@code meta.schema_version}) is never handed to a handler and never retried. Under {@code delete} it is
+ * acknowledged; with a dead-letter queue the ADR-0035 record ({@link PoisonRecord}, {@code reason: "poison"},
+ * {@code original_queue} = the consumed queue) is written to it and the message acknowledged only after the write
+ * succeeded (same recover-on-failure rule as above); with no dead-letter queue it is acknowledged and the drop is
+ * logged.
+ *
+ * <p>The same acknowledge-only-after-the-write rule holds for every other DLQ write (an unmapped URN under
+ * {@code dead_letter}, a handler that exhausted max-tries): otherwise the session is {@code recover()}ed and the
+ * failure goes to {@code onError}.
  *
  * <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.
  */
 public final class ArtemisConsumer {
 
+    private static final System.Logger LOGGER = System.getLogger(ArtemisConsumer.class.getName());
+
     /**
      * Notified of a non-conformant message, an unmapped URN, a throwing handler, or a message rejected by the
      * §7.7 schema-version gate. {@code envelope} is {@code null} when the body was not decoded.
@@ -111,6 +128,19 @@ private void handle(Message message) throws JMSException {
             BabelQueueException rejected = new BabelQueueException("Rejected a BabelQueue message with unsupported "
                 + JmsMessages.SCHEMA_VERSION + " property '" + declaredVersion + "' (supported: "
                 + EnvelopeCodec.SCHEMA_VERSION + "); body not decoded.");
+            if (UnknownUrnStrategy.DELETE.equals(unknownUrn)) {
+                message.acknowledge();
+                report(rejected, null, message);
+                return;
+            }
+            if (!deadLetterEnabled()) {
+                LOGGER.log(System.Logger.Level.WARNING,
+                    "BabelQueue Artemis: dropped a version-gated message (no dead-letter queue configured): {0}",
+                    rejected.getMessage());
+                message.acknowledge();
+                report(rejected, null, message);
+                return;
+            }
             // CLIENT_ACKNOWLEDGE is session-wide: a later message's acknowledge() would also acknowledge this
             // one. So if the DLQ write fails, force redelivery (recover) instead of leaving it unacknowledged —
             // the message is never lost and never reaches the handler; it is rejected again when redelivered.
@@ -130,12 +160,16 @@ private void handle(Message message) throws JMSException {
             return;
         }
 
-        Envelope envelope = reconcile(EnvelopeCodec.decode(JmsMessages.text(message)), message);
+        // Strict UTF-8 over the exact body bytes (a TextMessage's text or a BytesMessage's content): a body that is not
+        // valid UTF-8 is never decoded leniently, it is poison (ADR-0037 decision 2).
+        String text = JmsMessages.strictUtf8(JmsMessages.bodyBytes(message));
+        Envelope envelope = reconcile(EnvelopeCodec.decode(text == null ? "" : text), message);
 
         if (!EnvelopeCodec.accepts(envelope)) {
-            report(new BabelQueueException("Rejected a non-conformant BabelQueue envelope from Artemis."),
-                envelope, message);
-            settle(message, envelope, () -> deadLetterRaw(message), "dead-letter a non-conformant message");
+            BabelQueueException cause = new BabelQueueException(
+                "Rejected a non-conformant BabelQueue envelope from Artemis.");
+            report(cause, envelope, message);
+            settlePoison(message, envelope, cause);
             return;
         }
 
@@ -155,6 +189,25 @@ private void handle(Message message) throws JMSException {
         }
     }
 
+    /**
+     * Settles a poison body (ADR-0037 decision 2): acknowledged under {@code delete}, written to the DLQ as the
+     * ADR-0035 record and acknowledged after the write, or acknowledged and logged when no DLQ is configured.
+     */
+    private void settlePoison(Message message, Envelope envelope, Throwable cause) throws JMSException {
+        if (UnknownUrnStrategy.DELETE.equals(unknownUrn)) {
+            message.acknowledge();
+            return;
+        }
+        if (!deadLetterEnabled()) {
+            LOGGER.log(System.Logger.Level.WARNING,
+                "BabelQueue Artemis: dropped a poison message (no dead-letter queue configured): {0}",
+                cause.getMessage());
+            message.acknowledge();
+            return;
+        }
+        settle(message, envelope, () -> deadLetterPoison(message, cause), "dead-letter a poison message");
+    }
+
     /** {@code attempts = max(body, JMSXDeliveryCount − 1)}: count 1 (or absent) keeps the body's count. */
     private static Envelope reconcile(Envelope envelope, Message message) {
         int count = JmsMessages.deliveryCount(message);
@@ -228,12 +281,63 @@ envelope, reason, originalQueue(envelope), envelope.attempts(),
         dlqProducer.send(session.createQueue(dlqQueue), JmsMessages.project(session, annotated));
     }
 
-    /** Forward an undecodable message's raw body to the DLQ (no envelope to annotate). */
+    /** Whether dead-lettering is on: a DLQ producer and queue name are configured. */
+    private boolean deadLetterEnabled() {
+        return dlqQueue != null && dlqProducer != null;
+    }
+
+    /** Writes the ADR-0035 record of a poison body to the DLQ. */
+    private void deadLetterPoison(Message message, Throwable cause) throws JMSException {
+        String record = poisonRecord(JmsMessages.bodyBytes(message),
+            consumedQueue(message), Math.max(JmsMessages.deliveryCount(message) - 1, 0), cause);
+        dlqProducer.send(session.createQueue(dlqQueue), session.createTextMessage(record));
+    }
+
+    /**
+     * The ADR-0035 DLQ record of a poison {@code body}: the one builder this consumer writes with (every conformance
+     * {@code poison_dlq} case reaches it through {@link #poll()}, invalid UTF-8 as a {@code BytesMessage}).
+     */
+    static String poisonRecord(byte[] body, String originalQueue, int attempts, Throwable cause) {
+        Map<String, Object> block = new LinkedHashMap<>();
+        block.put("reason", "poison");
+        block.put("error", cause.getMessage());
+        block.put("exception", cause.getClass().getName());
+        block.put("failed_at", System.currentTimeMillis());
+        block.put("original_queue", originalQueue);
+        block.put("attempts", attempts);
+        block.put("lang", EnvelopeCodec.SOURCE_LANG);
+        return PoisonRecord.build(body, JsonValues.encode(block));
+    }
+
+    /** The consumed queue's name: the message's JMS destination, else the DLQ name without its {@code .dlq} suffix. */
+    private String consumedQueue(Message message) {
+        try {
+            if (message.getJMSDestination() instanceof Queue queue && queue.getQueueName() != null) {
+                return queue.getQueueName();
+            }
+        } catch (JMSException unreadable) {
+            // fall through to the DLQ-derived name
+        }
+        return dlqQueue != null && dlqQueue.endsWith(".dlq")
+            ? dlqQueue.substring(0, dlqQueue.length() - ".dlq".length())
+            : "";
+    }
+
+    /** Forward a version-gated message's raw body to the DLQ, exactly as received (no envelope to annotate). */
     private void deadLetterRaw(Message message) throws JMSException {
-        if (dlqQueue == null || dlqProducer == null) {
+        if (!deadLetterEnabled()) {
             return;
         }
-        TextMessage forwarded = session.createTextMessage(JmsMessages.text(message));
+        byte[] body = JmsMessages.bodyBytes(message);
+        Message forwarded;
+        if (message instanceof BytesMessage) {
+            // Keep the message kind: the bytes (valid UTF-8 or not) are forwarded exactly as received.
+            BytesMessage bytes = session.createBytesMessage();
+            bytes.writeBytes(body);
+            forwarded = bytes;
+        } else {
+            forwarded = session.createTextMessage(new String(body, StandardCharsets.UTF_8));
+        }
         dlqProducer.send(session.createQueue(dlqQueue), forwarded);
     }