Release v1.3.1

CHANGELOG.md

@@ -9,6 +9,12 @@ The envelope wire format is versioned separately by `meta.schema_version`
 
 ## [Unreleased]
 
+## [1.3.1] - 2026-10-26
+
+### Changed
+
+- The poison `dead_letter` block is now built by core `PoisonRecord.build(body, originalQueue, attempts, error, exception)` (babelqueue-core 1.14.0) instead of by hand; the DLQ record is byte-for-byte unchanged (`reason: "poison"`, `failed_at`, `lang: "java"`). The local `PoisonRecord` copy was removed.
+
 ## [1.3.0] - 2026-10-24
 
 ### Added

pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>com.babelqueue</groupId>
     <artifactId>babelqueue-kafka</artifactId>
-    <version>1.3.0</version>
+    <version>1.3.1</version>
     <packaging>jar</packaging>
 
     <name>BabelQueue Apache Kafka</name>
@@ -50,7 +50,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.13.0</babelqueue-core.version>
+        <babelqueue-core.version>1.14.0</babelqueue-core.version>
         <kafka.version>3.9.2</kafka.version>
         <junit.version>5.10.3</junit.version>
         <mockito.version>5.12.0</mockito.version>

src/main/java/com/babelqueue/kafka/KafkaConsumer.java

@@ -4,15 +4,14 @@
 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 java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
-import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -366,7 +365,9 @@ private void deadLetterPoison(ConsumerRecord<byte[], byte[]> record, Throwable c
         }
         requireProducer();
         byte[] body = record.value() == null ? new byte[0] : record.value();
-        String json = PoisonRecord.build(body, JsonValues.encode(poisonBlock(record, cause)));
+        String json = PoisonRecord.build(body, originalTopic(record),
+            Math.max(KafkaHeaders.integer(record.headers(), KafkaHeaders.ATTEMPTS, 0), 0),
+            cause.getMessage(), cause.getClass().getName());
         List<Header> headers;
         try {
             headers = KafkaHeaders.of(EnvelopeCodec.decode(json, warning -> { }));
@@ -380,18 +381,6 @@ private void deadLetterPoison(ConsumerRecord<byte[], byte[]> record, Throwable c
             json.getBytes(StandardCharsets.UTF_8), headers)).get();
     }
 
-    private static Map<String, Object> poisonBlock(ConsumerRecord<byte[], byte[]> record, Throwable cause) {
-        Map<String, Object> deadLetter = new LinkedHashMap<>();
-        deadLetter.put("reason", "poison");
-        deadLetter.put("error", cause.getMessage());
-        deadLetter.put("exception", cause.getClass().getName());
-        deadLetter.put("failed_at", System.currentTimeMillis());
-        deadLetter.put("original_queue", originalTopic(record));
-        deadLetter.put("attempts", Math.max(KafkaHeaders.integer(record.headers(), KafkaHeaders.ATTEMPTS, 0), 0));
-        deadLetter.put("lang", EnvelopeCodec.SOURCE_LANG);
-        return deadLetter;
-    }
-
     /**
      * Send and wait for the broker's acknowledgement; a failure surfaces so the caller never commits past it. The
      * original record {@code key} is carried unchanged (a null key stays null, an empty key stays empty, none is