Release v1.3.0 (ADR-0037)

.github/scripts/check-conformance-results.py

@@ -4,13 +4,14 @@
 usage: check-conformance-results.py <results.json> <vendor-dir> <expected-conformance-version>
 
 This package carries the `pulsar` binding only, so the in-scope manifest sections are those whose
-scope is `pulsar` (`pulsar`, `pulsar_exhaustion`); the `core` sections are run and reported by
+scope is `pulsar` (`pulsar`, `pulsar_exhaustion`, `version_gate_pulsar`); the `core` sections are run and reported by
 babelqueue-core. Fails (exit 1) when results.json
   * does not conform to schema/results.schema.json,
   * reports a conformance_version other than the pinned one, or bindings other than ["pulsar"],
-  * has a `fail` case, or a `skip` other than a `pulsar_exhaustion` row that carries an
-    `on_unknown_urn` strategy (this consumer has no such option; ADR-0036 decision 1 tells the Java,
-    .NET and Node runners to skip those rows) and cites ADR-0036,
+  * has a `fail` case, or a `skip` other than a `pulsar_exhaustion` / `version_gate_pulsar` row that
+    carries an `on_unknown_urn` strategy (this consumer has no such option; ADR-0036 decision 1 and
+    ADR-0037 decision 1 tell the Java, .NET and Node runners to skip those rows; a `version_gate_pulsar`
+    row with strategy `fail` is asserted, not skipped) and cites the ADR (ADR-0036 / ADR-0037),
   * omits a pulsar-scope section or a case id that the manifest defines for it, or lists a section
     that is not in scope.
 """
@@ -78,21 +79,29 @@ def main():
     if bad:
         errors.append("fail cases: " + ", ".join(bad))
 
-    strategy_rows = {
-        c["name"] for c in manifest["pulsar_exhaustion"]["cases"] if "on_unknown_urn" in c
+    # section -> (ADR that authorises the skip, rows that carry a non-default on_unknown_urn strategy)
+    skippable = {
+        "pulsar_exhaustion": ("ADR-0036", {
+            c["name"] for c in manifest["pulsar_exhaustion"]["cases"] if "on_unknown_urn" in c
+        }),
+        "version_gate_pulsar": ("ADR-0037", {
+            c["name"] for c in manifest["version_gate_pulsar"]["cases"]
+            if c.get("on_unknown_urn", "fail") != "fail"
+        }),
     }
     for s, cs in sections.items():
         for i, r in cs.items():
             if r.get("status") != "skip":
                 continue
-            if s != "pulsar_exhaustion" or i not in strategy_rows:
+            if s not in skippable or i not in skippable[s][1]:
                 errors.append(f"skip {s}/{i} is not an on_unknown_urn strategy row")
-            elif "ADR-0036" not in r.get("reason", ""):
-                errors.append(f"skip {s}/{i} does not cite ADR-0036")
-    skipped = {i for i, r in sections.get("pulsar_exhaustion", {}).items() if r.get("status") == "skip"}
-    if skipped != strategy_rows:
-        errors.append("the on_unknown_urn strategy rows must be reported as skip, not pass: "
-                      + ", ".join(sorted(strategy_rows ^ skipped)))
+            elif skippable[s][0] not in r.get("reason", ""):
+                errors.append(f"skip {s}/{i} does not cite {skippable[s][0]}")
+    for s, (_, rows) in skippable.items():
+        skipped = {i for i, r in sections.get(s, {}).items() if r.get("status") == "skip"}
+        if skipped != rows:
+            errors.append(f"the {s} on_unknown_urn strategy rows must be reported as skip, not pass: "
+                          + ", ".join(sorted(rows ^ skipped)))
 
     if errors:
         print("results.json check FAILED:", file=sys.stderr)

.github/workflows/ci.yml

@@ -12,7 +12,7 @@ jobs:
   test:
     runs-on: ubuntu-latest
     env:
-      BQ_CONFORMANCE_VERSION: 1.7.0
+      BQ_CONFORMANCE_VERSION: 1.8.0
     strategy:
       fail-fast: false
       matrix:
@@ -39,8 +39,8 @@ jobs:
         with:
           python-version: "3.12"
 
-      # Schema-valid, no `fail`, the only `skip`s are the pulsar_exhaustion on_unknown_urn rows
-      # (ADR-0036), and every pulsar-scope section and case is present.
+      # Schema-valid, no `fail`, the only `skip`s are the pulsar_exhaustion and version_gate_pulsar on_unknown_urn rows
+      # (ADR-0036, ADR-0037), and every pulsar-scope section and case is present.
       - name: Validate results.json
         run: |
           python -m pip install --quiet jsonschema==4.23.0
@@ -51,7 +51,7 @@ jobs:
     name: Conformance suite in sync
     runs-on: ubuntu-latest
     env:
-      BQ_CONFORMANCE_VERSION: 1.7.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,25 @@ The envelope wire format is versioned separately by `meta.schema_version`
 
 ## [Unreleased]
 
+## [1.3.0] - 2026-10-26
+
+MINOR (ADR-0037 decision 1). The envelope stays at `schema_version: 1`.
+
+### Changed (behaviour change)
+
+- **A version-gated message is no longer redelivered without limit.** A delivery whose `bq-schema-version`
+  property is present, not blank and not exactly `1` used to be negatively acknowledged forever. It is now
+  settled on its **first** delivery (never decoded, no handler or `onUnknownUrn` hook, no retry, whatever
+  `bq-attempts`, the redelivery count or `maxAttempts` say): with a `deadLetterProducer` the body is sent to
+  `<queue>.dlq` **exactly as received** (byte for byte, no `dead_letter` block, no re-encoding) and then
+  acknowledged; a failed write negative-acknowledges it, so the write is retried on redelivery. **Without a
+  dead-letter producer the message is acknowledged and dropped** (logged at `WARNING`). `onError` is still
+  notified, with a `null` envelope. This consumer has no unknown-URN strategy option, so the configured strategy
+  is `fail`.
+- Vendored the cross-SDK conformance suite v1.8.0 (`version_gate_pulsar` runner added; its non-`fail` strategy
+  rows are skipped, citing ADR-0037, because this consumer has no such option) and moved the CI pin and the
+  `results.json` check to it.
+
 ## [1.2.0] - 2026-10-25
 
 Implements ADR-0036 decisions 1 and 2 (post-R1 parity). The envelope stays at `schema_version: 1`.

README.md

@@ -15,7 +15,7 @@ the .NET, Python, Go and Node SDKs. Implements
 <dependency>
   <groupId>com.babelqueue</groupId>
   <artifactId>babelqueue-pulsar</artifactId>
-  <version>1.2.0</version>
+  <version>1.3.0</version>
 </dependency>
 ```
 
@@ -52,7 +52,11 @@ property (produced outside BabelQueue) falls back to the body's `job`/`urn`.
 Before decoding the body, the consumer version-gates on the `bq-schema-version` property
 (§5.7 step 3): a value that is not exactly the schema version the core supports, compared exactly as sent with no
 trimming (`"2"`, `"x"`, `"01"`, `" 1"`, …) is rejected without decoding — `onError` is called with a `null` envelope and the
-message is `negativeAcknowledge`d (this gate is not bounded by `maxAttempts`). A missing or blank (empty or
+message is settled on its **first** delivery (ADR-0037; never retried, outside `maxAttempts`): with a
+`deadLetterProducer` the body is sent to `<queue>.dlq` **exactly as received** (no `dead_letter` block) and then
+acknowledged — a failed write negative-acknowledges it, so the gate rejects the redelivery again and the write is
+retried; without one it is acknowledged and the drop is logged at `WARNING`. This consumer has no unknown-URN
+strategy option (it behaves as `fail`) and an `onUnknownUrn` hook never affects a gated message. A missing or blank (empty or
 ASCII-whitespace-only: space, tab, LF, VT, FF, CR; NBSP, U+0085, U+FEFF etc. are not blank) property changes nothing; a `"1"` property still goes through the post-decode `accepts()` check.
 
 ## Contract mapping (§5)
@@ -128,7 +132,8 @@ one it takes the bounded retry, then is dropped.
 **Upgrading:** before this release a failing message was redelivered without a limit; with the defaults it is now
 dropped at its third delivery. Raise `maxAttempts` to retry longer and set `deadLetterProducer` to keep what is
 exhausted. `build()` now also requires `client(...)`, `retryProducer(...)` or a declared `Shared`/`Key_Shared`
-`subscriptionType(...)` (see above). A message rejected by the `bq-schema-version` gate is unchanged (negative acknowledgement).
+`subscriptionType(...)` (see above). A message rejected by the `bq-schema-version` gate used to be negatively acknowledged without a limit; it is now
+settled on its first delivery (raw body to the DLQ and acknowledged, or acknowledged and dropped without a DLQ producer).
 
 ## Build & test
 

pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>com.babelqueue</groupId>
     <artifactId>babelqueue-pulsar</artifactId>
-    <version>1.2.0</version>
+    <version>1.3.0</version>
     <packaging>jar</packaging>
 
     <name>BabelQueue Apache Pulsar</name>

src/main/java/com/babelqueue/pulsar/PulsarConsumer.java

@@ -45,8 +45,11 @@
  * decoded: when it is present and is not exactly the schema version the core supports
  * ({@link EnvelopeCodec#SCHEMA_VERSION}, compared exactly as sent, as its canonical decimal string, with no
  * trimming — so {@code "2"}, {@code "x"}, {@code "01"} and {@code " 1"} are all unknown), the body is never decoded or routed; {@code onError}
- * is notified (with a {@code null} envelope) and the message is {@code negativeAcknowledge}d; this gate is
- * outside the max-attempts bound below. A missing or blank (empty or ASCII-whitespace-only: space, tab, LF, VT, FF, CR) property changes nothing, and a
+ * is notified (with a {@code null} envelope) and the message is settled on its <b>first</b> delivery
+ * (ADR-0037 decision 1; outside the max-attempts bound below, no retry): with a {@link Builder#deadLetterProducer
+ * dead-letter producer} the body is written to the DLQ exactly as received (no {@code dead_letter} block) and then
+ * acknowledged — a failed write {@code negativeAcknowledge}s it, so the write is retried; without one it is
+ * acknowledged and the drop is logged. The gate reads no strategy ({@code onUnknownUrn} never affects it). A missing or blank (empty or ASCII-whitespace-only: space, tab, LF, VT, FF, CR) property changes nothing, and a
  * {@code "1"} property still goes through the post-decode {@link EnvelopeCodec#accepts} check.
  *
  * <h2>Max attempts and poison (ADR-0036, broker-bindings §5.5)</h2>
@@ -175,10 +178,7 @@ private void handle(Message<byte[]> message) throws PulsarClientException {
         String declaredVersion = rawProperty(properties, SCHEMA_VERSION_PROPERTY);
         if (declaredVersion != null && !isBlankSchemaVersion(declaredVersion)
             && !declaredVersion.equals(Integer.toString(EnvelopeCodec.SCHEMA_VERSION))) {
-            report(new BabelQueueException("Rejected a Pulsar message: unsupported "
-                + SCHEMA_VERSION_PROPERTY + " property '" + declaredVersion + "' (supported: "
-                + EnvelopeCodec.SCHEMA_VERSION + "); body not decoded."), null, message);
-            consumer.negativeAcknowledge(message);
+            settleVersionGated(message, declaredVersion);
             return;
         }
 
@@ -266,6 +266,32 @@ record = EnvelopeCodec.encode(annotated).getBytes(StandardCharsets.UTF_8);
         writeDeadLetterAndAcknowledge(message, envelope, record, properties);
     }
 
+    /**
+     * Settles a version-gated delivery on its first delivery (ADR-0037 decision 1, broker-bindings §5.7): the body is
+     * never decoded, no handler or {@code onUnknownUrn} hook runs and nothing is retried, whatever {@code bq-attempts},
+     * the redelivery count or {@code maxAttempts} say. This consumer has no unknown-URN strategy option, so the
+     * configured strategy is {@code fail}: with a {@link Builder#deadLetterProducer dead-letter producer} the body is
+     * written to the DLQ <b>exactly as received</b> (no {@code dead_letter} block, no re-encoding) and the original is
+     * acknowledged only after the write succeeded (a failed write negative-acknowledges it, so the gate rejects the
+     * redelivery again and the write is retried); without one the delivery is acknowledged and the drop is logged.
+     */
+    private void settleVersionGated(Message<byte[]> message, String declaredVersion) throws PulsarClientException {
+        BabelQueueException cause = new BabelQueueException("Rejected a Pulsar message: unsupported "
+            + SCHEMA_VERSION_PROPERTY + " property '" + declaredVersion + "' (supported: "
+            + EnvelopeCodec.SCHEMA_VERSION + "); body not decoded.");
+        report(cause, null, message);
+        if (deadLetterProducer == null) {
+            LOGGER.log(System.Logger.Level.WARNING,
+                "BabelQueue Pulsar: dropped a version-gated message from {0} (no dead-letter producer configured): {1}",
+                queueName(message), cause.getMessage());
+            consumer.acknowledge(message);
+            return;
+        }
+        byte[] body = message.getValue() == null ? new byte[0] : message.getValue();
+        Map<String, String> original = message.getProperties();
+        writeDeadLetterAndAcknowledge(message, null, body, original == null ? Map.of() : original);
+    }
+
     /**
      * Routes a poison body (ADR-0036 decision 2): with a dead-letter producer it is written to the DLQ in the
      * ADR-0035 shape without any retry; without one it takes the bounded retry, then is dropped.

src/test/java/com/babelqueue/pulsar/PulsarConformanceResults.java

@@ -26,12 +26,12 @@
  * when the test JVM exits. With the variable unset nothing is written.
  *
  * <p>This package carries the {@code pulsar} binding only, so exactly the manifest sections whose scope is
- * {@code pulsar} are in scope ({@code pulsar}, {@code pulsar_exhaustion}); the {@code core} sections are run and
+ * {@code pulsar} are in scope ({@code pulsar}, {@code pulsar_exhaustion}, {@code version_gate_pulsar}); the {@code core} sections are run and
  * reported by {@code babelqueue-core}. Every case id the manifest defines for an in-scope section ends up in the
  * file: a case that was never executed is written as {@code fail}, so a runner that forgets a case cannot go
- * green. The only {@code skip}s are the {@code pulsar_exhaustion} rows with an {@code on_unknown_urn} strategy,
- * which this consumer does not have (ADR-0036 decision 1; the suite tells Java, .NET and Node runners to skip
- * them).
+ * green. The only {@code skip}s are the {@code pulsar_exhaustion} (ADR-0036 decision 1) and
+ * {@code version_gate_pulsar} (ADR-0037 decision 1) rows with an {@code on_unknown_urn} strategy other than
+ * {@code fail}, which this consumer does not have (the suite tells Java, .NET and Node runners to skip them).
  */
 final class PulsarConformanceResults {
 
@@ -49,7 +49,7 @@ private PulsarConformanceResults() {
     }
 
     /**
-     * Wraps {@code body}: a normal return is {@code pass}, a JUnit abort is {@code skip} (reason prefixed with
+     * Wraps {@code body}: a normal return is {@code pass}, a JUnit abort is {@code skip} (reason carries
      * the authorising ADR), anything else is {@code fail} and is rethrown.
      */
     static Executable record(String section, String id, Executable body) {
@@ -68,10 +68,10 @@ static Executable record(String section, String id, Executable body) {
         };
     }
 
-    /** Records {@code section}/{@code id} as skipped without running anything. */
+    /** Records {@code section}/{@code id} as skipped without running anything; {@code reason} cites its ADR. */
     static void skip(String section, String id, String reason) {
         install();
-        put(section, id, "skip", "ADR-0036: " + reason);
+        put(section, id, "skip", reason);
     }
 
     private static void put(String section, String id, String status, String reason) {

src/test/java/com/babelqueue/pulsar/PulsarConformanceResultsTest.java

@@ -27,10 +27,12 @@ private static JSONObject manifest() throws Exception {
     @Test
     void inScopeSectionsAreExactlyTheManifestsPulsarScopedOnes() throws Exception {
         Map<String, List<String>> expected = PulsarConformanceResults.expectedCases(manifest());
-        assertEquals(java.util.Set.of("pulsar", "pulsar_exhaustion"), expected.keySet());
+        assertEquals(java.util.Set.of("pulsar", "pulsar_exhaustion", "version_gate_pulsar"), expected.keySet());
         assertTrue(expected.get("pulsar").contains("property_projection"));
         assertTrue(expected.get("pulsar").contains("schema_version_gate/0"));
         assertTrue(expected.get("pulsar_exhaustion").contains("failed-reconciled-2-dlq-on-dead-letters"));
+        assertEquals(17, expected.get("version_gate_pulsar").size());
+        assertTrue(expected.get("version_gate_pulsar").contains("default-dlq-on-write-ok"));
     }
 
     @Test