Add the kafka-error-handling module

README.md

@@ -8,6 +8,14 @@ module's `scripts/run-all.sh`, never typed by hand.
 | Module | Article | What it demonstrates |
 |---|---|---|
 | [`kafka-basics/`](kafka-basics/README.md) | [Spring Boot 4.1 and Apache Kafka: Producer, Consumer and Serialisation from Scratch](https://ankurm.com/spring-boot-4-1-kafka-producer-consumer-serialisation/) | The on-ramp: what the starter gives you, the two Jackson serializer families, where a key lands and why, and which defaults are Kafka's rather than Spring's |
+| [`kafka-error-handling/`](kafka-error-handling/README.md) | [Kafka Error Handling with Spring Kafka 4.1: DLT, Retry Topics and Poison Pills](https://ankurm.com/spring-kafka-4-1-error-handling-dlt-retry-topics/) | What the default error handler really does, why a poison pill stops a partition, and what blocking retries cost that retry topics do not |
+| [`rabbitmq/`](rabbitmq/README.md) | [Spring Boot and RabbitMQ: Exchanges, Queues, Bindings and a Working Dead-Letter Queue](https://ankurm.com/spring-boot-rabbitmq-exchanges-dead-letter-queue/) | All four exchange types against a real broker, manual acknowledgement, and a dead-letter path exercised through both rejection and TTL expiry |
+
+The two brokers make an instructive pair. Kafka's consumer holds an offset and the broker
+remembers nothing about individual records; RabbitMQ's broker owns the message until it is
+acknowledged and can route, expire and dead-letter it on its own. Almost every difference in how
+you handle failure follows from that one sentence — which is why Kafka needs a retry topic
+to do what RabbitMQ does with a queue argument.
 
 They are meant to be read in order. `kafka-basics` establishes that the default acknowledgement
 mode is `BATCH` and therefore that delivery is at-least-once; everything the error-handling

kafka-error-handling/README.md

@@ -0,0 +1,59 @@
+# `kafka-error-handling` — retries, DLT and poison pills
+
+Companion project for
+[**Kafka Error Handling with Spring Kafka 4.1: DLT, Retry Topics and Poison Pills**](https://ankurm.com/spring-kafka-4-1-error-handling-dlt-retry-topics/)
+on ankurm.com.
+
+Six tests against a **real Kafka broker** started in-process in KRaft mode. No Docker, no local
+install. `./scripts/run-all.sh` regenerates everything under [`docs/output/`](docs/output/).
+
+## Versions
+
+| | Version |
+|---|---|
+| JDK | 25 (Temurin 25.0.4.1+1) |
+| Spring Boot | 4.1.1 |
+| Spring Kafka | 4.1.1 |
+| kafka-clients | 4.2.1 (Boot-managed) |
+
+## Profiles
+
+| Profile | What it wires |
+|---|---|
+| `dlt` | `DefaultErrorHandler` + `DeadLetterPublishingRecoverer`, `FixedBackOff(1000, 2)`, `PermanentFailure` classified non-retryable |
+| `dltbytes` | the same, with a `byte[]`-aware template map so poison pills keep their original bytes |
+| `retrytopic` | `@RetryableTopic` non-blocking retries with a `@DltHandler` |
+| `defaults` | the stock `DefaultErrorHandler`, for reading its behaviour |
+
+## Documentation
+
+1. [Two kinds of failure, and why they need different machinery](docs/01-two-kinds-of-failure.md)
+2. [What the default actually does](docs/02-default-error-handler.md)
+3. [Poison pills](docs/03-poison-pills.md)
+4. [The dead-letter topic](docs/04-the-dlt.md)
+5. [Non-blocking retries with `@RetryableTopic`](docs/05-retryable-topic.md)
+
+## Captured output
+
+| File | Shows |
+|---|---|
+| [`default-backoff.txt`](docs/output/default-backoff.txt) | ten deliveries, zero delay |
+| [`retry-and-dlt.txt`](docs/output/retry-and-dlt.txt) | measured back-off and the DLT headers |
+| [`poison-pill.txt`](docs/output/poison-pill.txt) | base64 payload, and the byte-aware fix |
+| [`retry-topics.txt`](docs/output/retry-topics.txt) | the non-blocking delivery trace |
+| [`tests.txt`](docs/output/tests.txt) | 6 tests |
+
+## Six things this module exists to prove
+
+1. **The default is ten deliveries, zero milliseconds apart, then the record is dropped.** Not
+   "retry with backoff", and not "dead-letter".
+2. **The DLT suffix is `-dlt`, not `.DLT`.** Get it wrong and the recoverer logs a WARN and the
+   record is lost — your safety net silently drops it.
+3. **`kafka_dlt-exception-fqcn` is always `ListenerExecutionFailedException`** for listener
+   failures. The useful header is `-exception-cause-fqcn`.
+4. **A poison pill reaches the DLT base64-encoded**, because the recoverer reuses the JSON
+   producer. A per-type template map fixes it; the module shows both transcripts.
+5. **`@RetryableTopic` names retry topics by delay** — `-retry-500`, `-retry-1000` — so changing
+   the multiplier renames them.
+6. **`@Backoff` from spring-retry no longer exists here.** Spring Kafka 4 ships its own
+   `@BackOff`, and the attribute is `backOff`.

kafka-error-handling/docs/01-two-kinds-of-failure.md

@@ -0,0 +1,53 @@
+[Module README](../README.md) · [DefaultErrorHandler →](02-default-error-handler.md)
+
+# 1. Two kinds of failure, and why they need different machinery
+
+Kafka delivery is at-least-once. The container commits offsets after your listener returns
+([the basics article](https://ankurm.com/spring-boot-4-1-kafka-producer-consumer-serialisation/)
+covers why), so if the listener throws, the offset does not move and the record comes back.
+Everything in this module is about what happens next.
+
+There are two failures, and conflating them is why a bad record can take a partition down for
+hours.
+
+## Failure inside the listener
+
+Your code threw. The record deserialized fine; the work failed. The container catches it, hands
+it to a `CommonErrorHandler`, and that decides whether to retry, how long to wait, and what to do
+when the attempts run out.
+
+This splits again, and the split matters more than any back-off setting:
+
+| | example | retrying it |
+|---|---|---|
+| **transient** | timeout, 503, deadlock, connection reset | may succeed |
+| **permanent** | validation failure, missing entity, malformed field | will fail identically |
+
+Retrying a permanent failure ten times buys nothing and costs ten times the latency plus nine
+misleading log lines. Spring Kafka lets you say so:
+
+```java
+handler.addNotRetryableExceptions(PermanentFailure.class);
+```
+
+That single line is worth more than tuning the back-off, and it is the one most people skip.
+
+## Failure before the listener
+
+The bytes on the topic are not what the deserializer expects. Someone changed a schema, or
+published with a different serializer, or your `__TypeId__` header names a class you do not
+trust.
+
+This one is nastier, because it happens **inside `poll()`**, before any listener exists to throw
+from. There is no error handler in the path. The consumer cannot advance past the offset,
+retries the same record on the next poll, fails again, and does that forever — at whatever rate
+the poll loop runs. The partition is stopped and the only symptom is a growing lag with a
+consumer that looks healthy.
+
+That is a **poison pill**, and the cure is a different mechanism from the one above:
+`ErrorHandlingDeserializer`, covered in [chapter 3](03-poison-pills.md).
+
+Keeping these two apart is the whole point of this module. One needs a retry policy; the other
+needs a wrapper around the deserializer. Neither fixes the other.
+
+[DefaultErrorHandler →](02-default-error-handler.md)

kafka-error-handling/docs/02-default-error-handler.md

@@ -0,0 +1,71 @@
+[← Two kinds of failure](01-two-kinds-of-failure.md) · [Module README](../README.md) · [Poison pills →](03-poison-pills.md)
+
+# 2. What the default actually does
+
+If you configure nothing, the container factory installs a `DefaultErrorHandler` with
+`SeekUtils.DEFAULT_BACK_OFF` and a recoverer that logs. Run the back-off and read it off
+([`docs/output/default-backoff.txt`](output/default-backoff.txt)):
+
+```
+=== DefaultErrorHandler default back-off ===
+  interval          0 ms
+  max attempts      9 retries
+  SeekUtils.DEFAULT_MAX_FAILURES = 10
+  retry intervals   [0, 0, 0, 0, 0, 0, 0, 0, 0]
+  total deliveries  10
+```
+
+**Ten deliveries, zero milliseconds apart, and then the record is dropped.**
+
+Both halves of that surprise people. It is not "retry with backoff" — it is ten immediate
+attempts as fast as the consumer thread can run them, which against a downstream that is
+overloaded is ten times the load at the worst moment. And "then dropped" means exactly that: the
+default recoverer logs the failure and the offset moves on. There is no dead-letter topic unless
+you make one.
+
+## Giving it a back-off and a destination
+
+```java
+@Bean
+DefaultErrorHandler errorHandler(KafkaOperations<String, Object> template) {
+    DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template);
+    DefaultErrorHandler handler = new DefaultErrorHandler(recoverer, new FixedBackOff(1000L, 2));
+    handler.addNotRetryableExceptions(PermanentFailure.class);
+    return handler;
+}
+```
+
+`FixedBackOff(1000L, 2)` is one delivery plus two retries. Measured
+([`docs/output/retry-and-dlt.txt`](output/retry-and-dlt.txt)):
+
+```
+=== transient failure ===
+  deliveries        3
+  gap between 1&2   1007 ms (FixedBackOff interval 1000)
+```
+
+and the classified permanent failure gets exactly one delivery before going to the DLT.
+
+**`ExponentialBackOffWithMaxRetries` is usually the better choice** than `FixedBackOff` for a
+transient downstream, because a fixed interval synchronises every consumer in the group into
+retrying at the same instant.
+
+## The cost of blocking retries
+
+`DefaultErrorHandler` retries **on the consumer thread**. For the whole back-off, that partition
+processes nothing else. `FixedBackOff(1000L, 2)` is three seconds of a stalled partition per
+failing record — fine. A one-minute exponential back-off over five attempts is five minutes, and
+if failures are correlated you have a stalled consumer group, not a retry policy.
+
+Two consequences worth planning for:
+
+- **`max.poll.interval.ms` is your ceiling.** Default five minutes. Block longer than that
+  between polls and the broker evicts the consumer from the group, triggering a rebalance —
+  which usually makes things worse. A back-off schedule that can exceed it is a bug.
+- **Ordering is preserved**, which is the one thing blocking retries give you that
+  [retry topics](05-retryable-topic.md) do not.
+
+That trade — ordering versus throughput under failure — is the real decision, and it is covered
+in [chapter 5](05-retryable-topic.md).
+
+[Poison pills &rarr;](03-poison-pills.md)

kafka-error-handling/docs/03-poison-pills.md

@@ -0,0 +1,95 @@
+[← DefaultErrorHandler](02-default-error-handler.md) · [Module README](../README.md) · [The DLT →](04-the-dlt.md)
+
+# 3. Poison pills
+
+A record whose bytes cannot be deserialized fails inside `poll()`, before any listener exists.
+No error handler is in the path. The offset cannot advance, so the next poll fetches the same
+record and fails identically. Forever.
+
+The consumer is up, the group is stable, no exception reaches your code, and lag grows. It is
+one of the few Kafka failures with no good symptom.
+
+## `ErrorHandlingDeserializer`
+
+```yaml
+spring:
+  kafka:
+    consumer:
+      value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
+      properties:
+        spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JacksonJsonDeserializer
+```
+
+It wraps the real deserializer, catches the failure, and returns a **null value with the
+exception in a header**. The record then flows normally into the container, the listener is
+skipped, and the error handler gets a record it can recover — which is to say, the poison pill
+becomes an ordinary failure.
+
+Use `spring.deserializer.key.delegate.class` for keys. A malformed key is rarer and just as
+fatal.
+
+## What arrives on the DLT
+
+From [`docs/output/poison-pill.txt`](output/poison-pill.txt):
+
+```
+kafka_dlt-exception-fqcn           org.springframework.kafka.support.serializer.DeserializationException
+kafka_dlt-exception-cause-fqcn     org.springframework.kafka.support.serializer.DeserializationException
+kafka_dlt-exception-message        failed to deserialize
+kafka_dlt-original-topic           payments
+kafka_dlt-original-partition       0x00000000
+kafka_dlt-original-offset          0x0000000000000000
+kafka_dlt-original-consumer-group  payments
+__TypeId__                         [B
+DLT payload -> "eyB0aGlzIGlzIG5vdCBqc29u"
+```
+
+The listener was never invoked — asserted in the test — and the record is off the partition,
+which is the whole win.
+
+Two details in that block are worth stopping on.
+
+**`original-partition` and `original-offset` are binary.** They are big-endian `int` and `long`,
+not text. Printing them as a string gives you mojibake, which is why the transcript renders them
+as hex. A DLT tool that treats every header as UTF-8 will show garbage for exactly the three
+fields you need in order to find the original record.
+
+**The payload is base64.** `"eyB0aGlzIGlzIG5vdCBqc29u"` decodes to `{ this is not json`. The
+recoverer publishes with the **application's** producer, whose value serializer is
+`JacksonJsonSerializer`; the failed value is a `byte[]`; Jackson writes a `byte[]` as a base64
+JSON string. So the DLT does not hold what arrived — it holds base64 of it, wrapped in quotes.
+
+Replaying that topic naively republishes a quoted base64 string, which fails to deserialize, and
+now you have a poison pill in your poison-pill queue.
+
+## The fix
+
+Give the recoverer a template per value type:
+
+```java
+Map<Class<?>, KafkaOperations<?, ?>> templates = new LinkedHashMap<>();
+templates.put(byte[].class, byteTemplate);     // ByteArraySerializer
+templates.put(Object.class, jsonTemplate);
+DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(templates);
+```
+
+```
+=== byte-aware DLT ===
+  DLT payload -> { this is not json
+```
+
+Byte for byte what was published. Replay is now a copy from one topic to another.
+
+Three things had to be right to get there, and each failed first — they are commented in
+[`ErrorHandlerConfiguration`](../src/main/java/com/ankurm/kafkaerrors/ErrorHandlerConfiguration.java):
+
+1. **`KafkaAutoConfiguration`'s template is `@ConditionalOnMissingBean(KafkaTemplate.class)`.**
+   Declaring `byteTemplate` removed the auto-configured `KafkaTemplate` from the context
+   entirely. Once you declare one template, you own all of them.
+2. With two templates present, `KafkaOperations<String, Object>` stops resolving, and
+   `@Qualifier` alone does **not** rescue it — the generic check runs first. Use
+   `KafkaOperations<?, ?>`, which is what the recoverer's constructor wants anyway.
+3. The map is `Map<Class<?>, KafkaOperations<?, ?>>`, matched by value type, with
+   `Object.class` as the fallback.
+
+[The DLT &rarr;](04-the-dlt.md)

kafka-error-handling/docs/04-the-dlt.md

@@ -0,0 +1,74 @@
+[← Poison pills](03-poison-pills.md) · [Module README](../README.md) · [Retry topics →](05-retryable-topic.md)
+
+# 4. The dead-letter topic
+
+## The suffix is `-dlt`, not `.DLT`
+
+```java
+public static final String RetryTopicConstants.DEFAULT_RETRY_SUFFIX = "-retry";
+public static final String RetryTopicConstants.DEFAULT_DLT_SUFFIX   = "-dlt";
+```
+
+Older Spring Kafka used `.DLT`, and most of the material online still says so. Getting it wrong
+is not an exception — it is this, at WARN, once per record:
+
+```
+o.s.k.l.DeadLetterPublishingRecoverer : Destination resolver returned non-existent partition
+payments-dlt-0, KafkaProducer will determine partition to use for this topic
+[Producer] ... {payments-dlt=UNKNOWN_TOPIC_OR_PARTITION}
+```
+
+and then, on a cluster with auto-topic-creation disabled, the record is **gone**. Your safety net
+dropped it and logged a warning. This module's tests were written against `payments.DLT` first
+and failed exactly this way.
+
+Two things follow: pre-create your DLT topics as part of provisioning, and alert on that WARN.
+
+## Same partition by default
+
+`DeadLetterPublishingRecoverer` publishes to the **same partition number** as the original. If
+your DLT has fewer partitions than the source topic, records from the high-numbered partitions
+have nowhere to go. Either give the DLT the same partition count, or set
+
+```java
+recoverer.setPartitionResolver((record, ex) -> null);   // let the producer choose
+```
+
+## The headers, and the one that will mislead you
+
+From [`docs/output/retry-and-dlt.txt`](output/retry-and-dlt.txt):
+
+```
+kafka_dlt-exception-fqcn           org.springframework.kafka.listener.ListenerExecutionFailedException
+kafka_dlt-exception-cause-fqcn     com.ankurm.kafkaerrors.Failures$TransientFailure
+kafka_dlt-original-topic           payments
+kafka_dlt-original-consumer-group  payments
+```
+
+**`kafka_dlt-exception-fqcn` is always the wrapper** for a listener failure. Build a DLT triage
+dashboard grouped by that header and every failure in the estate lands in one bucket called
+`ListenerExecutionFailedException`. The field you want is `kafka_dlt-exception-cause-fqcn`.
+
+(For a deserialization failure there is no wrapper, so the two headers agree. That inconsistency
+is worth knowing if you are writing a tool over them.)
+
+`kafka_dlt-original-consumer-group` is the one that saves you when several groups consume the
+same topic and share a DLT.
+
+## Replay
+
+A DLT is only useful if you can put records back. The mechanics are a copy:
+
+1. read from `<topic>-dlt` with a **byte-array** deserializer — the payload may be the thing that
+   could not be deserialized
+2. read `kafka_dlt-original-topic` and `kafka_dlt-original-consumer-group` to decide where it
+   belongs and whether it is yours
+3. republish to the original topic, **stripping the `kafka_dlt-*` headers** so a second failure
+   is not confused with the first
+4. do it deliberately, in bounded batches, after the cause is fixed
+
+Automatic replay is almost always wrong: the records are on the DLT precisely because something
+was not transient, and a loop that moves them back on a timer is a slow-motion outage. A replay
+you run by hand, having read the failure, is the tool worth building.
+
+[Retry topics &rarr;](05-retryable-topic.md)

kafka-error-handling/docs/05-retryable-topic.md

@@ -0,0 +1,89 @@
+[← The DLT](04-the-dlt.md) · [Module README](../README.md)
+
+# 5. Non-blocking retries with `@RetryableTopic`
+
+Blocking retries stall the partition. `@RetryableTopic` republishes the failed record to a
+separate topic and lets the main partition carry on.
+
+```java
+@RetryableTopic(attempts = "4", backOff = @BackOff(delay = 500, multiplier = 2.0),
+        sameIntervalTopicReuseStrategy = SameIntervalTopicReuseStrategy.SINGLE_TOPIC,
+        exclude = Failures.PermanentFailure.class)
+@KafkaListener(topics = "invoices", groupId = "invoices")
+public void onInvoice(ConsumerRecord<String, Payment> record, ...) { ... }
+```
+
+**Two API changes in Spring Kafka 4.x will stop older examples compiling:**
+
+- the attribute is **`backOff`**, not `backoff`
+- the annotation is **`org.springframework.kafka.annotation.BackOff`**, not
+  `org.springframework.retry.annotation.Backoff`. Spring Kafka 4 dropped the spring-retry
+  dependency and brought its own.
+
+The failure is `package org.springframework.retry.annotation does not exist`, which reads like a
+missing dependency and is not.
+
+Also new in 4.1: `sameIntervalTopicReuseStrategy` defaults to `SINGLE_TOPIC` in
+`RetryTopicConfigurationBuilder`, aligning it with the annotation's default.
+
+## What it actually does
+
+From [`docs/output/retry-topics.txt`](output/retry-topics.txt) — a failing record and a good one
+published back to back on the same partition:
+
+```
+=== @RetryableTopic delivery trace ===
+  +0      ms  invoices                     transient-1
+  +531    ms  invoices-retry-500           transient-1
+  +550    ms  invoices                     ok-1
+  +1554   ms  invoices-retry-1000          transient-1
+  +3560   ms  invoices-retry-2000          transient-1
+  DLT: [transient-1 on invoices-dlt]
+```
+
+Read the third line. `ok-1` was processed at +550 ms, while `transient-1` was still two retries
+from giving up. With a blocking handler it would have waited for the whole schedule.
+
+**Retry topics are named by the delay, not the attempt number.** `invoices-retry-500`,
+`invoices-retry-1000`, `invoices-retry-2000` — that is
+`TopicSuffixingStrategy.SUFFIX_WITH_DELAY_VALUE`, the default. So provisioning topics ahead of
+time means knowing your whole back-off schedule in advance, and **changing the multiplier changes
+the topic names**, orphaning whatever is still sitting in the old ones. Deploy that change the
+way you would a rename.
+
+## The cost
+
+**Per-key ordering is gone for any record that fails.** That is not a side effect; it is the
+mechanism. If `invoice-7` fails and `invoice-7`'s next event succeeds, they are processed out of
+order, and no configuration prevents it.
+
+So the decision is not "blocking or non-blocking", it is:
+
+| | blocking (`DefaultErrorHandler`) | non-blocking (`@RetryableTopic`) |
+|---|---|---|
+| ordering under failure | preserved | lost for the failing key |
+| partition throughput under failure | stalled | unaffected |
+| topics to provision | 1 + DLT | 1 + one per distinct delay + DLT |
+| long back-offs | limited by `max.poll.interval.ms` | unlimited |
+
+If your consumer is idempotent and order-insensitive — most notification, indexing and cache-warm
+consumers are — retry topics are strictly better. If it applies state transitions per key,
+blocking retries with a short schedule and a fast DLT are usually the safer answer.
+
+Use `exclude` (or `include`) rather than retrying everything: a `PermanentFailure` here skips the
+retry topics entirely and goes straight to `invoices-dlt`.
+
+## `@DltHandler`
+
+```java
+@DltHandler
+public void onDlt(ConsumerRecord<String, Payment> record,
+        @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) { ... }
+```
+
+Without one, the framework still creates and populates the DLT — it just logs and moves on, and
+nothing in your application has looked at the record. A `@DltHandler` that increments a counter
+and writes a structured log line is the minimum worth having, because a DLT nobody watches is a
+queue that grows until someone notices the disk.
+
+[Module README](../README.md)

kafka-error-handling/docs/output/default-backoff.txt

@@ -0,0 +1,6 @@
+=== DefaultErrorHandler default back-off ===
+  interval          0 ms
+  max attempts      9 retries
+  SeekUtils.DEFAULT_MAX_FAILURES = 10
+  retry intervals   [0, 0, 0, 0, 0, 0, 0, 0, 0]
+  total deliveries  10

kafka-error-handling/docs/output/poison-pill.txt

@@ -0,0 +1,15 @@
+=== poison pill on the DLT ===
+  kafka_dlt-exception-fqcn           org.springframework.kafka.support.serializer.DeserializationException
+  kafka_dlt-exception-cause-fqcn     org.springframework.kafka.support.serializer.DeserializationException
+  kafka_dlt-exception-message        failed to deserialize
+  kafka_dlt-exception-stacktrace     org.springframework.kafka.support.serializer.DeserializationException: failed to deseriali...
+  kafka_dlt-original-topic           payments
+  kafka_dlt-original-partition       0x00000000
+  kafka_dlt-original-offset          0x0000000000000000
+  kafka_dlt-original-timestamp       0x000001a04bd35409
+  kafka_dlt-original-timestamp-type  CreateTime
+  kafka_dlt-original-consumer-group  payments
+  __TypeId__                         [B
+  DLT payload -> "eyB0aGlzIGlzIG5vdCBqc29u"
+=== byte-aware DLT ===
+  DLT payload -> { this is not json

kafka-error-handling/docs/output/retry-and-dlt.txt

@@ -0,0 +1,15 @@
+=== transient failure ===
+  deliveries        3
+  gap between 1&2   1007 ms (FixedBackOff interval 1000)
+=== transient failure on the DLT ===
+  kafka_dlt-exception-fqcn           org.springframework.kafka.listener.ListenerExecutionFailedException
+  kafka_dlt-exception-cause-fqcn     com.ankurm.kafkaerrors.Failures$TransientFailure
+  kafka_dlt-exception-message        Listener method 'public void com.ankurm.kafkaerrors.PaymentListener.onPayment(org.apache.k...
+  kafka_dlt-exception-stacktrace     org.springframework.kafka.listener.ListenerExecutionFailedException: Listener method 'publ...
+  kafka_dlt-original-topic           payments
+  kafka_dlt-original-partition       0x00000000
+  kafka_dlt-original-offset          0x0000000000000002
+  kafka_dlt-original-timestamp       0x000001a04bd35f33
+  kafka_dlt-original-timestamp-type  CreateTime
+  kafka_dlt-original-consumer-group  payments
+  __TypeId__                         com.ankurm.kafkaerrors.Payment

kafka-error-handling/docs/output/retry-topics.txt

@@ -0,0 +1,8 @@
+=== @RetryableTopic delivery trace ===
+  +0      ms  invoices                     transient-1
+  +522    ms  invoices-retry-500           transient-1
+  +539    ms  invoices                     ok-1
+  +1548   ms  invoices-retry-1000          transient-1
+  +3554   ms  invoices-retry-2000          transient-1
+  DLT: [transient-1 on invoices-dlt]
+  topics touched: [invoices, invoices-retry-1000, invoices-retry-2000, invoices-retry-500]

kafka-error-handling/docs/output/tests.txt

@@ -0,0 +1,4 @@
+Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 9.009 s -- in com.ankurm.kafkaerrors.RetryableTopicTest
+Tests run: 3, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 6.809 s -- in com.ankurm.kafkaerrors.DeadLetterTopicTest
+Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 0.003 s -- in com.ankurm.kafkaerrors.DefaultBackOffTest
+Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 2.935 s -- in com.ankurm.kafkaerrors.ByteAwareDltTest

kafka-error-handling/pom.xml

@@ -0,0 +1,84 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0">
+  <modelVersion>4.0.0</modelVersion>
+
+  <!-- Inheriting spring-boot-starter-parent so spring-kafka, kafka-clients, Jackson and the
+       test stack are all Boot-managed. Boot 4.1.1 manages Spring Kafka 4.1.1 and
+       kafka-clients 4.3.1; do not pin those yourself. -->
+  <parent>
+    <groupId>org.springframework.boot</groupId>
+    <artifactId>spring-boot-starter-parent</artifactId>
+    <version>4.1.1</version>
+    <relativePath/>
+  </parent>
+
+  <groupId>com.ankurm</groupId>
+  <artifactId>kafka-error-handling</artifactId>
+  <version>1.0</version>
+  <packaging>jar</packaging>
+
+  <properties>
+    <java.version>25</java.version>
+    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+  </properties>
+
+  <dependencies>
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter</artifactId>
+    </dependency>
+    <!-- spring-boot-starter-kafka, NOT a bare org.springframework.kafka:spring-kafka
+         dependency. In Boot 4 the auto-configuration lives in the spring-boot-kafka module
+         (package org.springframework.boot.kafka.autoconfigure), which the starter brings and
+         spring-kafka does not. Depending on spring-kafka alone compiles, starts, and gives you
+         no KafkaTemplate bean: "No qualifying bean of type KafkaTemplate<...>". Every Boot 3
+         tutorial gets this wrong now. See docs/01-the-on-ramp.md. -->
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-kafka</artifactId>
+    </dependency>
+    <!-- JsonSerializer/JsonDeserializer need a Jackson ObjectMapper. Boot 4 moved to
+         Jackson 3 (tools.jackson), which changes the import in your own code. -->
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-jackson</artifactId>
+    </dependency>
+
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-test</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <!-- The broker used by the committed transcripts: a real Kafka broker in KRaft mode,
+         started in-process. No Docker required, which is why docs/output/ can be regenerated
+         anywhere. See docs/07-testing.md for the Testcontainers route. -->
+    <dependency>
+      <groupId>org.springframework.kafka</groupId>
+      <artifactId>spring-kafka-test</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-testcontainers</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <!-- NOT org.testcontainers:kafka. Testcontainers 2.x renamed every module with a
+         "testcontainers-" prefix, and Boot 4.1.1 imports testcontainers-bom 2.0.5, which
+         manages the new name only. The old coordinate stopped at 1.21.4 and fails the build
+         with "'dependencies.dependency.version' ... is missing", which does not mention the
+         rename. See docs/07-testing.md. -->
+    <dependency>
+      <groupId>org.testcontainers</groupId>
+      <artifactId>testcontainers-kafka</artifactId>
+      <scope>test</scope>
+    </dependency>
+  </dependencies>
+
+  <build>
+    <plugins>
+      <plugin>
+        <groupId>org.springframework.boot</groupId>
+        <artifactId>spring-boot-maven-plugin</artifactId>
+      </plugin>
+    </plugins>
+  </build>
+</project>

kafka-error-handling/scripts/run-all.sh

@@ -0,0 +1,14 @@
+#!/usr/bin/env bash
+# Regenerate every file under docs/output/. A real Kafka broker starts in-process in KRaft mode;
+# no Docker and no local install.
+set -eu
+cd "$(dirname "$0")/.."
+mvn -B test 2>&1 | tr -d '\000' | tee /tmp/eh-test.log > /dev/null
+sed -n '/=== DefaultErrorHandler default back-off ===/,/total deliveries/p' /tmp/eh-test.log > docs/output/default-backoff.txt
+sed -n '/=== transient failure ===/,/gap between/p' /tmp/eh-test.log > docs/output/retry-and-dlt.txt
+sed -n '/=== transient failure on the DLT ===/,/__TypeId__/p' /tmp/eh-test.log >> docs/output/retry-and-dlt.txt
+sed -n '/=== poison pill on the DLT ===/,/DLT payload/p' /tmp/eh-test.log > docs/output/poison-pill.txt
+sed -n '/=== byte-aware DLT ===/,/DLT payload/p' /tmp/eh-test.log >> docs/output/poison-pill.txt
+sed -n '/=== @RetryableTopic delivery trace ===/,/topics touched/p' /tmp/eh-test.log > docs/output/retry-topics.txt
+grep -aE 'Tests run:.*in com\.ankurm' /tmp/eh-test.log | sed 's/^\[INFO\] //' > docs/output/tests.txt
+echo "regenerated:"; ls -1 docs/output/

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}

rabbitmq/README.md

@@ -0,0 +1,80 @@
+# `rabbitmq` — exchanges, bindings and a dead-letter queue that works
+
+Companion project for
+[**Spring Boot and RabbitMQ: Exchanges, Queues, Bindings and a Working Dead-Letter Queue**](https://ankurm.com/spring-boot-rabbitmq-exchanges-dead-letter-queue/)
+on ankurm.com.
+
+Nine tests against a **real RabbitMQ broker**. All four exchange types, manual acknowledgement,
+a dead-letter path exercised through `basicNack` and through TTL expiry, and the three ways a
+topology loses work quietly.
+
+## Versions
+
+| | Version | Notes |
+|---|---|---|
+| JDK | 25 (Temurin 25.0.4.1+1) | current LTS |
+| Spring Boot | 4.1.1 | |
+| Spring AMQP / Spring Rabbit | 4.1.1 | Boot-managed |
+| `com.rabbitmq:amqp-client` | **5.30.0** | Boot-managed. Central has 5.35.0 |
+| Testcontainers | 2.0.5 | artifact is `testcontainers-rabbitmq`, not `rabbitmq` |
+| **Broker for the committed transcripts** | **RabbitMQ 3.10.25** | see the note below |
+
+Client-side versions were read from `repo1.maven.org/.../maven-metadata.xml` and Boot's
+`spring-boot-dependencies` POM.
+
+> **About the broker version.** The machine that regenerates `docs/output/` has no Docker daemon
+> and no root, so `scripts/broker.sh` starts a generic-unix RabbitMQ against an extracted Erlang
+> 24 runtime, and the newest release that pairs with Erlang 24 is 3.10.25. Everything exercised
+> here — the four exchange types, `x-dead-letter-exchange`, `x-dead-letter-routing-key`,
+> `x-message-ttl`, `x-death`, manual ack, `mandatory` returns, `PRECONDITION_FAILED` on
+> inequivalent arguments — is AMQP 0-9-1 behaviour that is unchanged in RabbitMQ 4.x. The one
+> broker-version difference worth knowing for 4.x is that **quorum queues are the default queue
+> type** and classic mirrored queues are gone; nothing in this module declares a queue type, so
+> the topology is valid on both. If you have Docker, run the same tests against
+> `rabbitmq:4.1-management` with
+> [`TestcontainersConfiguration`](src/test/java/com/ankurm/rabbit/TestcontainersConfiguration.java).
+
+## Quickstart
+
+```bash
+# with Docker: point the tests at Testcontainers, or run your own broker on 5672
+mvn test
+
+# without Docker:
+export ERL_ROOT=/path/to/erlang RABBITMQ_HOME=/path/to/rabbitmq_server-3.10.25
+./scripts/run-all.sh
+```
+
+## Documentation
+
+1. [The on-ramp](docs/01-the-on-ramp.md)
+2. [Four exchange types](docs/02-exchanges.md)
+3. [The message that goes nowhere and says nothing](docs/03-the-silent-drop.md)
+4. [A dead-letter queue that actually works](docs/04-dead-lettering.md)
+5. [Acknowledgement](docs/05-acknowledgement.md)
+6. [Changing your mind about a queue](docs/06-changing-your-mind.md)
+
+## Captured output
+
+| File | Shows |
+|---|---|
+| [`topic-wildcards.txt`](docs/output/topic-wildcards.txt) | `*` vs `#` over five routing keys |
+| [`dead-letter.txt`](docs/output/dead-letter.txt) | `x-death` for `rejected` and for `expired` |
+| [`requeue-loop.txt`](docs/output/requeue-loop.txt) | 199 redeliveries, 0 dead-lettered |
+| [`unroutable.txt`](docs/output/unroutable.txt) | `312 NO_ROUTE` |
+| [`precondition-failed.txt`](docs/output/precondition-failed.txt) | `406` on an inequivalent argument |
+| [`tests.txt`](docs/output/tests.txt) | 9 tests |
+
+## Five things this module exists to prove
+
+1. **An unroutable message is discarded silently**, and finding out needs `publisher-returns`
+   *and* `mandatory` — two settings in two different places. Setting only `mandatory` does
+   nothing.
+2. **`requeue=true` never dead-letters.** 200 attempts, 199 redeliveries, an empty DLQ and a
+   queue depth that stays at 1 the whole time.
+3. **`x-death.reason` distinguishes `rejected` from `expired`**, which is the difference between
+   a consumer that refused the work and a consumer that never got to it.
+4. **Queue arguments are immutable** — `406 PRECONDITION_FAILED`, wrapped in an exception whose
+   own message is the string `java.io.IOException`.
+5. **Boot auto-configures no JSON converter for RabbitMQ**, and the converter you want is
+   `JacksonJsonMessageConverter`, not `Jackson2JsonMessageConverter`.

rabbitmq/docs/01-the-on-ramp.md

@@ -0,0 +1,59 @@
+[Module README](../README.md) · [Exchanges →](02-exchanges.md)
+
+# 1. The on-ramp
+
+Two dependency decisions decide whether anything works, and neither produces a helpful error.
+
+## Use the starter, not `spring-rabbit`
+
+Boot 4 moved every auto-configuration out of `spring-boot-autoconfigure` into a per-technology
+module. RabbitMQ's lives in `spring-boot-amqp`, package
+`org.springframework.boot.amqp.autoconfigure`, and a bare `org.springframework.amqp:spring-rabbit`
+dependency does not bring it. You get no `RabbitTemplate`, no `RabbitAdmin`, no listener
+container factory — and a context that starts cleanly.
+
+```xml
+<dependency>
+  <groupId>org.springframework.boot</groupId>
+  <artifactId>spring-boot-starter-amqp</artifactId>
+</dependency>
+```
+
+## Boot does not give you a JSON converter
+
+Spring Kafka defaults to serializers you configure. Spring AMQP defaults to
+`SimpleMessageConverter`, which handles `String`, `byte[]` and `Serializable` and refuses
+everything else:
+
+```
+IllegalArgumentException: SimpleMessageConverter only supports String, byte[] and
+Serializable payloads, received: com.ankurm.rabbit.OrderMessage
+```
+
+One bean fixes both directions, because the auto-configured `RabbitTemplate` and the listener
+container factory both look for a `MessageConverter`:
+
+```java
+@Bean
+MessageConverter messageConverter() {
+    return new JacksonJsonMessageConverter();
+}
+```
+
+**Note the class name.** Spring AMQP 4.1 ships `Jackson2JsonMessageConverter` (Jackson 2) and
+`JacksonJsonMessageConverter` (Jackson 3) side by side — exactly as Spring Kafka ships
+`JsonSerializer` and `JacksonJsonSerializer`. Boot 4 is a Jackson 3 application. The rule across
+both stacks is the same: **if the class name contains a `2`, it belongs to the previous major
+version of Jackson.**
+
+## Versions
+
+Boot 4.1.1 manages Spring AMQP **4.1.1** and `com.rabbitmq:amqp-client` **5.30.0**. Maven Central
+has amqp-client 5.35.0; overriding the managed version to reach it is a change you should have a
+reason for.
+
+One API change to know about: `RabbitAdmin.QUEUE_MESSAGE_COUNT` now holds a `Long`. Every older
+example casts it to `Integer`, which is a `ClassCastException` at runtime and not at compile
+time.
+
+[Exchanges &rarr;](02-exchanges.md)

rabbitmq/docs/02-exchanges.md

@@ -0,0 +1,71 @@
+[← The on-ramp](01-the-on-ramp.md) · [Module README](../README.md) · [The silent drop →](03-the-silent-drop.md)
+
+# 2. Four exchange types
+
+A producer never publishes to a queue. It publishes to an **exchange** with a **routing key**,
+and bindings decide where that lands. The whole topology is in
+[`Topology.java`](../src/main/java/com/ankurm/rabbit/Topology.java) as beans; `RabbitAdmin`
+declares them when the connection opens.
+
+## Direct — exact match
+
+Binding key `new` receives routing key `new`. Nothing else. This is the workhorse: one queue per
+command type.
+
+## Fanout — routing key ignored entirely
+
+Every bound queue gets a copy. `audit.all` and `analytics.all` both receive it, and the routing
+key you passed is not consulted at all. Use it for broadcast; use it knowing that adding a queue
+adds a full copy of the traffic.
+
+## Topic — wildcards over dot-separated words
+
+`*` is **exactly one word**. `#` is **zero or more**. The distinction is the one people get
+wrong, so here it is against a real broker
+([`docs/output/topic-wildcards.txt`](output/topic-wildcards.txt)):
+
+```
+routing key              orders.eu    orders.high  note
+order.eu.high            true         true         matches both
+order.eu.low             true         false        matches order.eu.* only
+order.us.high            false        true         matches order.#.high only
+order.eu.west.high       false        true         matches order.#.high only - * is one word
+order.high               false        true         matches order.#.high - # can be zero words
+```
+
+Bindings are `order.eu.*` and `order.#.high`. Two rows are worth pausing on:
+
+- `order.eu.west.high` does **not** match `order.eu.*`, because `*` matches one word and `west.high`
+  is two. Regex intuition says otherwise.
+- `order.high` **does** match `order.#.high`, because `#` matches zero words. So a binding you
+  wrote to mean "something in the middle" also matches "nothing in the middle".
+
+Design routing keys most-general-to-most-specific (`order.eu.west.high`, not
+`high.order.eu.west`), because `#` and `*` work left to right and a hierarchy you can bind
+usefully is one that starts broad.
+
+## Headers — match a map, ignore the routing key
+
+`x-match=all` requires every named header to be present **and equal**. `x-match=any` requires
+one. From `RoutingTest`:
+
+| Message headers | `x-match=all` queue | `x-match=any` queue |
+|---|---|---|
+| `priority=high, region=eu` | yes | yes |
+| `priority=high` | no | yes |
+| `priority=high, region=us` | no | yes |
+
+The third row is the one to remember: `all` matches on **value**, not on presence. A header that
+is there with the wrong value fails the same way a missing one does.
+
+Headers exchanges are slower than topic exchanges and much less common. Reach for them when the
+routing criteria are genuinely multi-dimensional and do not compose into a hierarchy.
+
+## The default exchange
+
+Publishing to the empty exchange name `""` routes by **queue name**, using the routing key as the
+queue name. Every queue is implicitly bound to it. That is how
+`convertAndSend("", "orders.work", message)` works, and it is the one piece of AMQP that behaves
+like a magic constant.
+
+[The silent drop →](03-the-silent-drop.md)

rabbitmq/docs/03-the-silent-drop.md

@@ -0,0 +1,78 @@
+[← Exchanges](02-exchanges.md) · [Module README](../README.md) · [Dead-lettering →](04-dead-lettering.md)
+
+# 3. The message that goes nowhere and says nothing
+
+Publish to `orders.direct` with routing key `amend`. No binding matches. The broker discards the
+message. `convertAndSend` returns normally. Nothing is logged, by the broker or by Spring, and
+the message is gone.
+
+This is correct AMQP behaviour and it is the single most expensive default in RabbitMQ, because
+the symptom is "the consumer isn't running" and the cause is three services away.
+
+**Two settings are required, and they live in different places:**
+
+```yaml
+spring:
+  rabbitmq:
+    publisher-returns: true      # on the connection factory
+    template:
+      mandatory: true            # on the template
+```
+
+`publisher-returns` puts the connection into a mode where returns are listened for at all.
+`mandatory` asks the broker to return *this* publish if it is unroutable. Setting only
+`mandatory` — which is what most examples show — changes nothing, and the test in this module
+was written that way first and failed.
+
+Then register a callback:
+
+```java
+template.setReturnsCallback((returned) -> log.error("unroutable: exchange={} key={} {} {}",
+        returned.getExchange(), returned.getRoutingKey(),
+        returned.getReplyCode(), returned.getReplyText()));
+```
+
+From [`docs/output/unroutable.txt`](output/unroutable.txt):
+
+```
+replyCode    312
+replyText    NO_ROUTE
+exchange     orders.direct
+routingKey   amend
+```
+
+`312 NO_ROUTE` is the code worth putting in an alert.
+
+## What returns still do not give you
+
+The callback fires **asynchronously, on the connection's thread**. `convertAndSend` had already
+returned successfully by then. So a return tells you a message was lost; it does not let you
+stop the caller from believing it succeeded.
+
+If the caller must know, you need **publisher confirms** as well:
+
+```yaml
+spring:
+  rabbitmq:
+    publisher-confirm-type: correlated
+```
+
+Confirms and returns answer different questions, and you usually want both:
+
+| | answers |
+|---|---|
+| **return** | the broker had nowhere to route this |
+| **confirm** | the broker took responsibility for this (`ack`), or refused it (`nack`) |
+
+A message that is unroutable gets a **return followed by an `ack`** — the broker successfully did
+nothing with it. So a confirm alone will not tell you the message vanished, which is the trap
+inside the trap.
+
+## The alternate exchange
+
+The topology-level version of the same fix: give an exchange an `alternate-exchange` argument and
+unroutable messages go there instead of being dropped, with no publisher-side configuration at
+all. It costs one exchange and one queue, and it catches the publishes made by services that
+forgot to set `mandatory`.
+
+[Dead-lettering →](04-dead-lettering.md)

rabbitmq/docs/04-dead-lettering.md

@@ -0,0 +1,97 @@
+[← The silent drop](03-the-silent-drop.md) · [Module README](../README.md) · [Acknowledgement →](05-acknowledgement.md)
+
+# 4. A dead-letter queue that actually works
+
+There is no "send to DLQ" operation in AMQP. A dead-letter exchange is a **queue argument**, and
+messages arrive there as a side effect of three specific events.
+
+```java
+QueueBuilder.durable("orders.work")
+        .deadLetterExchange("orders.dlx")
+        .deadLetterRoutingKey("failed")
+        .build();
+```
+
+`deadLetterRoutingKey` matters more than it looks. Without it, the message is republished with
+its **original** routing key, so your DLQ binding has to anticipate every routing key the
+upstream might have used. Setting it to a constant means one binding catches everything.
+
+## The three triggers
+
+| Event | `x-death` reason |
+|---|---|
+| `basicNack` / `basicReject` with `requeue=false` | `rejected` |
+| message TTL expires (`x-message-ttl` or per-message) | `expired` |
+| queue length or byte limit exceeded (`x-max-length`) | `maxlen` |
+
+And the one that is **not** a trigger: `requeue=true`. More on that below.
+
+## What arrives
+
+Both captured from real runs, in
+[`docs/output/dead-letter.txt`](output/dead-letter.txt):
+
+```
+=== x-death after basicNack(requeue=false) ===
+  reason           rejected
+  count            1
+  exchange
+  time             Sat Aug 29 09:53:17 IST 2026
+  routing-keys     [orders.work]
+  queue            orders.work
+
+=== x-death after x-message-ttl expiry ===
+  reason           expired
+  count            1
+  queue            orders.ttl
+```
+
+Same destination queue, different `reason`. That field is the whole diagnostic: `rejected` means
+a consumer looked at the work and refused it; `expired` means nobody got to it in time. Those are
+different incidents with different fixes, and they are indistinguishable without reading the
+header.
+
+`x-death` is a **list**, not a map — one entry per queue the message has been dead-lettered from,
+and `count` accumulates. That is how you build a retry limit: read
+`x-death[0].count`, and stop republishing past a threshold.
+
+## The infinite loop
+
+```java
+channel.basicNack(deliveryTag, false, true);   // requeue = true
+```
+
+From [`docs/output/requeue-loop.txt`](output/requeue-loop.txt), 200 delivery attempts:
+
+```
+redelivered      199 times
+dead-lettered    0
+still on queue   1
+```
+
+The dead-letter exchange is never consulted. `x-death` is never written. There is no counter to
+exhaust and no backoff. One poison message with an unconditional `requeue=true` handler will
+saturate a consumer indefinitely, and the queue depth stays at 1 the whole time — so a depth
+alarm will not fire either. The only visible signal is CPU.
+
+**`requeue=true` is only correct when the failure is transient and you have a delay.** With no
+delay, a message that fails because a database is down comes straight back while the database is
+still down. The `redelivered` flag on the envelope is your one piece of state:
+
+```java
+boolean retried = response.getEnvelope().isRedeliver();
+channel.basicNack(tag, false, !retried);   // one requeue, then dead-letter
+```
+
+That is a one-attempt retry with no extra infrastructure. For anything more, use a delay queue:
+a queue with a TTL and a DLX pointing back at the work exchange turns "expire after 30 seconds"
+into "retry in 30 seconds".
+
+## Do not point the DLQ's DLX at its own source
+
+If `orders.dlq` dead-letters to an exchange bound back to `orders.work`, a message that fails
+consistently cycles between the two forever, gaining an `x-death` entry each time until the
+header itself is the biggest thing in the message. The DLQ in this module deliberately has no
+dead-letter exchange at all.
+
+[Acknowledgement →](05-acknowledgement.md)

rabbitmq/docs/05-acknowledgement.md

@@ -0,0 +1,82 @@
+[← Dead-lettering](04-dead-lettering.md) · [Module README](../README.md) · [Changing your mind →](06-changing-your-mind.md)
+
+# 5. Acknowledgement
+
+```yaml
+spring:
+  rabbitmq:
+    listener:
+      simple:
+        acknowledge-mode: manual
+        prefetch: 1
+```
+
+Three modes, and the default is not the one you want for work that matters.
+
+| Mode | Behaviour |
+|---|---|
+| `auto` (default) | the container acks after the listener returns, nacks on exception |
+| `manual` | your code calls `basicAck` / `basicNack`; nothing else does |
+| `none` | the broker forgets the message the moment it is delivered |
+
+(Boot publishes no default for `acknowledge-mode` either; the field initialiser in
+`AbstractMessageListenerContainer` is `AcknowledgeMode.AUTO`.)
+
+`none` is genuinely fire-and-forget: no redelivery, ever, and no flow control either, because
+`prefetch` is meaningless without acknowledgement. It is the right choice for metrics and the
+wrong choice for everything else.
+
+`auto` is a reasonable default and its name is misleading: it does not mean "auto-ack" in the
+AMQP sense (that is `none`). It means the container decides, and it decides by whether your
+listener threw.
+
+`manual` is what you want when the acknowledgement must happen after some other durable side
+effect — a database commit, a downstream call. In `manual` mode the listener takes a `Channel`
+and a delivery tag:
+
+```java
+@RabbitListener(queues = "orders.work")
+void onOrder(OrderMessage message, Channel channel,
+        @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
+    try {
+        process(message);
+        channel.basicAck(tag, false);
+    }
+    catch (TransientFailure ex) {
+        channel.basicNack(tag, false, !alreadyRedelivered);
+    }
+    catch (PermanentFailure ex) {
+        channel.basicNack(tag, false, false);   // dead-letter it
+    }
+}
+```
+
+The two catch blocks are the point. A single `catch (Exception)` that nacks with `requeue=true`
+is the loop from [chapter 4](04-dead-lettering.md); a single one that nacks with `requeue=false`
+dead-letters transient failures that would have succeeded on a retry. Deciding which kind of
+failure you had is work you cannot delegate to the container.
+
+## `prefetch` is the flow-control knob
+
+`prefetch` is how many un-acknowledged messages the broker will send a consumer. Spring Boot sets
+no default, so `AbstractMessageListenerContainer.DEFAULT_PREFETCH_COUNT` applies, and that
+constant is **250** — throughput-oriented: a slow consumer holds 250 messages that no
+other consumer can take, so one stuck worker parks 250 messages behind it.
+
+Set it to 1 when messages are expensive and unevenly sized — it gives you round-robin
+distribution across consumers. Leave it high when messages are small and uniform and you want
+throughput. This is the single most effective RabbitMQ tuning parameter and it is almost never
+touched.
+
+## The forgotten ack
+
+Not calling `basicAck` does not lose the message and does not throw. The message stays
+un-acknowledged, counting against `prefetch`, until the channel closes. With `prefetch: 1` the
+consumer stops after exactly one message; with the default it stops after 250. Both look like the
+consumer "just stopped", hours apart in wall-clock time depending on traffic.
+
+`rabbitmqctl list_queues name messages_ready messages_unacknowledged` is the command that
+distinguishes them. A large `messages_unacknowledged` with a live consumer means somebody forgot
+an ack.
+
+[Changing your mind →](06-changing-your-mind.md)

rabbitmq/docs/06-changing-your-mind.md

@@ -0,0 +1,58 @@
+[← Acknowledgement](05-acknowledgement.md) · [Module README](../README.md)
+
+# 6. Changing your mind about a queue
+
+Queue arguments are immutable. There is no `ALTER QUEUE`. Redeclaring an existing queue with
+different arguments is a channel-level error
+([`docs/output/precondition-failed.txt`](output/precondition-failed.txt)):
+
+```
+PRECONDITION_FAILED - inequivalent arg 'x-message-ttl' for queue 'orders.ttl' in vhost '/':
+received '9999' but current is '1500', class-id=50, method-id=10
+```
+
+Reply code **406**. The message is precise and helpful, and you will probably never see it,
+because of what Spring wraps it in:
+
+```
+AmqpIOException: java.io.IOException
+  caused by: null
+    caused by: channel error; protocol method: #method<channel.close>(reply-code=406, ...)
+```
+
+`AmqpIOException`'s own message is the string `java.io.IOException`. The next cause's message is
+`null`. The useful text is two levels down. This is why the failure gets reported as "some
+IOException from RabbitMQ" and why it is worth writing a log statement that walks the cause
+chain.
+
+## What this means operationally
+
+Changing a TTL, a max-length, a dead-letter exchange or a queue type on a live queue is not a
+config edit. It is a migration:
+
+1. declare the new queue under a new name
+2. bind it alongside the old one
+3. move consumers over
+4. drain the old queue
+5. delete it
+
+Plan the rename into the change from the start — `orders.work.v2` is not ugly, it is honest.
+
+## Related failure modes
+
+**A failed declaration stops the ones after it.** `RabbitAdmin` declares beans on connection, and
+a `PRECONDITION_FAILED` closes the channel. Declarations queued behind it on that channel do not
+happen. So one stale queue argument can leave half your topology undeclared, and the symptom is a
+*different* queue being missing.
+
+**Declarations happen on connection, not on context refresh.** A broken topology does not fail
+startup; it fails the first publish. If you want it to fail at boot, force a connection early —
+`connectionFactory.createConnection()` in an `ApplicationRunner` is enough.
+
+**`missing-queues-fatal` defaults differently for the two container types.** It is `true` for
+`spring.rabbitmq.listener.simple` and `false` for `spring.rabbitmq.listener.direct`. So the same
+missing queue kills a simple container and is quietly tolerated by a direct one, which makes
+"it works in that service" an unhelpful data point. Combined with the point above, a topology
+error can present as a listener startup problem in one service and as silence in another.
+
+[Module README](../README.md)

rabbitmq/docs/output/dead-letter.txt

@@ -0,0 +1,14 @@
+=== x-death after basicNack(requeue=false) ===
+  reason           rejected
+  count            1
+  exchange         
+  time             Sat Aug 29 09:55:45 IST 2026
+  routing-keys     [orders.work]
+  queue            orders.work
+=== x-death after x-message-ttl expiry ===
+  reason           expired
+  count            1
+  exchange         
+  time             Sat Aug 29 09:55:47 IST 2026
+  routing-keys     [orders.ttl]
+  queue            orders.ttl

rabbitmq/docs/output/precondition-failed.txt

@@ -0,0 +1,2 @@
+=== redeclaring orders.ttl with x-message-ttl=9999 ===
+  java.io.IOException | null | channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - inequivalent arg 'x-message-ttl' for queue 'orders.ttl' in vhost '/': received '9999' but current is '1500', class-id=50, method-id=10) | channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - inequivalent arg 'x-message-ttl' for queue 'orders.ttl' in vhost '/': received '9999' but current is '1500', class-id=50, method-id=10) | 

rabbitmq/docs/output/requeue-loop.txt

@@ -0,0 +1,4 @@
+=== basicNack(requeue=true), 200 attempts ===
+  redelivered      199 times
+  dead-lettered    0
+  still on queue   1

rabbitmq/docs/output/tests.txt

@@ -0,0 +1,3 @@
+Tests run: 2, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 5.178 s -- in com.ankurm.rabbit.TopologyTrapsTest
+Tests run: 3, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 1.797 s -- in com.ankurm.rabbit.DeadLetterTest
+Tests run: 4, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 4.511 s -- in com.ankurm.rabbit.RoutingTest

rabbitmq/docs/output/topic-wildcards.txt

@@ -0,0 +1,7 @@
+=== topic exchange ===
+routing key              orders.eu    orders.high  note
+order.eu.high            true         true         matches both
+order.eu.low             true         false        matches order.eu.* only
+order.us.high            false        true         matches order.#.high only
+order.eu.west.high       false        true         matches order.#.high only - * is one word
+order.high               false        true         matches order.#.high - # can be zero words

rabbitmq/docs/output/unroutable.txt

@@ -0,0 +1,5 @@
+=== returned message ===
+  replyCode    312
+  replyText    NO_ROUTE
+  exchange     orders.direct
+  routingKey   amend

rabbitmq/pom.xml

@@ -0,0 +1,67 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0">
+  <modelVersion>4.0.0</modelVersion>
+
+  <parent>
+    <groupId>org.springframework.boot</groupId>
+    <artifactId>spring-boot-starter-parent</artifactId>
+    <version>4.1.1</version>
+    <relativePath/>
+  </parent>
+
+  <groupId>com.ankurm</groupId>
+  <artifactId>rabbitmq-exchanges-and-dlq</artifactId>
+  <version>1.0</version>
+  <packaging>jar</packaging>
+
+  <properties>
+    <java.version>25</java.version>
+    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+  </properties>
+
+  <dependencies>
+    <!-- spring-boot-starter-amqp, not a bare org.springframework.amqp:spring-rabbit. In Boot 4
+         the auto-configuration lives in the spring-boot-amqp module (package
+         org.springframework.boot.amqp.autoconfigure); depending on spring-rabbit alone gives
+         you no RabbitTemplate and no listener container factory. Same trap as Kafka. -->
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-amqp</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-jackson</artifactId>
+    </dependency>
+
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-test</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.springframework.amqp</groupId>
+      <artifactId>spring-rabbit-test</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <!-- NOT org.testcontainers:rabbitmq, which stopped at 1.21.4. Testcontainers 2.x prefixed
+         every module artifact and Boot 4.1.1 imports testcontainers-bom 2.0.5. -->
+    <dependency>
+      <groupId>org.testcontainers</groupId>
+      <artifactId>testcontainers-rabbitmq</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-testcontainers</artifactId>
+      <scope>test</scope>
+    </dependency>
+  </dependencies>
+
+  <build>
+    <plugins>
+      <plugin>
+        <groupId>org.springframework.boot</groupId>
+        <artifactId>spring-boot-maven-plugin</artifactId>
+      </plugin>
+    </plugins>
+  </build>
+</project>

rabbitmq/scripts/broker.sh

@@ -0,0 +1,28 @@
+#!/usr/bin/env bash
+# Start a real RabbitMQ broker with no Docker and no root.
+#
+# RabbitMQ is an Erlang application, so it needs an Erlang runtime and epmd on the path - that
+# is the whole dependency. Point ERL_ROOT at an Erlang installation and RABBITMQ_HOME at an
+# unpacked rabbitmq-server-generic-unix tarball, and it starts in about fifteen seconds.
+#
+# If you have Docker, use Testcontainers instead: see src/test/java/.../TestcontainersConfiguration.java
+set -eu
+: "${ERL_ROOT:?set ERL_ROOT to an Erlang installation directory}"
+: "${RABBITMQ_HOME:?set RABBITMQ_HOME to an unpacked rabbitmq-server-generic-unix directory}"
+
+export PATH="$ERL_ROOT/bin:$ERL_ROOT/erts-"*/bin":$PATH"
+export RABBITMQ_MNESIA_BASE="${RABBITMQ_MNESIA_BASE:-/tmp/rmq/data}"
+export RABBITMQ_LOG_BASE="${RABBITMQ_LOG_BASE:-/tmp/rmq/log}"
+export RABBITMQ_NODENAME="${RABBITMQ_NODENAME:-rabbit@localhost}"
+export HOME="${HOME:-/tmp/rmq}"
+mkdir -p "$RABBITMQ_MNESIA_BASE" "$RABBITMQ_LOG_BASE"
+
+epmd -daemon 2>/dev/null || true
+sleep 1
+setsid nohup "$RABBITMQ_HOME/sbin/rabbitmq-server" > /tmp/rmq/boot.log 2>&1 < /dev/null &
+for _ in $(seq 1 60); do
+  if (echo > /dev/tcp/127.0.0.1/5672) 2>/dev/null; then echo "broker ready on 5672"; exit 0; fi
+  sleep 1
+done
+echo "broker did not start; see /tmp/rmq/boot.log" >&2
+exit 1

rabbitmq/scripts/run-all.sh

@@ -0,0 +1,18 @@
+#!/usr/bin/env bash
+# Regenerate every file under docs/output/.
+#
+# Starts a real RabbitMQ broker first. Set ERL_ROOT and RABBITMQ_HOME (see scripts/broker.sh),
+# or point spring.rabbitmq.host at a broker you already have and skip the first line.
+set -eu
+cd "$(dirname "$0")/.."
+./scripts/broker.sh
+mvn -B test 2>&1 | tee /tmp/rabbit-test.log > /dev/null
+
+sed -n '/=== topic exchange ===/,/^order\.high/p' /tmp/rabbit-test.log > docs/output/topic-wildcards.txt
+sed -n '/=== x-death after basicNack/,/^  queue/p' /tmp/rabbit-test.log > docs/output/dead-letter.txt
+sed -n '/=== x-death after x-message-ttl/,/^  queue/p' /tmp/rabbit-test.log >> docs/output/dead-letter.txt
+sed -n '/=== basicNack(requeue=true)/,/still on queue/p' /tmp/rabbit-test.log > docs/output/requeue-loop.txt
+sed -n '/=== returned message ===/,/routingKey/p' /tmp/rabbit-test.log > docs/output/unroutable.txt
+sed -n '/=== redeclaring orders.ttl/,+1p' /tmp/rabbit-test.log > docs/output/precondition-failed.txt
+grep -E 'Tests run:.*in com\.ankurm' /tmp/rabbit-test.log | sed 's/^\[INFO\] //' > docs/output/tests.txt
+echo "regenerated:"; ls -1 docs/output/

rabbitmq/src/main/java/com/ankurm/rabbit/ConverterConfiguration.java

@@ -0,0 +1,36 @@
+package com.ankurm.rabbit;
+
+import org.springframework.amqp.support.converter.JacksonJsonMessageConverter;
+import org.springframework.amqp.support.converter.MessageConverter;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * Spring Boot does <b>not</b> auto-configure a JSON message converter for RabbitMQ. The default
+ * is {@code SimpleMessageConverter}, which handles {@code String}, {@code byte[]} and
+ * {@code Serializable} and nothing else:
+ *
+ * <pre>
+ * IllegalArgumentException: SimpleMessageConverter only supports String, byte[] and
+ * Serializable payloads, received: com.ankurm.rabbit.OrderMessage
+ * </pre>
+ *
+ * <p>Declaring one {@code MessageConverter} bean fixes both directions &mdash; the auto-configured
+ * {@code RabbitTemplate} and the listener container factory both pick it up.
+ *
+ * <p>Note the class name. Spring AMQP 4.1 ships {@code Jackson2JsonMessageConverter} (Jackson 2)
+ * and {@code JacksonJsonMessageConverter} (Jackson 3) side by side, exactly as Spring Kafka
+ * ships {@code JsonSerializer} and {@code JacksonJsonSerializer}. Boot 4 is a Jackson 3
+ * application; pick the one without the 2.
+ *
+ * @see <a href="../../../../../docs/01-the-on-ramp.md">docs/01-the-on-ramp.md</a>
+ */
+@Configuration(proxyBeanMethods = false)
+public class ConverterConfiguration {
+
+	@Bean
+	MessageConverter messageConverter() {
+		return new JacksonJsonMessageConverter();
+	}
+
+}

rabbitmq/src/main/java/com/ankurm/rabbit/OrderMessage.java

@@ -0,0 +1,13 @@
+package com.ankurm.rabbit;
+
+/**
+ * The payload. Kept to a String field so the tests are about routing and acknowledgement rather
+ * than about converters.
+ */
+public record OrderMessage(String orderId, String detail) {
+
+	public static OrderMessage of(String orderId) {
+		return new OrderMessage(orderId, "detail for " + orderId);
+	}
+
+}

rabbitmq/src/main/java/com/ankurm/rabbit/RabbitDemoApplication.java

@@ -0,0 +1,20 @@
+package com.ankurm.rabbit;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+/**
+ * All four AMQP 0-9-1 exchange types, a working dead-letter path, and the traps that make
+ * RabbitMQ feel unpredictable until you know them.
+ *
+ * <p>Everything here runs against a real broker. See the module README for how the scripts start
+ * one without Docker.
+ */
+@SpringBootApplication
+public class RabbitDemoApplication {
+
+	public static void main(String[] args) {
+		SpringApplication.run(RabbitDemoApplication.class, args);
+	}
+
+}

rabbitmq/src/main/java/com/ankurm/rabbit/Topology.java

@@ -0,0 +1,217 @@
+package com.ankurm.rabbit;
+
+import org.springframework.amqp.core.Binding;
+import org.springframework.amqp.core.BindingBuilder;
+import org.springframework.amqp.core.DirectExchange;
+import org.springframework.amqp.core.FanoutExchange;
+import org.springframework.amqp.core.HeadersExchange;
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.core.QueueBuilder;
+import org.springframework.amqp.core.TopicExchange;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import java.util.Map;
+
+/**
+ * The whole topology as beans. {@code RabbitAdmin} declares every {@code Exchange},
+ * {@code Queue} and {@code Binding} bean when the connection is first opened.
+ *
+ * <p>Two things about that are worth knowing before you rely on it:
+ * <ul>
+ *   <li>Declaration is <b>idempotent only if the arguments match</b>. Redeclaring an existing
+ *       queue with different arguments is a channel-level {@code PRECONDITION_FAILED} &mdash; see
+ *       docs/06-changing-your-mind.md.</li>
+ *   <li>Declarations happen on connection, not on context refresh, so a topology error surfaces
+ *       at the first publish rather than at startup.</li>
+ * </ul>
+ *
+ * @see <a href="../../../../../docs/02-exchanges.md">docs/02-exchanges.md</a>
+ */
+@Configuration(proxyBeanMethods = false)
+public class Topology {
+
+	public static final String DIRECT = "orders.direct";
+
+	public static final String FANOUT = "orders.fanout";
+
+	public static final String TOPIC = "orders.topic";
+
+	public static final String HEADERS = "orders.headers";
+
+	public static final String DLX = "orders.dlx";
+
+	public static final String Q_NEW = "orders.new";
+
+	public static final String Q_CANCEL = "orders.cancel";
+
+	public static final String Q_AUDIT = "audit.all";
+
+	public static final String Q_ANALYTICS = "analytics.all";
+
+	public static final String Q_EU = "orders.eu";
+
+	public static final String Q_HIGH = "orders.high";
+
+	public static final String Q_PRIORITY = "orders.priority";
+
+	public static final String Q_ANY = "orders.any";
+
+	public static final String Q_WORK = "orders.work";
+
+	public static final String Q_TTL = "orders.ttl";
+
+	public static final String Q_DLQ = "orders.dlq";
+
+	// --- direct: routing key must match the binding key exactly -----------------------------
+
+	@Bean
+	DirectExchange directExchange() {
+		return new DirectExchange(DIRECT);
+	}
+
+	@Bean
+	Queue newQueue() {
+		return QueueBuilder.durable(Q_NEW).build();
+	}
+
+	@Bean
+	Queue cancelQueue() {
+		return QueueBuilder.durable(Q_CANCEL).build();
+	}
+
+	@Bean
+	Binding bindNew(DirectExchange directExchange, Queue newQueue) {
+		return BindingBuilder.bind(newQueue).to(directExchange).with("new");
+	}
+
+	@Bean
+	Binding bindCancel(DirectExchange directExchange, Queue cancelQueue) {
+		return BindingBuilder.bind(cancelQueue).to(directExchange).with("cancel");
+	}
+
+	// --- fanout: routing key ignored entirely -----------------------------------------------
+
+	@Bean
+	FanoutExchange fanoutExchange() {
+		return new FanoutExchange(FANOUT);
+	}
+
+	@Bean
+	Queue auditQueue() {
+		return QueueBuilder.durable(Q_AUDIT).build();
+	}
+
+	@Bean
+	Queue analyticsQueue() {
+		return QueueBuilder.durable(Q_ANALYTICS).build();
+	}
+
+	@Bean
+	Binding bindAudit(FanoutExchange fanoutExchange, Queue auditQueue) {
+		return BindingBuilder.bind(auditQueue).to(fanoutExchange);
+	}
+
+	@Bean
+	Binding bindAnalytics(FanoutExchange fanoutExchange, Queue analyticsQueue) {
+		return BindingBuilder.bind(analyticsQueue).to(fanoutExchange);
+	}
+
+	// --- topic: '*' is exactly one word, '#' is zero or more -------------------------------
+
+	@Bean
+	TopicExchange topicExchange() {
+		return new TopicExchange(TOPIC);
+	}
+
+	@Bean
+	Queue euQueue() {
+		return QueueBuilder.durable(Q_EU).build();
+	}
+
+	@Bean
+	Queue highQueue() {
+		return QueueBuilder.durable(Q_HIGH).build();
+	}
+
+	@Bean
+	Binding bindEu(TopicExchange topicExchange, Queue euQueue) {
+		return BindingBuilder.bind(euQueue).to(topicExchange).with("order.eu.*");
+	}
+
+	@Bean
+	Binding bindHigh(TopicExchange topicExchange, Queue highQueue) {
+		return BindingBuilder.bind(highQueue).to(topicExchange).with("order.#.high");
+	}
+
+	// --- headers: routing key ignored, header map matched ------------------------------------
+
+	@Bean
+	HeadersExchange headersExchange() {
+		return new HeadersExchange(HEADERS);
+	}
+
+	@Bean
+	Queue priorityQueue() {
+		return QueueBuilder.durable(Q_PRIORITY).build();
+	}
+
+	@Bean
+	Queue anyQueue() {
+		return QueueBuilder.durable(Q_ANY).build();
+	}
+
+	/** x-match=all: every header must match. */
+	@Bean
+	Binding bindPriority(HeadersExchange headersExchange, Queue priorityQueue) {
+		return BindingBuilder.bind(priorityQueue).to(headersExchange)
+				.whereAll(Map.of("priority", "high", "region", "eu")).match();
+	}
+
+	/** x-match=any: one is enough. */
+	@Bean
+	Binding bindAny(HeadersExchange headersExchange, Queue anyQueue) {
+		return BindingBuilder.bind(anyQueue).to(headersExchange)
+				.whereAny(Map.of("priority", "high", "region", "eu")).match();
+	}
+
+	// --- the dead-letter path ----------------------------------------------------------------
+
+	@Bean
+	DirectExchange deadLetterExchange() {
+		return new DirectExchange(DLX);
+	}
+
+	@Bean
+	Queue deadLetterQueue() {
+		// Deliberately NO dead-letter-exchange on the DLQ itself. Pointing a DLQ's DLX back at
+		// the exchange that feeds it is the classic infinite loop.
+		return QueueBuilder.durable(Q_DLQ).build();
+	}
+
+	@Bean
+	Binding bindDlq(DirectExchange deadLetterExchange, Queue deadLetterQueue) {
+		// The routing key used when dead-lettering is the message's ORIGINAL routing key unless
+		// deadLetterRoutingKey() overrides it. Both work queues below set it explicitly to
+		// "failed", so this one binding catches everything.
+		return BindingBuilder.bind(deadLetterQueue).to(deadLetterExchange).with("failed");
+	}
+
+	@Bean
+	Queue workQueue() {
+		return QueueBuilder.durable(Q_WORK)
+				.deadLetterExchange(DLX)
+				.deadLetterRoutingKey("failed")
+				.build();
+	}
+
+	@Bean
+	Queue ttlQueue() {
+		return QueueBuilder.durable(Q_TTL)
+				.ttl(1500)
+				.deadLetterExchange(DLX)
+				.deadLetterRoutingKey("failed")
+				.build();
+	}
+
+}

rabbitmq/src/main/resources/application.yaml

@@ -0,0 +1,31 @@
+spring:
+  application:
+    name: rabbitmq-exchanges-and-dlq
+  main:
+    banner-mode: off
+  rabbitmq:
+    host: localhost
+    port: 5672
+    username: guest
+    password: guest
+    # BOTH of these are required to find out about an unroutable message, and they are set in
+    # two different places. publisher-returns puts the connection factory into a mode where it
+    # listens for returns at all; mandatory asks the broker to return this particular publish.
+    # Setting only mandatory - which is what most examples show - changes nothing at all.
+    # See docs/03-the-silent-drop.md.
+    publisher-returns: true
+    template:
+      mandatory: true
+    listener:
+      simple:
+        # MANUAL means nothing is removed from the queue until your code says so - and that
+        # basicNack(requeue=false) is what feeds the dead-letter exchange.
+        acknowledge-mode: manual
+        prefetch: 1
+        # Listener containers are not started automatically here; each test starts the one it
+        # needs, so that a redelivery loop in one test cannot consume another test's messages.
+        auto-startup: false
+logging:
+  level:
+    root: WARN
+    com.ankurm: INFO

rabbitmq/src/test/java/com/ankurm/rabbit/DeadLetterTest.java

@@ -0,0 +1,143 @@
+package com.ankurm.rabbit;
+
+import com.rabbitmq.client.GetResponse;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.amqp.rabbit.core.RabbitAdmin;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * The dead-letter path, driven with the AMQP primitives directly rather than through a listener
+ * container. {@code basicGet} + {@code basicNack} is what a container does underneath, and doing
+ * it by hand removes all the timing from the test.
+ *
+ * @see <a href="../../../../../docs/04-dead-lettering.md">docs/04-dead-lettering.md</a>
+ */
+@SpringBootTest
+class DeadLetterTest {
+
+	@Autowired
+	RabbitTemplate template;
+
+	@Autowired
+	RabbitAdmin admin;
+
+	@BeforeEach
+	void drain() {
+		List.of(Topology.Q_WORK, Topology.Q_TTL, Topology.Q_DLQ)
+				.forEach((q) -> this.admin.purgeQueue(q, false));
+	}
+
+	private int depth(String queue) {
+		Properties properties = this.admin.getQueueProperties(queue);
+		return (properties == null) ? -1
+				: ((Number) properties.get(RabbitAdmin.QUEUE_MESSAGE_COUNT)).intValue();
+	}
+
+	private int settled(String queue, int expected) throws InterruptedException {
+		for (int i = 0; i < 150; i++) {
+			if (depth(queue) == expected) {
+				return expected;
+			}
+			Thread.sleep(20);
+		}
+		return depth(queue);
+	}
+
+	@Test
+	void nackWithoutRequeueDeadLettersAndStampsTheReason() throws Exception {
+		this.template.convertAndSend(Topology.DIRECT, "ignored", OrderMessage.of("o-1"),
+				(m) -> m, null);
+		// Publish straight to the work queue via the default exchange: the empty exchange name
+		// routes by queue name, which is the one piece of AMQP that behaves like a magic
+		// constant and is worth knowing.
+		this.template.convertAndSend("", Topology.Q_WORK, OrderMessage.of("o-1"));
+		assertThat(settled(Topology.Q_WORK, 1)).isEqualTo(1);
+
+		this.template.execute((channel) -> {
+			GetResponse response = channel.basicGet(Topology.Q_WORK, false);
+			assertThat(response).isNotNull();
+			// requeue=false is the entire dead-letter trigger. There is no separate "send to
+			// DLQ" call in AMQP.
+			channel.basicNack(response.getEnvelope().getDeliveryTag(), false, false);
+			return null;
+		});
+
+		assertThat(settled(Topology.Q_DLQ, 1)).isEqualTo(1);
+		assertThat(depth(Topology.Q_WORK)).isZero();
+
+		Map<String, Object> death = firstDeath(Topology.Q_DLQ);
+		System.out.println("=== x-death after basicNack(requeue=false) ===");
+		death.forEach((k, v) -> System.out.printf("  %-16s %s%n", k, v));
+		assertThat(death).containsEntry("reason", "rejected");
+		assertThat(death).containsEntry("queue", Topology.Q_WORK);
+		assertThat(((Number) death.get("count")).intValue()).isEqualTo(1);
+	}
+
+	@Test
+	void messageTtlExpiryAlsoDeadLettersButWithADifferentReason() throws Exception {
+		this.template.convertAndSend("", Topology.Q_TTL, OrderMessage.of("o-ttl"));
+		// orders.ttl carries x-message-ttl=1500. Nobody consumes it; the broker expires it.
+		assertThat(settled(Topology.Q_DLQ, 1)).isEqualTo(1);
+
+		Map<String, Object> death = firstDeath(Topology.Q_DLQ);
+		System.out.println("=== x-death after x-message-ttl expiry ===");
+		death.forEach((k, v) -> System.out.printf("  %-16s %s%n", k, v));
+		// Same destination, different reason. This is the field that tells an operator whether
+		// the consumer rejected the work or never got to it.
+		assertThat(death).containsEntry("reason", "expired");
+		assertThat(death).containsEntry("queue", Topology.Q_TTL);
+	}
+
+	@Test
+	void nackWithRequeueIsAnInfiniteLoopAndNeverDeadLetters() throws Exception {
+		this.template.convertAndSend("", Topology.Q_WORK, OrderMessage.of("o-loop"));
+		assertThat(settled(Topology.Q_WORK, 1)).isEqualTo(1);
+
+		int redeliveries = this.template.execute((channel) -> {
+			int count = 0;
+			for (int i = 0; i < 200; i++) {
+				GetResponse response = channel.basicGet(Topology.Q_WORK, false);
+				if (response == null) {
+					break;
+				}
+				if (response.getEnvelope().isRedeliver()) {
+					count++;
+				}
+				// requeue=true puts it back. The dead-letter exchange is never consulted,
+				// x-death is never written, and the loop has no counter to exhaust.
+				channel.basicNack(response.getEnvelope().getDeliveryTag(), false, true);
+			}
+			return count;
+		});
+
+		System.out.println("=== basicNack(requeue=true), 200 attempts ===");
+		System.out.println("  redelivered      " + redeliveries + " times");
+		System.out.println("  dead-lettered    " + depth(Topology.Q_DLQ));
+		System.out.println("  still on queue   " + depth(Topology.Q_WORK));
+
+		assertThat(redeliveries).isGreaterThan(150);
+		// The message is exactly where it started, having been processed 200 times.
+		assertThat(depth(Topology.Q_DLQ)).isZero();
+		assertThat(settled(Topology.Q_WORK, 1)).isEqualTo(1);
+	}
+
+	@SuppressWarnings("unchecked")
+	private Map<String, Object> firstDeath(String queue) {
+		var message = this.template.receive(queue, 5000);
+		assertThat(message).isNotNull();
+		List<Map<String, Object>> deaths =
+				(List<Map<String, Object>>) message.getMessageProperties().getHeader("x-death");
+		assertThat(deaths).isNotNull().isNotEmpty();
+		return deaths.get(0);
+	}
+
+}

rabbitmq/src/test/java/com/ankurm/rabbit/RoutingTest.java

@@ -0,0 +1,152 @@
+package com.ankurm.rabbit;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.rabbit.core.RabbitAdmin;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * All four exchange types, driven with real publishes and counted by draining the queues.
+ *
+ * @see <a href="../../../../../docs/02-exchanges.md">docs/02-exchanges.md</a>
+ */
+@SpringBootTest
+class RoutingTest {
+
+	@Autowired
+	RabbitTemplate template;
+
+	@Autowired
+	RabbitAdmin admin;
+
+	private static final List<String> QUEUES = List.of(Topology.Q_NEW, Topology.Q_CANCEL,
+			Topology.Q_AUDIT, Topology.Q_ANALYTICS, Topology.Q_EU, Topology.Q_HIGH,
+			Topology.Q_PRIORITY, Topology.Q_ANY);
+
+	@BeforeEach
+	void drain() {
+		QUEUES.forEach((q) -> this.admin.purgeQueue(q, false));
+	}
+
+	/**
+	 * Ready message count, straight from the broker.
+	 *
+	 * <p>Note the {@code Number}: {@code RabbitAdmin.QUEUE_MESSAGE_COUNT} holds a {@code Long}
+	 * in Spring AMQP 4.1. Casting it to {@code Integer}, as every older example does, is a
+	 * {@code ClassCastException} at runtime.
+	 */
+	private int depth(String queue) {
+		Properties properties = this.admin.getQueueProperties(queue);
+		return (properties == null) ? -1
+				: ((Number) properties.get(RabbitAdmin.QUEUE_MESSAGE_COUNT)).intValue();
+	}
+
+	/** Publishing is asynchronous; give the broker a moment to route before counting. */
+	private int settledDepth(String queue) {
+		int last = -1;
+		for (int i = 0; i < 50; i++) {
+			last = depth(queue);
+			if (last > 0) {
+				return last;
+			}
+			try {
+				Thread.sleep(20);
+			}
+			catch (InterruptedException ex) {
+				Thread.currentThread().interrupt();
+			}
+		}
+		return last;
+	}
+
+	@Test
+	void directExchangeMatchesTheRoutingKeyExactly() {
+		this.template.convertAndSend(Topology.DIRECT, "new", OrderMessage.of("o-1"));
+		this.template.convertAndSend(Topology.DIRECT, "cancel", OrderMessage.of("o-2"));
+		// No binding for "amend". The broker discards it - see UnroutableTest.
+		this.template.convertAndSend(Topology.DIRECT, "amend", OrderMessage.of("o-3"));
+
+		assertThat(settledDepth(Topology.Q_NEW)).isEqualTo(1);
+		assertThat(settledDepth(Topology.Q_CANCEL)).isEqualTo(1);
+	}
+
+	@Test
+	void fanoutIgnoresTheRoutingKeyCompletely() {
+		this.template.convertAndSend(Topology.FANOUT, "this-is-ignored", OrderMessage.of("o-1"));
+
+		assertThat(settledDepth(Topology.Q_AUDIT)).isEqualTo(1);
+		assertThat(settledDepth(Topology.Q_ANALYTICS)).isEqualTo(1);
+	}
+
+	@Test
+	void topicWildcardsAreWordsNotCharacters() {
+		// binding "order.eu.*"   - exactly one word after order.eu
+		// binding "order.#.high" - zero or more words between order. and .high
+		Map<String, String> routingKeys = new LinkedHashMap<>();
+		routingKeys.put("order.eu.high", "matches both");
+		routingKeys.put("order.eu.low", "matches order.eu.* only");
+		routingKeys.put("order.us.high", "matches order.#.high only");
+		routingKeys.put("order.eu.west.high", "matches order.#.high only - * is one word");
+		routingKeys.put("order.high", "matches order.#.high - # can be zero words");
+
+		System.out.println("=== topic exchange ===");
+		System.out.printf("%-24s %-12s %-12s %s%n", "routing key", "orders.eu", "orders.high", "note");
+		for (Map.Entry<String, String> entry : routingKeys.entrySet()) {
+			this.admin.purgeQueue(Topology.Q_EU, false);
+			this.admin.purgeQueue(Topology.Q_HIGH, false);
+			this.template.convertAndSend(Topology.TOPIC, entry.getKey(), OrderMessage.of("o"));
+			System.out.printf("%-24s %-12s %-12s %s%n", entry.getKey(), settledDepth(Topology.Q_EU) == 1,
+					settledDepth(Topology.Q_HIGH) == 1, entry.getValue());
+		}
+
+		this.admin.purgeQueue(Topology.Q_EU, false);
+		this.admin.purgeQueue(Topology.Q_HIGH, false);
+		// The one everybody gets wrong: '*' is one WORD, so it does not match two.
+		this.template.convertAndSend(Topology.TOPIC, "order.eu.west.high", OrderMessage.of("o"));
+		assertThat(depth(Topology.Q_EU)).isEqualTo(0);
+		assertThat(settledDepth(Topology.Q_HIGH)).isEqualTo(1);
+
+		this.admin.purgeQueue(Topology.Q_HIGH, false);
+		// And '#' really does match zero words.
+		this.template.convertAndSend(Topology.TOPIC, "order.high", OrderMessage.of("o"));
+		assertThat(settledDepth(Topology.Q_HIGH)).isEqualTo(1);
+	}
+
+	@Test
+	void headersExchangeMatchesTheHeaderMapNotTheRoutingKey() {
+		send(Map.of("priority", "high", "region", "eu"));
+		assertThat(settledDepth(Topology.Q_PRIORITY)).isEqualTo(1);   // x-match=all: both present
+		assertThat(settledDepth(Topology.Q_ANY)).isEqualTo(1);        // x-match=any: either is enough
+
+		drain();
+		send(Map.of("priority", "high"));
+		assertThat(depth(Topology.Q_PRIORITY)).isEqualTo(0);   // all: region missing
+		assertThat(settledDepth(Topology.Q_ANY)).isEqualTo(1);
+
+		drain();
+		send(Map.of("priority", "high", "region", "us"));
+		// x-match=all needs every header to match by VALUE, not merely to be present.
+		assertThat(depth(Topology.Q_PRIORITY)).isEqualTo(0);
+		assertThat(settledDepth(Topology.Q_ANY)).isEqualTo(1);
+	}
+
+	private void send(Map<String, Object> headers) {
+		this.template.convertAndSend(Topology.HEADERS, "routing-key-is-ignored",
+				OrderMessage.of("o-1"), (message) -> {
+					MessageProperties properties = message.getMessageProperties();
+					headers.forEach(properties::setHeader);
+					return message;
+				});
+	}
+
+}

rabbitmq/src/test/java/com/ankurm/rabbit/TestcontainersConfiguration.java

@@ -0,0 +1,33 @@
+package com.ankurm.rabbit;
+
+import org.springframework.boot.test.context.TestConfiguration;
+import org.springframework.boot.testcontainers.service.connection.ServiceConnection;
+import org.springframework.context.annotation.Bean;
+import org.testcontainers.rabbitmq.RabbitMQContainer;
+import org.testcontainers.utility.DockerImageName;
+
+/**
+ * The Testcontainers route, for machines that have a Docker daemon.
+ *
+ * <p>{@code @ServiceConnection} supplies host, port, username and password to the
+ * auto-configuration, so no {@code spring.rabbitmq.*} property and no
+ * {@code @DynamicPropertySource} block is needed.
+ *
+ * <p>The Maven coordinate is <b>{@code org.testcontainers:testcontainers-rabbitmq}</b>.
+ * Testcontainers 2.x prefixed every module artifact; the old {@code org.testcontainers:rabbitmq}
+ * stopped at 1.21.4 and is not managed by the Boot 4.1 BOM.
+ *
+ * <p>The committed transcripts under {@code docs/output/} came from a broker started by
+ * {@code scripts/broker.sh} instead, on a machine with no Docker &mdash; see the module README
+ * for why that matters and what version it was.
+ */
+@TestConfiguration(proxyBeanMethods = false)
+public class TestcontainersConfiguration {
+
+	@Bean
+	@ServiceConnection
+	RabbitMQContainer rabbitContainer() {
+		return new RabbitMQContainer(DockerImageName.parse("rabbitmq:4.1-management"));
+	}
+
+}

rabbitmq/src/test/java/com/ankurm/rabbit/TopologyTrapsTest.java

@@ -0,0 +1,100 @@
+package com.ankurm.rabbit;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.amqp.AmqpIOException;
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.core.QueueBuilder;
+import org.springframework.amqp.core.ReturnedMessage;
+import org.springframework.amqp.rabbit.core.RabbitAdmin;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+
+/**
+ * Two ways a RabbitMQ topology loses your work quietly.
+ *
+ * @see <a href="../../../../../docs/03-the-silent-drop.md">docs/03-the-silent-drop.md</a>
+ * @see <a href="../../../../../docs/06-changing-your-mind.md">docs/06-changing-your-mind.md</a>
+ */
+@SpringBootTest
+class TopologyTrapsTest {
+
+	@Autowired
+	RabbitTemplate template;
+
+	@Autowired
+	RabbitAdmin admin;
+
+	@Test
+	void anUnroutableMessageIsDiscardedUnlessYouAskForItBack() throws Exception {
+		List<ReturnedMessage> returned = new CopyOnWriteArrayList<>();
+		this.template.setReturnsCallback(returned::add);
+
+		// "amend" matches no binding on orders.direct. The broker has nowhere to put it.
+		this.template.convertAndSend(Topology.DIRECT, "amend", OrderMessage.of("o-1"));
+		for (int i = 0; i < 100 && returned.isEmpty(); i++) {
+			Thread.sleep(20);
+		}
+
+		// spring.rabbitmq.template.mandatory=true is what turns a silent discard into a return.
+		assertThat(this.template.isMandatoryFor(null)).isTrue();
+		assertThat(returned).hasSize(1);
+		ReturnedMessage message = returned.get(0);
+		System.out.println("=== returned message ===");
+		System.out.println("  replyCode    " + message.getReplyCode());
+		System.out.println("  replyText    " + message.getReplyText());
+		System.out.println("  exchange     " + message.getExchange());
+		System.out.println("  routingKey   " + message.getRoutingKey());
+
+		assertThat(message.getReplyCode()).isEqualTo(312);
+		assertThat(message.getReplyText()).isEqualTo("NO_ROUTE");
+		assertThat(message.getRoutingKey()).isEqualTo("amend");
+		// Note what did NOT happen: convertAndSend returned normally. Publishing is fire and
+		// forget at the protocol level, so even with mandatory=true the failure arrives
+		// asynchronously on another thread. Nothing throws.
+	}
+
+	@Test
+	void redeclaringAQueueWithDifferentArgumentsIsAPreconditionFailure() {
+		// orders.ttl already exists with x-message-ttl=1500. Same name, different argument.
+		Queue conflicting = QueueBuilder.durable(Topology.Q_TTL)
+				.ttl(9999)
+				.deadLetterExchange(Topology.DLX)
+				.deadLetterRoutingKey("failed")
+				.build();
+
+		assertThatExceptionOfType(AmqpIOException.class)
+				.isThrownBy(() -> this.admin.declareQueue(conflicting))
+				.satisfies((ex) -> {
+					String detail = rootMessage(ex);
+					System.out.println("=== redeclaring orders.ttl with x-message-ttl=9999 ===");
+					System.out.println("  " + detail);
+					assertThat(detail).contains("PRECONDITION_FAILED")
+							.contains("inequivalent arg 'x-message-ttl'");
+				});
+
+		// Queue arguments are immutable. There is no ALTER QUEUE. Changing a TTL, a max-length
+		// or a dead-letter exchange on an existing queue means: declare a new queue, move the
+		// consumers, drain the old one, delete it. Plan the rename into the change.
+	}
+
+	private static String rootMessage(Throwable throwable) {
+		// The useful text is on the cause. AmqpIOException's own message is just
+		// "java.io.IOException", which is why this failure is so often reported as "IOException"
+		// with no further detail.
+		Throwable current = throwable;
+		StringBuilder all = new StringBuilder();
+		while (current != null) {
+			all.append(current.getMessage()).append(" | ");
+			current = current.getCause();
+		}
+		return all.toString();
+	}
+
+}