Add the kafka-error-handling module

kafka-error-handling/src/main/java/com/ankurm/kafkaerrors/ErrorHandlerConfiguration.java

@@ -0,0 +1,114 @@
+package com.ankurm.kafkaerrors;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Profile;
+import org.springframework.kafka.core.KafkaOperations;
+import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
+import org.springframework.kafka.listener.DefaultErrorHandler;
+import org.springframework.boot.kafka.autoconfigure.KafkaProperties;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.springframework.util.backoff.FixedBackOff;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * The error handler, in three configurations selected by profile.
+ *
+ * <p>A {@code CommonErrorHandler} bean replaces the container factory's default one wholesale.
+ * That default is a {@code DefaultErrorHandler} with {@code SeekUtils.DEFAULT_BACK_OFF} and a
+ * recoverer that only logs &mdash; so out of the box a record is retried and then <b>dropped</b>,
+ * which is the behaviour most people are surprised by.
+ *
+ * @see <a href="../../../../../docs/02-default-error-handler.md">docs/02-default-error-handler.md</a>
+ */
+@Configuration(proxyBeanMethods = false)
+public class ErrorHandlerConfiguration {
+
+	private static final Logger log = LoggerFactory.getLogger(ErrorHandlerConfiguration.class);
+
+	/**
+	 * Publishes the exhausted record to {@code <topic>.DLT}. Without a recoverer bean the record
+	 * is logged and discarded; this is the one line that turns "we lost it" into "it is on a
+	 * topic we can replay".
+	 */
+	@Bean
+	@Profile("dlt")
+	DefaultErrorHandler dltErrorHandler(KafkaOperations<String, Object> template) {
+		DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template);
+		// Three attempts, one second apart, so the timing is visible in a transcript.
+		DefaultErrorHandler handler = new DefaultErrorHandler(recoverer, new FixedBackOff(1000L, 2));
+		// Classification matters more than the back-off. A PermanentFailure retried ten times is
+		// ten times the latency for the same answer, and nine extra log lines that look like an
+		// outage.
+		handler.addNotRetryableExceptions(Failures.PermanentFailure.class);
+		handler.setRetryListeners((record, ex, deliveryAttempt) ->
+				log.info("retry listener: attempt {} for offset {} ({})", deliveryAttempt,
+						record.offset(), ex.getClass().getSimpleName()));
+		return handler;
+	}
+
+	/**
+	 * The same thing, publishing raw bytes correctly.
+	 *
+	 * <p>{@code DeadLetterPublishingRecoverer} takes a map from value type to template. Give it
+	 * a {@code byte[]} template backed by {@code ByteArraySerializer} and a deserialization
+	 * failure lands on the DLT as the exact bytes that arrived, instead of as base64 of them.
+	 * Without this, replaying a poison-pill DLT republishes a quoted base64 string.
+	 */
+	// Three things had to be right here, and each one failed first:
+	//
+	// 1. KafkaAutoConfiguration's template is @ConditionalOnMissingBean(KafkaTemplate.class),
+	//    so declaring byteTemplate below REMOVES the auto-configured KafkaTemplate from the
+	//    context entirely. Once you declare one, you own all of them - hence jsonTemplate.
+	// 2. With two templates present, KafkaOperations<String, Object> no longer resolves; the
+	//    generic check runs before the qualifier, so @Qualifier alone does not rescue it.
+	// 3. The recoverer's constructor takes Map<Class<?>, KafkaOperations<?, ?>>, so wildcards
+	//    are what it wants anyway.
+	@Bean
+	@Profile("dltbytes")
+	DefaultErrorHandler byteAwareDltErrorHandler(
+			@Qualifier("jsonTemplate") KafkaOperations<?, ?> jsonTemplate,
+			@Qualifier("byteTemplate") KafkaOperations<?, ?> byteTemplate) {
+		Map<Class<?>, KafkaOperations<?, ?>> templates = new java.util.LinkedHashMap<>();
+		templates.put(byte[].class, byteTemplate);
+		templates.put(Object.class, jsonTemplate);
+		DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(templates);
+		DefaultErrorHandler handler = new DefaultErrorHandler(recoverer, new FixedBackOff(0L, 0));
+		handler.addNotRetryableExceptions(Failures.PermanentFailure.class);
+		return handler;
+	}
+
+	/**
+	 * Replaces the auto-configured template, which backed off the moment {@code byteTemplate}
+	 * was declared.
+	 */
+	@Bean
+	@Profile("dltbytes")
+	KafkaTemplate<String, Object> jsonTemplate(KafkaProperties properties) {
+		return new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(properties.buildProducerProperties()));
+	}
+
+	@Bean
+	@Profile("dltbytes")
+	KafkaTemplate<String, byte[]> byteTemplate(KafkaProperties properties) {
+		Map<String, Object> config = new HashMap<>(properties.buildProducerProperties());
+		config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
+		return new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(config));
+	}
+
+	/** The stock handler, declared explicitly so a test can read its behaviour. */
+	@Bean
+	@Profile("defaults")
+	DefaultErrorHandler defaultErrorHandler() {
+		return new DefaultErrorHandler();
+	}
+
+}

kafka-error-handling/src/main/java/com/ankurm/kafkaerrors/ErrorHandlingApplication.java

@@ -0,0 +1,20 @@
+package com.ankurm.kafkaerrors;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+/**
+ * What Spring Kafka does when your listener throws, and what it does when the record cannot even
+ * be turned into an object.
+ *
+ * <p>Those are two different failures with two different cures, and conflating them is why a
+ * poison pill takes a partition down for hours.
+ */
+@SpringBootApplication
+public class ErrorHandlingApplication {
+
+	public static void main(String[] args) {
+		SpringApplication.run(ErrorHandlingApplication.class, args);
+	}
+
+}

kafka-error-handling/src/main/java/com/ankurm/kafkaerrors/Failures.java

@@ -0,0 +1,33 @@
+package com.ankurm.kafkaerrors;
+
+/**
+ * The two failure classes that Spring Kafka treats completely differently, made explicit so the
+ * tests can be about the distinction rather than about a generic RuntimeException.
+ */
+public final class Failures {
+
+	/** Worth retrying: a timeout, a 503, a deadlock. The same input may succeed later. */
+	public static class TransientFailure extends RuntimeException {
+
+		public TransientFailure(String message) {
+			super(message);
+		}
+
+	}
+
+	/**
+	 * Not worth retrying: a validation failure, a missing referenced entity, a malformed field.
+	 * The same input will fail identically ten times, and retrying it is pure latency.
+	 */
+	public static class PermanentFailure extends RuntimeException {
+
+		public PermanentFailure(String message) {
+			super(message);
+		}
+
+	}
+
+	private Failures() {
+	}
+
+}

kafka-error-handling/src/main/java/com/ankurm/kafkaerrors/Payment.java

@@ -0,0 +1,10 @@
+package com.ankurm.kafkaerrors;
+
+/** The payload. {@code paymentId} doubles as the record key. */
+public record Payment(String paymentId, String status) {
+
+	public static Payment of(String paymentId) {
+		return new Payment(paymentId, "PENDING");
+	}
+
+}

kafka-error-handling/src/main/java/com/ankurm/kafkaerrors/PaymentListener.java

@@ -0,0 +1,59 @@
+package com.ankurm.kafkaerrors;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+/**
+ * A listener that fails on demand, so the retry machinery can be observed rather than described.
+ *
+ * <p>Payment ids beginning {@code transient-} throw a retryable exception; ids beginning
+ * {@code permanent-} throw a non-retryable one; everything else succeeds.
+ *
+ * @see <a href="../../../../../docs/02-default-error-handler.md">docs/02-default-error-handler.md</a>
+ */
+@Component
+public class PaymentListener {
+
+	private static final Logger log = LoggerFactory.getLogger(PaymentListener.class);
+
+	public static final String TOPIC = "payments";
+
+	/** Every delivery attempt, with the wall-clock time it happened. */
+	public record Attempt(String paymentId, long atMillis, int partition, long offset) {
+	}
+
+	private final List<Attempt> attempts = new CopyOnWriteArrayList<>();
+
+	@KafkaListener(topics = TOPIC, groupId = "payments")
+	public void onPayment(ConsumerRecord<String, Payment> record) {
+		Payment payment = record.value();
+		this.attempts.add(new Attempt(payment.paymentId(), System.currentTimeMillis(),
+				record.partition(), record.offset()));
+		log.info("attempt {} for {}", this.attempts.size(), payment.paymentId());
+		if (payment.paymentId().startsWith("transient-")) {
+			throw new Failures.TransientFailure("downstream unavailable for " + payment.paymentId());
+		}
+		if (payment.paymentId().startsWith("permanent-")) {
+			throw new Failures.PermanentFailure("invalid payment " + payment.paymentId());
+		}
+	}
+
+	public List<Attempt> attempts() {
+		return this.attempts;
+	}
+
+	public List<Attempt> attemptsFor(String paymentId) {
+		return this.attempts.stream().filter((a) -> a.paymentId().equals(paymentId)).toList();
+	}
+
+	public void clear() {
+		this.attempts.clear();
+	}
+
+}

kafka-error-handling/src/main/java/com/ankurm/kafkaerrors/RetryableTopicListener.java

@@ -0,0 +1,85 @@
+package com.ankurm.kafkaerrors;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.context.annotation.Profile;
+import org.springframework.kafka.annotation.BackOff;
+import org.springframework.kafka.annotation.DltHandler;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.annotation.RetryableTopic;
+import org.springframework.kafka.retrytopic.SameIntervalTopicReuseStrategy;
+import org.springframework.kafka.support.KafkaHeaders;
+import org.springframework.messaging.handler.annotation.Header;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+/**
+ * Non-blocking retries. Instead of holding the consumer thread while it re-delivers,
+ * {@code @RetryableTopic} republishes the failed record to a separate retry topic with a
+ * timestamp header, and a container for that topic waits before processing it.
+ *
+ * <p>The consequence people miss: <b>the main partition advances immediately</b>. That is the
+ * entire point &mdash; and it is also why per-key ordering is gone for any record that fails.
+ * Record 2 for the same key is processed while record 1 is sitting in a retry topic.
+ *
+ * @see <a href="../../../../../docs/05-retryable-topic.md">docs/05-retryable-topic.md</a>
+ */
+@Component
+@Profile("retrytopic")
+public class RetryableTopicListener {
+
+	private static final Logger log = LoggerFactory.getLogger(RetryableTopicListener.class);
+
+	public static final String TOPIC = "invoices";
+
+	public record Delivery(String topic, String paymentId, long atMillis) {
+	}
+
+	private final List<Delivery> deliveries = new CopyOnWriteArrayList<>();
+
+	private final List<Delivery> dead = new CopyOnWriteArrayList<>();
+
+	// backOff, not backoff - and org.springframework.kafka.annotation.BackOff, not
+	// org.springframework.retry.annotation.Backoff. Spring Kafka 4.x dropped the spring-retry
+	// dependency and brought its own annotation, so every pre-4.x @RetryableTopic example is a
+	// compile error: "package org.springframework.retry.annotation does not exist".
+	@RetryableTopic(attempts = "4", backOff = @BackOff(delay = 500, multiplier = 2.0),
+			sameIntervalTopicReuseStrategy = SameIntervalTopicReuseStrategy.SINGLE_TOPIC,
+			exclude = Failures.PermanentFailure.class)
+	@KafkaListener(topics = TOPIC, groupId = "invoices")
+	public void onInvoice(ConsumerRecord<String, Payment> record,
+			@Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
+		this.deliveries.add(new Delivery(topic, record.value().paymentId(), System.currentTimeMillis()));
+		log.info("delivery {} on topic {}", this.deliveries.size(), topic);
+		if (record.value().paymentId().startsWith("transient-")) {
+			throw new Failures.TransientFailure("still failing");
+		}
+		if (record.value().paymentId().startsWith("permanent-")) {
+			throw new Failures.PermanentFailure("never going to work");
+		}
+	}
+
+	/**
+	 * Where a record lands once the attempts are exhausted. Without a {@code @DltHandler} the
+	 * framework logs it and moves on &mdash; the topic still exists and still has the record, but
+	 * nothing in your application has looked at it.
+	 */
+	@DltHandler
+	public void onDlt(ConsumerRecord<String, Payment> record,
+			@Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
+		this.dead.add(new Delivery(topic, record.value().paymentId(), System.currentTimeMillis()));
+		log.info("DLT handler: {} on {}", record.value().paymentId(), topic);
+	}
+
+	public List<Delivery> deliveries() {
+		return this.deliveries;
+	}
+
+	public List<Delivery> dead() {
+		return this.dead;
+	}
+
+}

kafka-error-handling/src/main/resources/application.yaml

@@ -0,0 +1,27 @@
+spring:
+  application:
+    name: kafka-error-handling
+  main:
+    banner-mode: off
+  kafka:
+    bootstrap-servers: localhost:9092
+    producer:
+      key-serializer: org.apache.kafka.common.serialization.StringSerializer
+      value-serializer: org.springframework.kafka.support.serializer.JacksonJsonSerializer
+    consumer:
+      group-id: payments
+      auto-offset-reset: earliest
+      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+      # ErrorHandlingDeserializer wraps the real one. Without it a record that cannot be
+      # deserialized is thrown BEFORE the listener exists, the container cannot advance past it,
+      # and the same offset is retried forever. See docs/03-poison-pills.md.
+      value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
+      properties:
+        spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JacksonJsonDeserializer
+        spring.json.trusted.packages: com.ankurm.kafkaerrors
+        spring.json.value.default.type: com.ankurm.kafkaerrors.Payment
+logging:
+  level:
+    root: WARN
+    com.ankurm: INFO
+    org.springframework.kafka.listener: WARN

kafka-error-handling/src/test/java/com/ankurm/kafkaerrors/ByteAwareDltTest.java

@@ -0,0 +1,83 @@
+package com.ankurm.kafkaerrors;
+
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.test.EmbeddedKafkaBroker;
+import org.springframework.kafka.test.context.EmbeddedKafka;
+import org.springframework.kafka.test.utils.KafkaTestUtils;
+import org.springframework.test.context.ActiveProfiles;
+
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * The fix for the base64 problem in {@link DeadLetterTopicTest}: a per-value-type template map
+ * so that a {@code byte[]} is published with a {@code ByteArraySerializer}.
+ *
+ * @see <a href="../../../../../docs/03-poison-pills.md">docs/03-poison-pills.md</a>
+ */
+@SpringBootTest
+@ActiveProfiles({ "test", "dltbytes" })
+@EmbeddedKafka(topics = { "payments", "payments-dlt" }, partitions = 1)
+class ByteAwareDltTest {
+
+	@Autowired
+	EmbeddedKafkaBroker broker;
+
+	@Test
+	void aByteArrayTemplateKeepsTheOriginalBytesIntact() {
+		Map<String, Object> producerProps = new HashMap<>(KafkaTestUtils.producerProps(this.broker.getBrokersAsString()));
+		producerProps.put("key.serializer", StringSerializer.class);
+		producerProps.put("value.serializer", ByteArraySerializer.class);
+		DefaultKafkaProducerFactory<String, byte[]> factory = new DefaultKafkaProducerFactory<>(producerProps);
+		try {
+			new KafkaTemplate<>(factory).send(PaymentListener.TOPIC, "poison-2",
+					"{ this is not json".getBytes(StandardCharsets.UTF_8));
+		}
+		finally {
+			factory.destroy();
+		}
+
+		Map<String, Object> consumerProps = new HashMap<>(
+				KafkaTestUtils.consumerProps(this.broker.getBrokersAsString(), "byte-dlt-reader", true));
+		consumerProps.put("key.deserializer", StringDeserializer.class);
+		consumerProps.put("value.deserializer", ByteArrayDeserializer.class);
+		consumerProps.put("auto.offset.reset", "earliest");
+		try (Consumer<String, byte[]> consumer = new KafkaConsumer<>(consumerProps)) {
+			consumer.subscribe(List.of("payments-dlt"));
+			long deadline = System.currentTimeMillis() + 30_000;
+			while (System.currentTimeMillis() < deadline) {
+				ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofMillis(500));
+				for (ConsumerRecord<String, byte[]> record : records) {
+					if ("poison-2".equals(record.key())) {
+						String payload = new String(record.value(), StandardCharsets.UTF_8);
+						System.out.println("=== byte-aware DLT ===");
+						System.out.println("  DLT payload -> " + payload);
+						// Byte for byte what was published. No base64, no quotes. Replay is now
+						// a copy from one topic to another.
+						assertThat(payload).isEqualTo("{ this is not json");
+						return;
+					}
+				}
+			}
+			throw new AssertionError("nothing arrived on payments-dlt");
+		}
+	}
+
+}

kafka-error-handling/src/test/java/com/ankurm/kafkaerrors/DeadLetterTopicTest.java

@@ -0,0 +1,192 @@
+package com.ankurm.kafkaerrors;
+
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer;
+import org.springframework.kafka.test.EmbeddedKafkaBroker;
+import org.springframework.kafka.test.context.EmbeddedKafka;
+import org.springframework.kafka.test.utils.KafkaTestUtils;
+import org.springframework.test.context.ActiveProfiles;
+
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+/**
+ * Retries, classification, and the two very different reasons a record ends up on the DLT.
+ *
+ * @see <a href="../../../../../docs/04-the-dlt.md">docs/04-the-dlt.md</a>
+ */
+@SpringBootTest
+@ActiveProfiles({ "test", "dlt" })
+@EmbeddedKafka(topics = { "payments", "payments-dlt" }, partitions = 1)
+class DeadLetterTopicTest {
+
+	@Autowired
+	KafkaTemplate<String, Payment> template;
+
+	@Autowired
+	PaymentListener listener;
+
+	@Autowired
+	EmbeddedKafkaBroker broker;
+
+	@BeforeEach
+	void clear() {
+		this.listener.clear();
+	}
+
+	/**
+	 * Reads the DLT as raw bytes, because a dead-lettered record may be exactly the thing that
+	 * could not be deserialized.
+	 *
+	 * <p>Note the topic name: {@code payments-dlt}. {@code RetryTopicConstants.DEFAULT_DLT_SUFFIX}
+	 * is {@code "-dlt"}, not {@code ".DLT"} &mdash; and if the topic does not exist, the
+	 * recoverer logs a WARN and the record is gone.
+	 */
+	private Consumer<String, byte[]> dltConsumer() {
+		Map<String, Object> props = new HashMap<>(
+				KafkaTestUtils.consumerProps(this.broker.getBrokersAsString(), "dlt-reader-" + System.nanoTime(), true));
+		props.put("key.deserializer", org.apache.kafka.common.serialization.StringDeserializer.class);
+		props.put("value.deserializer", org.apache.kafka.common.serialization.ByteArrayDeserializer.class);
+		props.put("auto.offset.reset", "earliest");
+		Consumer<String, byte[]> consumer = new org.apache.kafka.clients.consumer.KafkaConsumer<>(props);
+		consumer.subscribe(java.util.List.of("payments-dlt"));
+		return consumer;
+	}
+
+	@Test
+	void aRetryableFailureIsRetriedAndThenPublishedToTheDlt() {
+		this.template.send(PaymentListener.TOPIC, "transient-1", Payment.of("transient-1"));
+
+		// FixedBackOff(1000, 2) = one delivery plus two retries.
+		await().atMost(Duration.ofSeconds(30))
+				.until(() -> this.listener.attemptsFor("transient-1").size() >= 3);
+		assertThat(this.listener.attemptsFor("transient-1")).hasSize(3);
+
+		var attempts = this.listener.attemptsFor("transient-1");
+		long gap = attempts.get(1).atMillis() - attempts.get(0).atMillis();
+		System.out.println("=== transient failure ===");
+		System.out.println("  deliveries        " + attempts.size());
+		System.out.println("  gap between 1&2   " + gap + " ms (FixedBackOff interval 1000)");
+		assertThat(gap).isGreaterThanOrEqualTo(900);
+
+		try (Consumer<String, byte[]> consumer = dltConsumer()) {
+			ConsumerRecord<String, byte[]> dead = pollFor(consumer, "transient-1");
+			assertThat(dead).isNotNull();
+			printHeaders("transient failure on the DLT", dead);
+			// The TOP-LEVEL exception header is always the wrapper. Group a DLT triage dashboard
+			// by kafka_dlt-exception-fqcn and every listener failure in the estate lands in one
+			// bucket called ListenerExecutionFailedException. The useful field is the cause.
+			assertThat(header(dead, "kafka_dlt-exception-fqcn"))
+					.isEqualTo("org.springframework.kafka.listener.ListenerExecutionFailedException");
+			assertThat(header(dead, "kafka_dlt-exception-cause-fqcn"))
+					.isEqualTo(Failures.TransientFailure.class.getName());
+			assertThat(header(dead, "kafka_dlt-original-topic")).isEqualTo("payments");
+		}
+	}
+
+	@Test
+	void aNonRetryableFailureGoesStraightToTheDltWithNoRetries() {
+		this.template.send(PaymentListener.TOPIC, "permanent-1", Payment.of("permanent-1"));
+
+		try (Consumer<String, byte[]> consumer = dltConsumer()) {
+			ConsumerRecord<String, byte[]> dead = pollFor(consumer, "permanent-1");
+			assertThat(dead).isNotNull();
+			// addNotRetryableExceptions(PermanentFailure.class) means exactly one delivery.
+			System.out.println("=== permanent failure ===");
+			System.out.println("  deliveries        " + this.listener.attemptsFor("permanent-1").size());
+			assertThat(this.listener.attemptsFor("permanent-1")).hasSize(1);
+			assertThat(header(dead, "kafka_dlt-exception-cause-fqcn"))
+					.isEqualTo(Failures.PermanentFailure.class.getName());
+		}
+	}
+
+	@Test
+	void aPoisonPillNeverReachesTheListenerAndIsStillRecovered() {
+		// Raw bytes that are not valid JSON, published with a byte[] serializer so nothing on
+		// the producing side objects. This is what a schema change from another team looks like.
+		Map<String, Object> props = new HashMap<>(KafkaTestUtils.producerProps(this.broker.getBrokersAsString()));
+		props.put("key.serializer", StringSerializer.class);
+		props.put("value.serializer", ByteArraySerializer.class);
+		DefaultKafkaProducerFactory<String, byte[]> factory = new DefaultKafkaProducerFactory<>(props);
+		try {
+			new KafkaTemplate<>(factory).send(PaymentListener.TOPIC, "poison-1",
+					"{ this is not json".getBytes(StandardCharsets.UTF_8));
+		}
+		finally {
+			factory.destroy();
+		}
+
+		try (Consumer<String, byte[]> consumer = dltConsumer()) {
+			ConsumerRecord<String, byte[]> dead = pollFor(consumer, "poison-1");
+			assertThat(dead).isNotNull();
+			printHeaders("poison pill on the DLT", dead);
+
+			// The listener was never invoked - the failure happened in the deserializer, before
+			// the listener existed. ErrorHandlingDeserializer is what turns that into a record
+			// the error handler can recover instead of a poll that fails forever.
+			assertThat(this.listener.attemptsFor("poison-1")).isEmpty();
+			assertThat(header(dead, "kafka_dlt-exception-fqcn"))
+					.contains("DeserializationException");
+			// And here is the sting. The recoverer publishes with the APPLICATION's producer,
+			// whose value serializer is JacksonJsonSerializer. The failed value is a byte[], and
+			// Jackson writes a byte[] as a base64 JSON string. So the DLT does not hold the
+			// original bytes - it holds base64 of them, wrapped in quotes.
+			String payload = new String(dead.value(), StandardCharsets.UTF_8);
+			System.out.println("  DLT payload -> " + payload);
+			assertThat(payload).isEqualTo("\"eyB0aGlzIGlzIG5vdCBqc29u\"");
+			assertThat(new String(java.util.Base64.getDecoder().decode(
+					payload.replace("\"", "")), StandardCharsets.UTF_8)).isEqualTo("{ this is not json");
+		}
+	}
+
+	private ConsumerRecord<String, byte[]> pollFor(Consumer<String, byte[]> consumer, String key) {
+		long deadline = System.currentTimeMillis() + 30_000;
+		while (System.currentTimeMillis() < deadline) {
+			ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofMillis(500));
+			for (ConsumerRecord<String, byte[]> record : records) {
+				if (key.equals(record.key())) {
+					return record;
+				}
+			}
+		}
+		return null;
+	}
+
+	private static String header(ConsumerRecord<String, byte[]> record, String name) {
+		var header = record.headers().lastHeader(name);
+		return (header == null) ? null : new String(header.value(), StandardCharsets.UTF_8);
+	}
+
+	private static void printHeaders(String title, ConsumerRecord<String, byte[]> record) {
+		System.out.println("=== " + title + " ===");
+		record.headers().forEach((h) -> {
+			// Some of these headers are binary (the original partition, offset and timestamp are
+			// big-endian ints and longs, not text), so render anything non-printable as hex.
+			String value = new String(h.value(), StandardCharsets.UTF_8).replaceAll("\\s+", " ");
+			if (!value.chars().allMatch((c) -> c >= 0x20 && c < 0x7f)) {
+				StringBuilder hex = new StringBuilder("0x");
+				for (byte b : h.value()) {
+					hex.append("%02x".formatted(b));
+				}
+				value = hex.toString();
+			}
+			System.out.printf("  %-34s %s%n", h.key(), value.length() > 90 ? value.substring(0, 90) + "..." : value);
+		});
+	}
+
+}

kafka-error-handling/src/test/java/com/ankurm/kafkaerrors/DefaultBackOffTest.java

@@ -0,0 +1,48 @@
+package com.ankurm.kafkaerrors;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.kafka.listener.SeekUtils;
+import org.springframework.util.backoff.BackOffExecution;
+import org.springframework.util.backoff.FixedBackOff;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * What the stock {@code DefaultErrorHandler} actually does, read out of its own back-off rather
+ * than out of the documentation. No broker and no context: a {@code BackOff} is a state machine
+ * you can just run.
+ *
+ * @see <a href="../../../../../docs/02-default-error-handler.md">docs/02-default-error-handler.md</a>
+ */
+class DefaultBackOffTest {
+
+	@Test
+	void theDefaultIsTenAttemptsWithNoDelayAtAll() {
+		FixedBackOff backOff = SeekUtils.DEFAULT_BACK_OFF;
+		BackOffExecution execution = backOff.start();
+
+		List<Long> intervals = new ArrayList<>();
+		long next;
+		while ((next = execution.nextBackOff()) != BackOffExecution.STOP) {
+			intervals.add(next);
+		}
+
+		System.out.println("=== DefaultErrorHandler default back-off ===");
+		System.out.println("  interval          " + backOff.getInterval() + " ms");
+		System.out.println("  max attempts      " + backOff.getMaxAttempts() + " retries");
+		System.out.println("  SeekUtils.DEFAULT_MAX_FAILURES = " + SeekUtils.DEFAULT_MAX_FAILURES);
+		System.out.println("  retry intervals   " + intervals);
+		System.out.println("  total deliveries  " + (intervals.size() + 1));
+
+		// Nine retries after the first delivery = ten deliveries, matching DEFAULT_MAX_FAILURES.
+		assertThat(intervals).hasSize(SeekUtils.DEFAULT_MAX_FAILURES - 1);
+		// And every one of them is zero. The default is not "retry with backoff"; it is
+		// "hammer the same record ten times as fast as the consumer thread can go, then give up".
+		assertThat(intervals).containsOnly(0L);
+		assertThat(backOff.getInterval()).isZero();
+	}
+
+}

kafka-error-handling/src/test/java/com/ankurm/kafkaerrors/RetryableTopicTest.java

@@ -0,0 +1,82 @@
+package com.ankurm.kafkaerrors;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.test.EmbeddedKafkaBroker;
+import org.springframework.kafka.test.context.EmbeddedKafka;
+import org.springframework.test.context.ActiveProfiles;
+
+import java.time.Duration;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+/**
+ * Non-blocking retries, and the topics they create.
+ *
+ * @see <a href="../../../../../docs/05-retryable-topic.md">docs/05-retryable-topic.md</a>
+ */
+@SpringBootTest
+@ActiveProfiles({ "test", "retrytopic" })
+@EmbeddedKafka(topics = { "invoices" }, partitions = 1)
+class RetryableTopicTest {
+
+	@Autowired
+	KafkaTemplate<String, Payment> template;
+
+	@Autowired
+	RetryableTopicListener listener;
+
+	@Autowired
+	EmbeddedKafkaBroker broker;
+
+	@Test
+	void retriesHappenOnSeparateTopicsAndTheMainPartitionKeepsMoving() {
+		this.template.send(RetryableTopicListener.TOPIC, "transient-1", Payment.of("transient-1"));
+		// Sent immediately after the failing one, on the same partition. If retries were
+		// blocking, this could not be processed until the first record gave up.
+		this.template.send(RetryableTopicListener.TOPIC, "ok-1", Payment.of("ok-1"));
+
+		await().atMost(Duration.ofSeconds(60)).until(() -> !this.listener.dead().isEmpty());
+
+		List<RetryableTopicListener.Delivery> deliveries = this.listener.deliveries();
+		System.out.println("=== @RetryableTopic delivery trace ===");
+		long start = deliveries.get(0).atMillis();
+		deliveries.forEach((d) -> System.out.printf("  +%-6d ms  %-28s %s%n",
+				d.atMillis() - start, d.topic(), d.paymentId()));
+		System.out.println("  DLT: " + this.listener.dead().stream()
+				.map((d) -> d.paymentId() + " on " + d.topic()).toList());
+		System.out.println("  topics touched: " + deliveries.stream()
+				.map(RetryableTopicListener.Delivery::topic).distinct().sorted().toList());
+
+		// attempts = "4" means the first delivery plus three retries.
+		assertThat(deliveries).filteredOn((d) -> d.paymentId().equals("transient-1")).hasSize(4);
+
+		// Retries land on generated topics, not on the original one.
+		// Retry topics are named by the DELAY, not by the attempt number. With a multiplier,
+		// each distinct interval gets its own topic: invoices-retry-500, -1000, -2000. That is
+		// TopicSuffixingStrategy.SUFFIX_WITH_DELAY_VALUE, the default. Provisioning a cluster
+		// for this means knowing the whole back-off schedule in advance.
+		assertThat(deliveries).filteredOn((d) -> d.paymentId().equals("transient-1"))
+				.extracting(RetryableTopicListener.Delivery::topic)
+				.containsExactly("invoices", "invoices-retry-500", "invoices-retry-1000",
+						"invoices-retry-2000");
+		assertThat(this.listener.dead()).extracting(RetryableTopicListener.Delivery::topic)
+				.containsExactly("invoices-dlt");
+
+		// The good record was processed while the bad one was still being retried. That is the
+		// whole benefit, and the whole cost: ordering on this partition is gone.
+		long okAt = deliveries.stream().filter((d) -> d.paymentId().equals("ok-1"))
+				.findFirst().orElseThrow().atMillis();
+		long lastRetryAt = deliveries.stream().filter((d) -> d.paymentId().equals("transient-1"))
+				.mapToLong(RetryableTopicListener.Delivery::atMillis).max().orElseThrow();
+		assertThat(okAt).isLessThan(lastRetryAt);
+
+		assertThat(this.listener.dead()).extracting(RetryableTopicListener.Delivery::paymentId)
+				.containsExactly("transient-1");
+	}
+
+}

kafka-error-handling/src/test/resources/application-test.yaml

@@ -0,0 +1,3 @@
+spring:
+  kafka:
+    bootstrap-servers: ${spring.embedded.kafka.brokers}