Release v1.1.0 (ADR-0037)

.github/workflows/ci.yml

@@ -33,7 +33,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,35 @@ The envelope wire format is versioned separately by `meta.schema_version`
 
 ## [Unreleased]
 
+## [1.1.0] - 2026-10-26
+
+MINOR (ADR-0037): the consumer gains an opt-in BabelQueue DLQ and settles poison bodies and version-gated
+deliveries on their first delivery. The envelope is unchanged (`schema_version` stays `1`).
+
+### Added
+
+- `AsbConsumer.Builder.deadLetterSender(ServiceBusSenderClient)` (opt-in): a sender bound to the `<queue>.dlq`
+  entity (an ordinary Service Bus queue, §4.5; it must exist and belongs to the caller). "Dead-lettering on"
+  means this sender is set.
+
+### Changed (behaviour)
+
+- **Poison body** (not JSON, not an envelope, no valid identity, or an unsupported `meta.schema_version`):
+  never handed to a handler and never retried by the SDK. With a DLQ sender it is sent as the ADR-0035 record
+  (`PoisonRecord`, `reason: "poison"`, `original_queue` = the receiver's entity, the exact body bytes kept) and
+  `Complete`d **only after** the send succeeded; a failed send `Abandon`s it and is reported. With no DLQ
+  sender nothing changes: it is abandoned and reported, so `MaxDeliveryCount` moves it to the native
+  sub-queue.
+- **Version-gated delivery** (`bq-schema-version` present, not null, not blank, not `1`): settled on the first
+  delivery, never decoded, no handler or hook, no retry. With a DLQ sender the **raw body, byte for byte** (no
+  `dead_letter` block) is sent to `<queue>.dlq` and the message `Complete`d after the send (a failed send
+  `Abandon`s it, so the write is retried on redelivery). With no DLQ sender nothing changes: it is dead-lettered
+  into the native `$DeadLetterQueue` (reason `unsupported schema version`). This consumer has no unknown-URN
+  strategy option, so the rule is `fail`; an `onUnknownUrn` hook never affects it.
+- Requires `babelqueue-core` 1.13.0 (was 1.11.0) for `PoisonRecord`.
+- Vendored the cross-SDK conformance suite v1.8.0 (`version_gate_asb`, `poison_dlq` runners added) and moved
+  the CI pin to it.
+
 ## [1.0.2] - 2026-10-17
 
 ### Fixed

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.0.2</version>
+  <version>1.1.0</version>
 </dependency>
 ```
 
@@ -71,8 +71,23 @@ Service Bus is purely additive.
 `ApplicationProperties["bq-schema-version"]`. Missing, null or blank (empty / ASCII whitespace
 only) → normal decode. Integer `1` (any integral width) and the string `"1"` → normal decode. Any
 other value (`2`, `1.5`, `"2"`, `"x"`, `" 1"`, other types) → the body is **not** decoded, the handler
-is not called, `onError` is notified with a `null` envelope, and the message is explicitly
-**dead-lettered** (reason `unsupported schema version`) — never abandoned.
+is not called, no hook runs, `onError` is notified with a `null` envelope, and the message is settled on its
+**first** delivery (ADR-0037): with a `deadLetterSender` (bound to `<queue>.dlq`) the body is sent there
+**exactly as received** (no `dead_letter` block) and `complete`d after the send succeeded (a failed send
+`abandon`s it, so the write is retried); without one it is explicitly **dead-lettered** to the native
+sub-queue (reason `unsupported schema version`) — never abandoned.
+
+**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 by the SDK. With a `deadLetterSender` it is
+sent to `<queue>.dlq` as the ADR-0035 record (`reason: "poison"`; the original bytes kept, spliced or under
+`raw`) and `complete`d after the send; without one it is `abandon`ed, so `MaxDeliveryCount` moves it to the
+native dead-letter sub-queue.
+
+```java
+ServiceBusSenderClient dlq = new ServiceBusClientBuilder()
+    .connectionString(cs).sender().queueOrTopicName("orders.dlq").buildClient();
+AsbConsumer.builder(receiver).handler(urn, handler).deadLetterSender(dlq).build();
+```
 
 ## Build & test
 

pom.xml

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

@@ -1,15 +1,21 @@
 package com.babelqueue.azureservicebus;
 
+import com.azure.core.util.BinaryData;
+import com.azure.messaging.servicebus.ServiceBusMessage;
 import com.azure.messaging.servicebus.ServiceBusReceivedMessage;
 import com.azure.messaging.servicebus.ServiceBusReceiverClient;
+import com.azure.messaging.servicebus.ServiceBusSenderClient;
 import com.azure.messaging.servicebus.models.DeadLetterOptions;
 import com.babelqueue.BabelQueueException;
 import com.babelqueue.Envelope;
 import com.babelqueue.EnvelopeCodec;
+import com.babelqueue.JsonValues;
+import com.babelqueue.PoisonRecord;
 import com.babelqueue.UnknownUrnException;
 import java.math.BigDecimal;
 import java.time.Duration;
 import java.util.HashMap;
+import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.Objects;
 import java.util.function.BooleanSupplier;
@@ -22,17 +28,29 @@
  * {@code DeliveryCount - 1} (broker-authoritative on ASB) for the handler. The poll loop never
  * stops on a bad message — observe via {@code onError}/{@code onUnknownUrn}.
  *
- * <p>Per §4.7 the {@code bq-schema-version} application property is checked <b>before</b> the body is
- * decoded: when it is present and is not the schema version the core supports
- * ({@link EnvelopeCodec#SCHEMA_VERSION}), the body is never decoded or routed. Producers write an
+ * <p><b>Version gate (§4.7, ADR-0037 decision 1).</b> The {@code bq-schema-version} application property is checked
+ * <b>before</b> the body is decoded: when it is present and is not the schema version the core supports
+ * ({@link EnvelopeCodec#SCHEMA_VERSION}), the body is never decoded or routed, no handler or
+ * {@code onUnknownUrn} hook runs and nothing is retried, on the first delivery. Producers write an
  * integer {@code 1}, so any integral numeric type with value {@code 1} and the string {@code "1"}
  * (compared exactly as sent, no trimming) are accepted; every other number ({@code 2}, {@code 0},
- * {@code 1.5}), string ({@code "2"}, {@code "x"}, {@code "01"}, {@code " 1"}) or type is rejected.
- * A rejected message is explicitly <b>dead-lettered</b> (reason {@code unsupported schema version}) —
- * never abandoned, which would redeliver it until {@code MaxDeliveryCount} — and {@code onError} is
- * notified (with a {@code null} envelope). A missing, {@code null}, or blank string (empty or
- * ASCII-whitespace-only: space, tab, LF, VT, FF, CR) property changes nothing, and a supported
- * version still goes through the post-decode {@link EnvelopeCodec#accepts} check.
+ * {@code 1.5}), string ({@code "2"}, {@code "x"}, {@code "01"}, {@code " 1"}) or type is rejected. A missing,
+ * {@code null}, or blank string (empty or ASCII-whitespace-only: space, tab, LF, VT, FF, CR) property changes
+ * nothing, and a supported version still goes through the post-decode {@link EnvelopeCodec#accepts} check. This
+ * consumer has no unknown-URN strategy option, so the configured strategy is {@code fail}; {@code onError} is
+ * notified (with a {@code null} envelope). With a {@link Builder#deadLetterSender dead-letter sender} the body is
+ * sent to {@code <queue>.dlq} <b>exactly as received</b> (no {@code dead_letter} block) and the message
+ * {@code Complete}d only after the send succeeded (a failed send {@code Abandon}s it, so the gate rejects the
+ * redelivery again and the write is retried). With none it is handed to the broker's native mechanism: explicitly
+ * <b>dead-lettered</b> (reason {@code unsupported schema version}) into {@code $DeadLetterQueue} — never abandoned,
+ * which would redeliver it until {@code MaxDeliveryCount}.
+ *
+ * <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 by the SDK. With a
+ * dead-letter sender the ADR-0035 record ({@link PoisonRecord}, {@code reason: "poison"}, {@code original_queue} =
+ * the receiver's entity) is sent to it and the message {@code Complete}d only after the send succeeded (a failed
+ * send {@code Abandon}s it). With none it is {@code Abandon}ed and reported — left to {@code MaxDeliveryCount}, which
+ * moves it to the native sub-queue.
  */
 public final class AsbConsumer {
 
@@ -62,6 +80,7 @@ public interface UnknownUrnHandler {
     private final Duration maxWaitTime;
     private final ErrorHandler onError;
     private final UnknownUrnHandler onUnknownUrn;
+    private final ServiceBusSenderClient deadLetterSender;
 
     private AsbConsumer(Builder builder) {
         this.receiver = builder.receiver;
@@ -70,6 +89,7 @@ private AsbConsumer(Builder builder) {
         this.maxWaitTime = builder.maxWaitTime;
         this.onError = builder.onError;
         this.onUnknownUrn = builder.onUnknownUrn;
+        this.deadLetterSender = builder.deadLetterSender;
     }
 
     public static Builder builder(ServiceBusReceiverClient receiver) {
@@ -106,21 +126,49 @@ private void handle(ServiceBusReceivedMessage message) {
             Object declared = message.getApplicationProperties().get(SCHEMA_VERSION_PROPERTY);
             String description = "Rejected a BabelQueue message with unsupported " + SCHEMA_VERSION_PROPERTY
                 + " property '" + truncateForReport(declared) + "' (supported: " + EnvelopeCodec.SCHEMA_VERSION + "); body not decoded.";
-            // Settle first: the message is terminal, so a throwing onError must not leave it unsettled.
-            receiver.deadLetter(message, new DeadLetterOptions()
-                .setDeadLetterReason(SCHEMA_VERSION_DEAD_LETTER_REASON)
-                .setDeadLetterErrorDescription(description));
+            if (deadLetterSender == null) {
+                // Native (rule 3): an explicit DeadLetter into $DeadLetterQueue, not an Abandon that would
+                // redeliver it until MaxDeliveryCount. Settle first: the message is terminal, so a throwing
+                // onError must not leave it unsettled.
+                receiver.deadLetter(message, new DeadLetterOptions()
+                    .setDeadLetterReason(SCHEMA_VERSION_DEAD_LETTER_REASON)
+                    .setDeadLetterErrorDescription(description));
+                report(new BabelQueueException(description), null, message);
+                return;
+            }
+            // Rule 4: the raw body, byte for byte, to <queue>.dlq; Complete only after the send. Rule 5: a failed
+            // send Abandons it (the gate rejects the redelivery again, so the write is retried).
+            try {
+                send(rawMessage(message));
+            } catch (RuntimeException sendFailure) {
+                receiver.abandon(message);
+                report(new BabelQueueException(description), null, message);
+                report(sendFailure, null, message);
+                return;
+            }
+            receiver.complete(message);
             report(new BabelQueueException(description), null, message);
             return;
         }
 
-        String body = message.getBody() == null ? "" : message.getBody().toString();
+        // Strict UTF-8: BinaryData.toString() would turn a bad byte into U+FFFD and let a corrupted body decode as a
+        // valid envelope. A body that is not valid UTF-8 is poison (ADR-0037 decision 2), never decoded.
+        String body = strictUtf8(bodyBytes(message));
+        if (body == null) {
+            BabelQueueException cause = new BabelQueueException(
+                "Rejected a BabelQueue message whose body is not valid UTF-8 from Azure Service Bus.");
+            Envelope none = EnvelopeCodec.decode("");
+            report(cause, none, message);
+            settlePoison(message, none, cause);
+            return;
+        }
         Envelope envelope = reconcile(EnvelopeCodec.decode(body), message.getDeliveryCount());
 
         if (!EnvelopeCodec.accepts(envelope)) {
-            report(new BabelQueueException("Rejected a non-conformant BabelQueue envelope from Azure Service Bus."),
-                envelope, message);
-            receiver.abandon(message);
+            BabelQueueException cause = new BabelQueueException(
+                "Rejected a non-conformant BabelQueue envelope from Azure Service Bus.");
+            report(cause, envelope, message);
+            settlePoison(message, envelope, cause);
             return;
         }
 
@@ -147,6 +195,91 @@ private void handle(ServiceBusReceivedMessage message) {
         }
     }
 
+    /**
+     * Settles a poison body (ADR-0037 decision 2): the ADR-0035 record to the DLQ sender and {@code Complete} after the
+     * send; a failed send {@code Abandon}s it. With no sender it is {@code Abandon}ed (native: {@code MaxDeliveryCount}).
+     */
+    private void settlePoison(ServiceBusReceivedMessage message, Envelope envelope, Throwable cause) {
+        if (deadLetterSender == null) {
+            receiver.abandon(message);
+            return;
+        }
+        try {
+            String record = poisonRecord(bodyBytes(message), entityName(), (int) Math.max(message.getDeliveryCount() - 1, 0),
+                cause);
+            ServiceBusMessage forwarded = new ServiceBusMessage(BinaryData.fromString(record))
+                .setContentType("application/json");
+            if (message.getMessageId() != null) {
+                forwarded.setMessageId(message.getMessageId());
+            }
+            send(forwarded);
+        } catch (RuntimeException sendFailure) {
+            receiver.abandon(message);
+            report(sendFailure, envelope, message);
+            return;
+        }
+        receiver.complete(message);
+    }
+
+    /**
+     * The ADR-0035 DLQ record of a poison {@code body}: the one builder this consumer writes with (conformance
+     * {@code poison_dlq} drives it for every body).
+     */
+    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 bytes as text, or {@code null} when they are not valid UTF-8 (never U+FFFD substitution). */
+    private static String strictUtf8(byte[] body) {
+        try {
+            return java.nio.charset.StandardCharsets.UTF_8.newDecoder()
+                .onMalformedInput(java.nio.charset.CodingErrorAction.REPORT)
+                .onUnmappableCharacter(java.nio.charset.CodingErrorAction.REPORT)
+                .decode(java.nio.ByteBuffer.wrap(body)).toString();
+        } catch (java.nio.charset.CharacterCodingException invalid) {
+            return null;
+        }
+    }
+
+    /** The exact bytes of the message body (empty when it has none). */
+    private static byte[] bodyBytes(ServiceBusReceivedMessage message) {
+        return message.getBody() == null ? new byte[0] : message.getBody().toBytes();
+    }
+
+    /** The receiver's entity: a queue name, or the topic of a topic subscription. */
+    private String entityName() {
+        String path = receiver.getEntityPath();
+        if (path == null) {
+            return "";
+        }
+        int subscription = path.indexOf("/subscriptions/");
+        return subscription >= 0 ? path.substring(0, subscription) : path;
+    }
+
+    /** A copy of {@code message} for the DLQ: the exact body bytes, content type and message id; nothing re-encoded. */
+    private static ServiceBusMessage rawMessage(ServiceBusReceivedMessage message) {
+        ServiceBusMessage forwarded = new ServiceBusMessage(BinaryData.fromBytes(bodyBytes(message)));
+        if (message.getContentType() != null) {
+            forwarded.setContentType(message.getContentType());
+        }
+        if (message.getMessageId() != null) {
+            forwarded.setMessageId(message.getMessageId());
+        }
+        return forwarded;
+    }
+
+    private void send(ServiceBusMessage message) {
+        deadLetterSender.sendMessage(message);
+    }
+
     /**
      * Renders a rejected property value for the dead-letter description and the error message: at most the first
      * {@value #MAX_REPORTED_VALUE_LENGTH} characters, followed by {@code ...} when cut. Service Bus caps the
@@ -232,6 +365,7 @@ public static final class Builder {
         private Duration maxWaitTime;
         private ErrorHandler onError;
         private UnknownUrnHandler onUnknownUrn;
+        private ServiceBusSenderClient deadLetterSender;
 
         private Builder(ServiceBusReceiverClient receiver) {
             this.receiver = Objects.requireNonNull(receiver, "receiver");
@@ -270,6 +404,18 @@ public Builder onUnknownUrn(UnknownUrnHandler handler) {
             return this;
         }
 
+        /**
+         * Turns BabelQueue dead-lettering on (opt-in, ADR-0037 decision 2): a sender bound to the {@code <queue>.dlq}
+         * entity (an ordinary Service Bus queue, §4.5; it must exist). A poison body is sent to it as the ADR-0035
+         * record and a version-gated delivery as its raw body; the source message is {@code Complete}d only after the
+         * send succeeded. Unset (the default): a poison body is abandoned and a version-gated delivery is
+         * dead-lettered into the native {@code $DeadLetterQueue}. The sender belongs to the caller and is not closed.
+         */
+        public Builder deadLetterSender(ServiceBusSenderClient sender) {
+            this.deadLetterSender = sender;
+            return this;
+        }
+
         public AsbConsumer build() {
             return new AsbConsumer(this);
         }