asmhatre/spring-messaging-demo
1
0
Fork 0
Code Activity

Compare commits

base: asmhatre:f4b0112e2c3c807bb340783dfa1cbf38fc2ae006
Branches Tags
asmhatre:main
...
compare: asmhatre:2919282a5c8da79a417c5089976522fd7cda3569
Branches Tags
asmhatre:main

2 Commits
f4b0112e2c ... 2919282a5c

Author SHA1 Message Date
Ankur Mhatre
2919282a5c Add the kafka-error-handling module 2026-08-29 10:14:23 +05:30
Ankur Mhatre
8b5587cf88 Add the rabbitmq module 2026-08-29 09:58:34 +05:30
51 changed files with 2769 additions and 0 deletions
Expand all files Collapse all files

8
README.md
View File

@@ -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

59
kafka-error-handling/README.md Normal file
View File

@@ -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`.

53
kafka-error-handling/docs/01-two-kinds-of-failure.md Normal file
View File

@@ -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)

71
kafka-error-handling/docs/02-default-error-handler.md Normal file
View File

@@ -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)

95
kafka-error-handling/docs/03-poison-pills.md Normal file
View File

@@ -0,0 +1,95 @@
[&larr; DefaultErrorHandler](02-default-error-handler.md) &middot; [Module README](../README.md) &middot; [The DLT &rarr;](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)

74
kafka-error-handling/docs/04-the-dlt.md Normal file
View File

@@ -0,0 +1,74 @@
[&larr; Poison pills](03-poison-pills.md) &middot; [Module README](../README.md) &middot; [Retry topics &rarr;](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)

89
kafka-error-handling/docs/05-retryable-topic.md Normal file
View File

@@ -0,0 +1,89 @@
[&larr; The DLT](04-the-dlt.md) &middot; [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)

6
kafka-error-handling/docs/output/default-backoff.txt Normal file
View File

@@ -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

15
kafka-error-handling/docs/output/poison-pill.txt Normal file
View File

@@ -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

15
kafka-error-handling/docs/output/retry-and-dlt.txt Normal file
View File

@@ -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

8
kafka-error-handling/docs/output/retry-topics.txt Normal file
View File

@@ -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]

4
kafka-error-handling/docs/output/tests.txt Normal file
View File

@@ -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

84
kafka-error-handling/pom.xml Normal file
View File

@@ -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>

14
kafka-error-handling/scripts/run-all.sh Executable file
View File

@@ -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/

114
kafka-error-handling/src/main/java/com/ankurm/kafkaerrors/ErrorHandlerConfiguration.java Normal file
View File

@@ -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();
}
}

20
kafka-error-handling/src/main/java/com/ankurm/kafkaerrors/ErrorHandlingApplication.java Normal file
View File

@@ -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);
}
}

33
kafka-error-handling/src/main/java/com/ankurm/kafkaerrors/Failures.java Normal file
View File

@@ -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() {
}
}

10
kafka-error-handling/src/main/java/com/ankurm/kafkaerrors/Payment.java Normal file
View File

@@ -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");
}
}

59
kafka-error-handling/src/main/java/com/ankurm/kafkaerrors/PaymentListener.java Normal file
View File

@@ -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();
}
}

85
kafka-error-handling/src/main/java/com/ankurm/kafkaerrors/RetryableTopicListener.java Normal file
View File

@@ -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;
}
}

27
kafka-error-handling/src/main/resources/application.yaml Normal file
View File

@@ -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

83
kafka-error-handling/src/test/java/com/ankurm/kafkaerrors/ByteAwareDltTest.java Normal file
View File

@@ -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");
}
}
}

192
kafka-error-handling/src/test/java/com/ankurm/kafkaerrors/DeadLetterTopicTest.java Normal file
View File

@@ -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);
});
}
}

48
kafka-error-handling/src/test/java/com/ankurm/kafkaerrors/DefaultBackOffTest.java Normal file
View File

@@ -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();
}
}

82
kafka-error-handling/src/test/java/com/ankurm/kafkaerrors/RetryableTopicTest.java Normal file
View File

@@ -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");
}
}

3
kafka-error-handling/src/test/resources/application-test.yaml Normal file
View File

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

80
rabbitmq/README.md Normal file
View File

@@ -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`.

59
rabbitmq/docs/01-the-on-ramp.md Normal file
View File

@@ -0,0 +1,59 @@
[Module README](../README.md) &middot; [Exchanges &rarr;](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)

71
rabbitmq/docs/02-exchanges.md Normal file
View File

@@ -0,0 +1,71 @@
[&larr; The on-ramp](01-the-on-ramp.md) &middot; [Module README](../README.md) &middot; [The silent drop &rarr;](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 &rarr;](03-the-silent-drop.md)

78
rabbitmq/docs/03-the-silent-drop.md Normal file
View File

@@ -0,0 +1,78 @@
[&larr; Exchanges](02-exchanges.md) &middot; [Module README](../README.md) &middot; [Dead-lettering &rarr;](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 &rarr;](04-dead-lettering.md)

97
rabbitmq/docs/04-dead-lettering.md Normal file
View File

@@ -0,0 +1,97 @@
[&larr; The silent drop](03-the-silent-drop.md) &middot; [Module README](../README.md) &middot; [Acknowledgement &rarr;](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 &rarr;](05-acknowledgement.md)

82
rabbitmq/docs/05-acknowledgement.md Normal file
View File

@@ -0,0 +1,82 @@
[&larr; Dead-lettering](04-dead-lettering.md) &middot; [Module README](../README.md) &middot; [Changing your mind &rarr;](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** &mdash; 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 &rarr;](06-changing-your-mind.md)

58
rabbitmq/docs/06-changing-your-mind.md Normal file
View File

@@ -0,0 +1,58 @@
[&larr; Acknowledgement](05-acknowledgement.md) &middot; [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)

14
rabbitmq/docs/output/dead-letter.txt Normal file
View File

@@ -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

2
rabbitmq/docs/output/precondition-failed.txt Normal file
View File

@@ -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) |

4
rabbitmq/docs/output/requeue-loop.txt Normal file
View File

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

3
rabbitmq/docs/output/tests.txt Normal file
View File

@@ -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

7
rabbitmq/docs/output/topic-wildcards.txt Normal file
View File

@@ -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

5
rabbitmq/docs/output/unroutable.txt Normal file
View File

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

67
rabbitmq/pom.xml Normal file
View File

@@ -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>

28
rabbitmq/scripts/broker.sh Executable file
View File

@@ -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

18
rabbitmq/scripts/run-all.sh Executable file
View File

@@ -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/

36
rabbitmq/src/main/java/com/ankurm/rabbit/ConverterConfiguration.java Normal file
View File

@@ -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();
}
}

13
rabbitmq/src/main/java/com/ankurm/rabbit/OrderMessage.java Normal file
View File

@@ -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);
}
}

20
rabbitmq/src/main/java/com/ankurm/rabbit/RabbitDemoApplication.java Normal file
View File

@@ -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);
}
}

217
rabbitmq/src/main/java/com/ankurm/rabbit/Topology.java Normal file
View File

@@ -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();
}
}

31
rabbitmq/src/main/resources/application.yaml Normal file
View File

@@ -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

143
rabbitmq/src/test/java/com/ankurm/rabbit/DeadLetterTest.java Normal file
View File

@@ -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);
}
}

152
rabbitmq/src/test/java/com/ankurm/rabbit/RoutingTest.java Normal file
View File

@@ -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;
});
}
}

33
rabbitmq/src/test/java/com/ankurm/rabbit/TestcontainersConfiguration.java Normal file
View File

@@ -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"));
}
}

100
rabbitmq/src/test/java/com/ankurm/rabbit/TopologyTrapsTest.java Normal file
View File

@@ -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();
}
}