From f4b0112e2c3c807bb340783dfa1cbf38fc2ae006 Mon Sep 17 00:00:00 2001 From: Ankur Mhatre Date: Sat, 29 Aug 2026 10:14:23 +0530 Subject: [PATCH] Add the kafka-error-handling module --- README.md | 1 + kafka-error-handling/README.md | 59 ++++++ .../docs/01-two-kinds-of-failure.md | 53 +++++ .../docs/02-default-error-handler.md | 71 +++++++ kafka-error-handling/docs/03-poison-pills.md | 95 +++++++++ kafka-error-handling/docs/04-the-dlt.md | 74 +++++++ .../docs/05-retryable-topic.md | 89 ++++++++ .../docs/output/default-backoff.txt | 6 + .../docs/output/poison-pill.txt | 15 ++ .../docs/output/retry-and-dlt.txt | 15 ++ .../docs/output/retry-topics.txt | 8 + kafka-error-handling/docs/output/tests.txt | 4 + kafka-error-handling/pom.xml | 84 ++++++++ kafka-error-handling/scripts/run-all.sh | 14 ++ .../ErrorHandlerConfiguration.java | 114 +++++++++++ .../kafkaerrors/ErrorHandlingApplication.java | 20 ++ .../java/com/ankurm/kafkaerrors/Failures.java | 33 +++ .../java/com/ankurm/kafkaerrors/Payment.java | 10 + .../ankurm/kafkaerrors/PaymentListener.java | 59 ++++++ .../kafkaerrors/RetryableTopicListener.java | 85 ++++++++ .../src/main/resources/application.yaml | 27 +++ .../ankurm/kafkaerrors/ByteAwareDltTest.java | 83 ++++++++ .../kafkaerrors/DeadLetterTopicTest.java | 192 ++++++++++++++++++ .../kafkaerrors/DefaultBackOffTest.java | 48 +++++ .../kafkaerrors/RetryableTopicTest.java | 82 ++++++++ .../src/test/resources/application-test.yaml | 3 + 26 files changed, 1344 insertions(+) create mode 100644 kafka-error-handling/README.md create mode 100644 kafka-error-handling/docs/01-two-kinds-of-failure.md create mode 100644 kafka-error-handling/docs/02-default-error-handler.md create mode 100644 kafka-error-handling/docs/03-poison-pills.md create mode 100644 kafka-error-handling/docs/04-the-dlt.md create mode 100644 kafka-error-handling/docs/05-retryable-topic.md create mode 100644 kafka-error-handling/docs/output/default-backoff.txt create mode 100644 kafka-error-handling/docs/output/poison-pill.txt create mode 100644 kafka-error-handling/docs/output/retry-and-dlt.txt create mode 100644 kafka-error-handling/docs/output/retry-topics.txt create mode 100644 kafka-error-handling/docs/output/tests.txt create mode 100644 kafka-error-handling/pom.xml create mode 100755 kafka-error-handling/scripts/run-all.sh create mode 100644 kafka-error-handling/src/main/java/com/ankurm/kafkaerrors/ErrorHandlerConfiguration.java create mode 100644 kafka-error-handling/src/main/java/com/ankurm/kafkaerrors/ErrorHandlingApplication.java create mode 100644 kafka-error-handling/src/main/java/com/ankurm/kafkaerrors/Failures.java create mode 100644 kafka-error-handling/src/main/java/com/ankurm/kafkaerrors/Payment.java create mode 100644 kafka-error-handling/src/main/java/com/ankurm/kafkaerrors/PaymentListener.java create mode 100644 kafka-error-handling/src/main/java/com/ankurm/kafkaerrors/RetryableTopicListener.java create mode 100644 kafka-error-handling/src/main/resources/application.yaml create mode 100644 kafka-error-handling/src/test/java/com/ankurm/kafkaerrors/ByteAwareDltTest.java create mode 100644 kafka-error-handling/src/test/java/com/ankurm/kafkaerrors/DeadLetterTopicTest.java create mode 100644 kafka-error-handling/src/test/java/com/ankurm/kafkaerrors/DefaultBackOffTest.java create mode 100644 kafka-error-handling/src/test/java/com/ankurm/kafkaerrors/RetryableTopicTest.java create mode 100644 kafka-error-handling/src/test/resources/application-test.yaml diff --git a/README.md b/README.md index d85f4a9..ad6412c 100644 --- a/README.md +++ b/README.md @@ -8,6 +8,7 @@ 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 diff --git a/kafka-error-handling/README.md b/kafka-error-handling/README.md new file mode 100644 index 0000000..2332f8c --- /dev/null +++ b/kafka-error-handling/README.md @@ -0,0 +1,59 @@ +# `kafka-error-handling` — retries, DLT and poison pills + +Companion project for +[**Kafka Error Handling with Spring Kafka 4.1: DLT, Retry Topics and Poison Pills**](https://ankurm.com/spring-kafka-4-1-error-handling-dlt-retry-topics/) +on ankurm.com. + +Six tests against a **real Kafka broker** started in-process in KRaft mode. No Docker, no local +install. `./scripts/run-all.sh` regenerates everything under [`docs/output/`](docs/output/). + +## Versions + +| | Version | +|---|---| +| JDK | 25 (Temurin 25.0.4.1+1) | +| Spring Boot | 4.1.1 | +| Spring Kafka | 4.1.1 | +| kafka-clients | 4.2.1 (Boot-managed) | + +## Profiles + +| Profile | What it wires | +|---|---| +| `dlt` | `DefaultErrorHandler` + `DeadLetterPublishingRecoverer`, `FixedBackOff(1000, 2)`, `PermanentFailure` classified non-retryable | +| `dltbytes` | the same, with a `byte[]`-aware template map so poison pills keep their original bytes | +| `retrytopic` | `@RetryableTopic` non-blocking retries with a `@DltHandler` | +| `defaults` | the stock `DefaultErrorHandler`, for reading its behaviour | + +## Documentation + +1. [Two kinds of failure, and why they need different machinery](docs/01-two-kinds-of-failure.md) +2. [What the default actually does](docs/02-default-error-handler.md) +3. [Poison pills](docs/03-poison-pills.md) +4. [The dead-letter topic](docs/04-the-dlt.md) +5. [Non-blocking retries with `@RetryableTopic`](docs/05-retryable-topic.md) + +## Captured output + +| File | Shows | +|---|---| +| [`default-backoff.txt`](docs/output/default-backoff.txt) | ten deliveries, zero delay | +| [`retry-and-dlt.txt`](docs/output/retry-and-dlt.txt) | measured back-off and the DLT headers | +| [`poison-pill.txt`](docs/output/poison-pill.txt) | base64 payload, and the byte-aware fix | +| [`retry-topics.txt`](docs/output/retry-topics.txt) | the non-blocking delivery trace | +| [`tests.txt`](docs/output/tests.txt) | 6 tests | + +## Six things this module exists to prove + +1. **The default is ten deliveries, zero milliseconds apart, then the record is dropped.** Not + "retry with backoff", and not "dead-letter". +2. **The DLT suffix is `-dlt`, not `.DLT`.** Get it wrong and the recoverer logs a WARN and the + record is lost — your safety net silently drops it. +3. **`kafka_dlt-exception-fqcn` is always `ListenerExecutionFailedException`** for listener + failures. The useful header is `-exception-cause-fqcn`. +4. **A poison pill reaches the DLT base64-encoded**, because the recoverer reuses the JSON + producer. A per-type template map fixes it; the module shows both transcripts. +5. **`@RetryableTopic` names retry topics by delay** — `-retry-500`, `-retry-1000` — so changing + the multiplier renames them. +6. **`@Backoff` from spring-retry no longer exists here.** Spring Kafka 4 ships its own + `@BackOff`, and the attribute is `backOff`. diff --git a/kafka-error-handling/docs/01-two-kinds-of-failure.md b/kafka-error-handling/docs/01-two-kinds-of-failure.md new file mode 100644 index 0000000..36c222f --- /dev/null +++ b/kafka-error-handling/docs/01-two-kinds-of-failure.md @@ -0,0 +1,53 @@ +[Module README](../README.md) · [DefaultErrorHandler →](02-default-error-handler.md) + +# 1. Two kinds of failure, and why they need different machinery + +Kafka delivery is at-least-once. The container commits offsets after your listener returns +([the basics article](https://ankurm.com/spring-boot-4-1-kafka-producer-consumer-serialisation/) +covers why), so if the listener throws, the offset does not move and the record comes back. +Everything in this module is about what happens next. + +There are two failures, and conflating them is why a bad record can take a partition down for +hours. + +## Failure inside the listener + +Your code threw. The record deserialized fine; the work failed. The container catches it, hands +it to a `CommonErrorHandler`, and that decides whether to retry, how long to wait, and what to do +when the attempts run out. + +This splits again, and the split matters more than any back-off setting: + +| | example | retrying it | +|---|---|---| +| **transient** | timeout, 503, deadlock, connection reset | may succeed | +| **permanent** | validation failure, missing entity, malformed field | will fail identically | + +Retrying a permanent failure ten times buys nothing and costs ten times the latency plus nine +misleading log lines. Spring Kafka lets you say so: + +```java +handler.addNotRetryableExceptions(PermanentFailure.class); +``` + +That single line is worth more than tuning the back-off, and it is the one most people skip. + +## Failure before the listener + +The bytes on the topic are not what the deserializer expects. Someone changed a schema, or +published with a different serializer, or your `__TypeId__` header names a class you do not +trust. + +This one is nastier, because it happens **inside `poll()`**, before any listener exists to throw +from. There is no error handler in the path. The consumer cannot advance past the offset, +retries the same record on the next poll, fails again, and does that forever — at whatever rate +the poll loop runs. The partition is stopped and the only symptom is a growing lag with a +consumer that looks healthy. + +That is a **poison pill**, and the cure is a different mechanism from the one above: +`ErrorHandlingDeserializer`, covered in [chapter 3](03-poison-pills.md). + +Keeping these two apart is the whole point of this module. One needs a retry policy; the other +needs a wrapper around the deserializer. Neither fixes the other. + +[DefaultErrorHandler →](02-default-error-handler.md) diff --git a/kafka-error-handling/docs/02-default-error-handler.md b/kafka-error-handling/docs/02-default-error-handler.md new file mode 100644 index 0000000..65d5892 --- /dev/null +++ b/kafka-error-handling/docs/02-default-error-handler.md @@ -0,0 +1,71 @@ +[← Two kinds of failure](01-two-kinds-of-failure.md) · [Module README](../README.md) · [Poison pills →](03-poison-pills.md) + +# 2. What the default actually does + +If you configure nothing, the container factory installs a `DefaultErrorHandler` with +`SeekUtils.DEFAULT_BACK_OFF` and a recoverer that logs. Run the back-off and read it off +([`docs/output/default-backoff.txt`](output/default-backoff.txt)): + +``` +=== DefaultErrorHandler default back-off === + interval 0 ms + max attempts 9 retries + SeekUtils.DEFAULT_MAX_FAILURES = 10 + retry intervals [0, 0, 0, 0, 0, 0, 0, 0, 0] + total deliveries 10 +``` + +**Ten deliveries, zero milliseconds apart, and then the record is dropped.** + +Both halves of that surprise people. It is not "retry with backoff" — it is ten immediate +attempts as fast as the consumer thread can run them, which against a downstream that is +overloaded is ten times the load at the worst moment. And "then dropped" means exactly that: the +default recoverer logs the failure and the offset moves on. There is no dead-letter topic unless +you make one. + +## Giving it a back-off and a destination + +```java +@Bean +DefaultErrorHandler errorHandler(KafkaOperations 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 →](03-poison-pills.md) diff --git a/kafka-error-handling/docs/03-poison-pills.md b/kafka-error-handling/docs/03-poison-pills.md new file mode 100644 index 0000000..4063689 --- /dev/null +++ b/kafka-error-handling/docs/03-poison-pills.md @@ -0,0 +1,95 @@ +[← DefaultErrorHandler](02-default-error-handler.md) · [Module README](../README.md) · [The DLT →](04-the-dlt.md) + +# 3. Poison pills + +A record whose bytes cannot be deserialized fails inside `poll()`, before any listener exists. +No error handler is in the path. The offset cannot advance, so the next poll fetches the same +record and fails identically. Forever. + +The consumer is up, the group is stable, no exception reaches your code, and lag grows. It is +one of the few Kafka failures with no good symptom. + +## `ErrorHandlingDeserializer` + +```yaml +spring: + kafka: + consumer: + value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer + properties: + spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JacksonJsonDeserializer +``` + +It wraps the real deserializer, catches the failure, and returns a **null value with the +exception in a header**. The record then flows normally into the container, the listener is +skipped, and the error handler gets a record it can recover — which is to say, the poison pill +becomes an ordinary failure. + +Use `spring.deserializer.key.delegate.class` for keys. A malformed key is rarer and just as +fatal. + +## What arrives on the DLT + +From [`docs/output/poison-pill.txt`](output/poison-pill.txt): + +``` +kafka_dlt-exception-fqcn org.springframework.kafka.support.serializer.DeserializationException +kafka_dlt-exception-cause-fqcn org.springframework.kafka.support.serializer.DeserializationException +kafka_dlt-exception-message failed to deserialize +kafka_dlt-original-topic payments +kafka_dlt-original-partition 0x00000000 +kafka_dlt-original-offset 0x0000000000000000 +kafka_dlt-original-consumer-group payments +__TypeId__ [B +DLT payload -> "eyB0aGlzIGlzIG5vdCBqc29u" +``` + +The listener was never invoked — asserted in the test — and the record is off the partition, +which is the whole win. + +Two details in that block are worth stopping on. + +**`original-partition` and `original-offset` are binary.** They are big-endian `int` and `long`, +not text. Printing them as a string gives you mojibake, which is why the transcript renders them +as hex. A DLT tool that treats every header as UTF-8 will show garbage for exactly the three +fields you need in order to find the original record. + +**The payload is base64.** `"eyB0aGlzIGlzIG5vdCBqc29u"` decodes to `{ this is not json`. The +recoverer publishes with the **application's** producer, whose value serializer is +`JacksonJsonSerializer`; the failed value is a `byte[]`; Jackson writes a `byte[]` as a base64 +JSON string. So the DLT does not hold what arrived — it holds base64 of it, wrapped in quotes. + +Replaying that topic naively republishes a quoted base64 string, which fails to deserialize, and +now you have a poison pill in your poison-pill queue. + +## The fix + +Give the recoverer a template per value type: + +```java +Map, 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` 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, KafkaOperations>`, matched by value type, with + `Object.class` as the fallback. + +[The DLT →](04-the-dlt.md) diff --git a/kafka-error-handling/docs/04-the-dlt.md b/kafka-error-handling/docs/04-the-dlt.md new file mode 100644 index 0000000..c74ba5e --- /dev/null +++ b/kafka-error-handling/docs/04-the-dlt.md @@ -0,0 +1,74 @@ +[← Poison pills](03-poison-pills.md) · [Module README](../README.md) · [Retry topics →](05-retryable-topic.md) + +# 4. The dead-letter topic + +## The suffix is `-dlt`, not `.DLT` + +```java +public static final String RetryTopicConstants.DEFAULT_RETRY_SUFFIX = "-retry"; +public static final String RetryTopicConstants.DEFAULT_DLT_SUFFIX = "-dlt"; +``` + +Older Spring Kafka used `.DLT`, and most of the material online still says so. Getting it wrong +is not an exception — it is this, at WARN, once per record: + +``` +o.s.k.l.DeadLetterPublishingRecoverer : Destination resolver returned non-existent partition +payments-dlt-0, KafkaProducer will determine partition to use for this topic +[Producer] ... {payments-dlt=UNKNOWN_TOPIC_OR_PARTITION} +``` + +and then, on a cluster with auto-topic-creation disabled, the record is **gone**. Your safety net +dropped it and logged a warning. This module's tests were written against `payments.DLT` first +and failed exactly this way. + +Two things follow: pre-create your DLT topics as part of provisioning, and alert on that WARN. + +## Same partition by default + +`DeadLetterPublishingRecoverer` publishes to the **same partition number** as the original. If +your DLT has fewer partitions than the source topic, records from the high-numbered partitions +have nowhere to go. Either give the DLT the same partition count, or set + +```java +recoverer.setPartitionResolver((record, ex) -> null); // let the producer choose +``` + +## The headers, and the one that will mislead you + +From [`docs/output/retry-and-dlt.txt`](output/retry-and-dlt.txt): + +``` +kafka_dlt-exception-fqcn org.springframework.kafka.listener.ListenerExecutionFailedException +kafka_dlt-exception-cause-fqcn com.ankurm.kafkaerrors.Failures$TransientFailure +kafka_dlt-original-topic payments +kafka_dlt-original-consumer-group payments +``` + +**`kafka_dlt-exception-fqcn` is always the wrapper** for a listener failure. Build a DLT triage +dashboard grouped by that header and every failure in the estate lands in one bucket called +`ListenerExecutionFailedException`. The field you want is `kafka_dlt-exception-cause-fqcn`. + +(For a deserialization failure there is no wrapper, so the two headers agree. That inconsistency +is worth knowing if you are writing a tool over them.) + +`kafka_dlt-original-consumer-group` is the one that saves you when several groups consume the +same topic and share a DLT. + +## Replay + +A DLT is only useful if you can put records back. The mechanics are a copy: + +1. read from `-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 →](05-retryable-topic.md) diff --git a/kafka-error-handling/docs/05-retryable-topic.md b/kafka-error-handling/docs/05-retryable-topic.md new file mode 100644 index 0000000..98bc7ac --- /dev/null +++ b/kafka-error-handling/docs/05-retryable-topic.md @@ -0,0 +1,89 @@ +[← The DLT](04-the-dlt.md) · [Module README](../README.md) + +# 5. Non-blocking retries with `@RetryableTopic` + +Blocking retries stall the partition. `@RetryableTopic` republishes the failed record to a +separate topic and lets the main partition carry on. + +```java +@RetryableTopic(attempts = "4", backOff = @BackOff(delay = 500, multiplier = 2.0), + sameIntervalTopicReuseStrategy = SameIntervalTopicReuseStrategy.SINGLE_TOPIC, + exclude = Failures.PermanentFailure.class) +@KafkaListener(topics = "invoices", groupId = "invoices") +public void onInvoice(ConsumerRecord 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 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) diff --git a/kafka-error-handling/docs/output/default-backoff.txt b/kafka-error-handling/docs/output/default-backoff.txt new file mode 100644 index 0000000..aad174e --- /dev/null +++ b/kafka-error-handling/docs/output/default-backoff.txt @@ -0,0 +1,6 @@ +=== DefaultErrorHandler default back-off === + interval 0 ms + max attempts 9 retries + SeekUtils.DEFAULT_MAX_FAILURES = 10 + retry intervals [0, 0, 0, 0, 0, 0, 0, 0, 0] + total deliveries 10 diff --git a/kafka-error-handling/docs/output/poison-pill.txt b/kafka-error-handling/docs/output/poison-pill.txt new file mode 100644 index 0000000..bcf8977 --- /dev/null +++ b/kafka-error-handling/docs/output/poison-pill.txt @@ -0,0 +1,15 @@ +=== poison pill on the DLT === + kafka_dlt-exception-fqcn org.springframework.kafka.support.serializer.DeserializationException + kafka_dlt-exception-cause-fqcn org.springframework.kafka.support.serializer.DeserializationException + kafka_dlt-exception-message failed to deserialize + kafka_dlt-exception-stacktrace org.springframework.kafka.support.serializer.DeserializationException: failed to deseriali... + kafka_dlt-original-topic payments + kafka_dlt-original-partition 0x00000000 + kafka_dlt-original-offset 0x0000000000000000 + kafka_dlt-original-timestamp 0x000001a04bd35409 + kafka_dlt-original-timestamp-type CreateTime + kafka_dlt-original-consumer-group payments + __TypeId__ [B + DLT payload -> "eyB0aGlzIGlzIG5vdCBqc29u" +=== byte-aware DLT === + DLT payload -> { this is not json diff --git a/kafka-error-handling/docs/output/retry-and-dlt.txt b/kafka-error-handling/docs/output/retry-and-dlt.txt new file mode 100644 index 0000000..76a9859 --- /dev/null +++ b/kafka-error-handling/docs/output/retry-and-dlt.txt @@ -0,0 +1,15 @@ +=== transient failure === + deliveries 3 + gap between 1&2 1007 ms (FixedBackOff interval 1000) +=== transient failure on the DLT === + kafka_dlt-exception-fqcn org.springframework.kafka.listener.ListenerExecutionFailedException + kafka_dlt-exception-cause-fqcn com.ankurm.kafkaerrors.Failures$TransientFailure + kafka_dlt-exception-message Listener method 'public void com.ankurm.kafkaerrors.PaymentListener.onPayment(org.apache.k... + kafka_dlt-exception-stacktrace org.springframework.kafka.listener.ListenerExecutionFailedException: Listener method 'publ... + kafka_dlt-original-topic payments + kafka_dlt-original-partition 0x00000000 + kafka_dlt-original-offset 0x0000000000000002 + kafka_dlt-original-timestamp 0x000001a04bd35f33 + kafka_dlt-original-timestamp-type CreateTime + kafka_dlt-original-consumer-group payments + __TypeId__ com.ankurm.kafkaerrors.Payment diff --git a/kafka-error-handling/docs/output/retry-topics.txt b/kafka-error-handling/docs/output/retry-topics.txt new file mode 100644 index 0000000..f814fbe --- /dev/null +++ b/kafka-error-handling/docs/output/retry-topics.txt @@ -0,0 +1,8 @@ +=== @RetryableTopic delivery trace === + +0 ms invoices transient-1 + +522 ms invoices-retry-500 transient-1 + +539 ms invoices ok-1 + +1548 ms invoices-retry-1000 transient-1 + +3554 ms invoices-retry-2000 transient-1 + DLT: [transient-1 on invoices-dlt] + topics touched: [invoices, invoices-retry-1000, invoices-retry-2000, invoices-retry-500] diff --git a/kafka-error-handling/docs/output/tests.txt b/kafka-error-handling/docs/output/tests.txt new file mode 100644 index 0000000..d8c9300 --- /dev/null +++ b/kafka-error-handling/docs/output/tests.txt @@ -0,0 +1,4 @@ +Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 9.009 s -- in com.ankurm.kafkaerrors.RetryableTopicTest +Tests run: 3, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 6.809 s -- in com.ankurm.kafkaerrors.DeadLetterTopicTest +Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 0.003 s -- in com.ankurm.kafkaerrors.DefaultBackOffTest +Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 2.935 s -- in com.ankurm.kafkaerrors.ByteAwareDltTest diff --git a/kafka-error-handling/pom.xml b/kafka-error-handling/pom.xml new file mode 100644 index 0000000..9387b9f --- /dev/null +++ b/kafka-error-handling/pom.xml @@ -0,0 +1,84 @@ + + 4.0.0 + + + + org.springframework.boot + spring-boot-starter-parent + 4.1.1 + + + + com.ankurm + kafka-error-handling + 1.0 + jar + + + 25 + UTF-8 + + + + + org.springframework.boot + spring-boot-starter + + + + org.springframework.boot + spring-boot-starter-kafka + + + + org.springframework.boot + spring-boot-starter-jackson + + + + org.springframework.boot + spring-boot-starter-test + test + + + + org.springframework.kafka + spring-kafka-test + test + + + org.springframework.boot + spring-boot-testcontainers + test + + + + org.testcontainers + testcontainers-kafka + test + + + + + + + org.springframework.boot + spring-boot-maven-plugin + + + + diff --git a/kafka-error-handling/scripts/run-all.sh b/kafka-error-handling/scripts/run-all.sh new file mode 100755 index 0000000..a5887ef --- /dev/null +++ b/kafka-error-handling/scripts/run-all.sh @@ -0,0 +1,14 @@ +#!/usr/bin/env bash +# Regenerate every file under docs/output/. A real Kafka broker starts in-process in KRaft mode; +# no Docker and no local install. +set -eu +cd "$(dirname "$0")/.." +mvn -B test 2>&1 | tr -d '\000' | tee /tmp/eh-test.log > /dev/null +sed -n '/=== DefaultErrorHandler default back-off ===/,/total deliveries/p' /tmp/eh-test.log > docs/output/default-backoff.txt +sed -n '/=== transient failure ===/,/gap between/p' /tmp/eh-test.log > docs/output/retry-and-dlt.txt +sed -n '/=== transient failure on the DLT ===/,/__TypeId__/p' /tmp/eh-test.log >> docs/output/retry-and-dlt.txt +sed -n '/=== poison pill on the DLT ===/,/DLT payload/p' /tmp/eh-test.log > docs/output/poison-pill.txt +sed -n '/=== byte-aware DLT ===/,/DLT payload/p' /tmp/eh-test.log >> docs/output/poison-pill.txt +sed -n '/=== @RetryableTopic delivery trace ===/,/topics touched/p' /tmp/eh-test.log > docs/output/retry-topics.txt +grep -aE 'Tests run:.*in com\.ankurm' /tmp/eh-test.log | sed 's/^\[INFO\] //' > docs/output/tests.txt +echo "regenerated:"; ls -1 docs/output/ diff --git a/kafka-error-handling/src/main/java/com/ankurm/kafkaerrors/ErrorHandlerConfiguration.java b/kafka-error-handling/src/main/java/com/ankurm/kafkaerrors/ErrorHandlerConfiguration.java new file mode 100644 index 0000000..038eff3 --- /dev/null +++ b/kafka-error-handling/src/main/java/com/ankurm/kafkaerrors/ErrorHandlerConfiguration.java @@ -0,0 +1,114 @@ +package com.ankurm.kafkaerrors; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Profile; +import org.springframework.kafka.core.KafkaOperations; +import org.springframework.kafka.listener.DeadLetterPublishingRecoverer; +import org.springframework.kafka.listener.DefaultErrorHandler; +import org.springframework.boot.kafka.autoconfigure.KafkaProperties; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.KafkaTemplate; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.common.serialization.ByteArraySerializer; +import org.springframework.util.backoff.FixedBackOff; + +import java.util.HashMap; +import java.util.Map; + +/** + * The error handler, in three configurations selected by profile. + * + *

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 — so out of the box a record is retried and then dropped, + * which is the behaviour most people are surprised by. + * + * @see docs/02-default-error-handler.md + */ +@Configuration(proxyBeanMethods = false) +public class ErrorHandlerConfiguration { + + private static final Logger log = LoggerFactory.getLogger(ErrorHandlerConfiguration.class); + + /** + * Publishes the exhausted record to {@code .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 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. + * + *

{@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 no longer resolves; the + // generic check runs before the qualifier, so @Qualifier alone does not rescue it. + // 3. The recoverer's constructor takes Map, KafkaOperations>, so wildcards + // are what it wants anyway. + @Bean + @Profile("dltbytes") + DefaultErrorHandler byteAwareDltErrorHandler( + @Qualifier("jsonTemplate") KafkaOperations jsonTemplate, + @Qualifier("byteTemplate") KafkaOperations byteTemplate) { + Map, 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 jsonTemplate(KafkaProperties properties) { + return new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(properties.buildProducerProperties())); + } + + @Bean + @Profile("dltbytes") + KafkaTemplate byteTemplate(KafkaProperties properties) { + Map 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(); + } + +} diff --git a/kafka-error-handling/src/main/java/com/ankurm/kafkaerrors/ErrorHandlingApplication.java b/kafka-error-handling/src/main/java/com/ankurm/kafkaerrors/ErrorHandlingApplication.java new file mode 100644 index 0000000..0dc08e4 --- /dev/null +++ b/kafka-error-handling/src/main/java/com/ankurm/kafkaerrors/ErrorHandlingApplication.java @@ -0,0 +1,20 @@ +package com.ankurm.kafkaerrors; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; + +/** + * What Spring Kafka does when your listener throws, and what it does when the record cannot even + * be turned into an object. + * + *

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); + } + +} diff --git a/kafka-error-handling/src/main/java/com/ankurm/kafkaerrors/Failures.java b/kafka-error-handling/src/main/java/com/ankurm/kafkaerrors/Failures.java new file mode 100644 index 0000000..ffd3f57 --- /dev/null +++ b/kafka-error-handling/src/main/java/com/ankurm/kafkaerrors/Failures.java @@ -0,0 +1,33 @@ +package com.ankurm.kafkaerrors; + +/** + * The two failure classes that Spring Kafka treats completely differently, made explicit so the + * tests can be about the distinction rather than about a generic RuntimeException. + */ +public final class Failures { + + /** Worth retrying: a timeout, a 503, a deadlock. The same input may succeed later. */ + public static class TransientFailure extends RuntimeException { + + public TransientFailure(String message) { + super(message); + } + + } + + /** + * Not worth retrying: a validation failure, a missing referenced entity, a malformed field. + * The same input will fail identically ten times, and retrying it is pure latency. + */ + public static class PermanentFailure extends RuntimeException { + + public PermanentFailure(String message) { + super(message); + } + + } + + private Failures() { + } + +} diff --git a/kafka-error-handling/src/main/java/com/ankurm/kafkaerrors/Payment.java b/kafka-error-handling/src/main/java/com/ankurm/kafkaerrors/Payment.java new file mode 100644 index 0000000..1025112 --- /dev/null +++ b/kafka-error-handling/src/main/java/com/ankurm/kafkaerrors/Payment.java @@ -0,0 +1,10 @@ +package com.ankurm.kafkaerrors; + +/** The payload. {@code paymentId} doubles as the record key. */ +public record Payment(String paymentId, String status) { + + public static Payment of(String paymentId) { + return new Payment(paymentId, "PENDING"); + } + +} diff --git a/kafka-error-handling/src/main/java/com/ankurm/kafkaerrors/PaymentListener.java b/kafka-error-handling/src/main/java/com/ankurm/kafkaerrors/PaymentListener.java new file mode 100644 index 0000000..4909408 --- /dev/null +++ b/kafka-error-handling/src/main/java/com/ankurm/kafkaerrors/PaymentListener.java @@ -0,0 +1,59 @@ +package com.ankurm.kafkaerrors; + +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.kafka.annotation.KafkaListener; +import org.springframework.stereotype.Component; + +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; + +/** + * A listener that fails on demand, so the retry machinery can be observed rather than described. + * + *

Payment ids beginning {@code transient-} throw a retryable exception; ids beginning + * {@code permanent-} throw a non-retryable one; everything else succeeds. + * + * @see docs/02-default-error-handler.md + */ +@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 attempts = new CopyOnWriteArrayList<>(); + + @KafkaListener(topics = TOPIC, groupId = "payments") + public void onPayment(ConsumerRecord 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 attempts() { + return this.attempts; + } + + public List attemptsFor(String paymentId) { + return this.attempts.stream().filter((a) -> a.paymentId().equals(paymentId)).toList(); + } + + public void clear() { + this.attempts.clear(); + } + +} diff --git a/kafka-error-handling/src/main/java/com/ankurm/kafkaerrors/RetryableTopicListener.java b/kafka-error-handling/src/main/java/com/ankurm/kafkaerrors/RetryableTopicListener.java new file mode 100644 index 0000000..d13128f --- /dev/null +++ b/kafka-error-handling/src/main/java/com/ankurm/kafkaerrors/RetryableTopicListener.java @@ -0,0 +1,85 @@ +package com.ankurm.kafkaerrors; + +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.context.annotation.Profile; +import org.springframework.kafka.annotation.BackOff; +import org.springframework.kafka.annotation.DltHandler; +import org.springframework.kafka.annotation.KafkaListener; +import org.springframework.kafka.annotation.RetryableTopic; +import org.springframework.kafka.retrytopic.SameIntervalTopicReuseStrategy; +import org.springframework.kafka.support.KafkaHeaders; +import org.springframework.messaging.handler.annotation.Header; +import org.springframework.stereotype.Component; + +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; + +/** + * Non-blocking retries. Instead of holding the consumer thread while it re-delivers, + * {@code @RetryableTopic} republishes the failed record to a separate retry topic with a + * timestamp header, and a container for that topic waits before processing it. + * + *

The consequence people miss: the main partition advances immediately. That is the + * entire point — 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 docs/05-retryable-topic.md + */ +@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 deliveries = new CopyOnWriteArrayList<>(); + + private final List 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 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 — the topic still exists and still has the record, but + * nothing in your application has looked at it. + */ + @DltHandler + public void onDlt(ConsumerRecord 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 deliveries() { + return this.deliveries; + } + + public List dead() { + return this.dead; + } + +} diff --git a/kafka-error-handling/src/main/resources/application.yaml b/kafka-error-handling/src/main/resources/application.yaml new file mode 100644 index 0000000..9fd7f55 --- /dev/null +++ b/kafka-error-handling/src/main/resources/application.yaml @@ -0,0 +1,27 @@ +spring: + application: + name: kafka-error-handling + main: + banner-mode: off + kafka: + bootstrap-servers: localhost:9092 + producer: + key-serializer: org.apache.kafka.common.serialization.StringSerializer + value-serializer: org.springframework.kafka.support.serializer.JacksonJsonSerializer + consumer: + group-id: payments + auto-offset-reset: earliest + key-deserializer: org.apache.kafka.common.serialization.StringDeserializer + # ErrorHandlingDeserializer wraps the real one. Without it a record that cannot be + # deserialized is thrown BEFORE the listener exists, the container cannot advance past it, + # and the same offset is retried forever. See docs/03-poison-pills.md. + value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer + properties: + spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JacksonJsonDeserializer + spring.json.trusted.packages: com.ankurm.kafkaerrors + spring.json.value.default.type: com.ankurm.kafkaerrors.Payment +logging: + level: + root: WARN + com.ankurm: INFO + org.springframework.kafka.listener: WARN diff --git a/kafka-error-handling/src/test/java/com/ankurm/kafkaerrors/ByteAwareDltTest.java b/kafka-error-handling/src/test/java/com/ankurm/kafkaerrors/ByteAwareDltTest.java new file mode 100644 index 0000000..24f24af --- /dev/null +++ b/kafka-error-handling/src/test/java/com/ankurm/kafkaerrors/ByteAwareDltTest.java @@ -0,0 +1,83 @@ +package com.ankurm.kafkaerrors; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.apache.kafka.common.serialization.ByteArrayDeserializer; +import org.apache.kafka.common.serialization.ByteArraySerializer; +import org.apache.kafka.common.serialization.StringDeserializer; +import org.apache.kafka.common.serialization.StringSerializer; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.test.EmbeddedKafkaBroker; +import org.springframework.kafka.test.context.EmbeddedKafka; +import org.springframework.kafka.test.utils.KafkaTestUtils; +import org.springframework.test.context.ActiveProfiles; + +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * The fix for the base64 problem in {@link DeadLetterTopicTest}: a per-value-type template map + * so that a {@code byte[]} is published with a {@code ByteArraySerializer}. + * + * @see docs/03-poison-pills.md + */ +@SpringBootTest +@ActiveProfiles({ "test", "dltbytes" }) +@EmbeddedKafka(topics = { "payments", "payments-dlt" }, partitions = 1) +class ByteAwareDltTest { + + @Autowired + EmbeddedKafkaBroker broker; + + @Test + void aByteArrayTemplateKeepsTheOriginalBytesIntact() { + Map producerProps = new HashMap<>(KafkaTestUtils.producerProps(this.broker.getBrokersAsString())); + producerProps.put("key.serializer", StringSerializer.class); + producerProps.put("value.serializer", ByteArraySerializer.class); + DefaultKafkaProducerFactory 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 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 consumer = new KafkaConsumer<>(consumerProps)) { + consumer.subscribe(List.of("payments-dlt")); + long deadline = System.currentTimeMillis() + 30_000; + while (System.currentTimeMillis() < deadline) { + ConsumerRecords records = consumer.poll(Duration.ofMillis(500)); + for (ConsumerRecord 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"); + } + } + +} diff --git a/kafka-error-handling/src/test/java/com/ankurm/kafkaerrors/DeadLetterTopicTest.java b/kafka-error-handling/src/test/java/com/ankurm/kafkaerrors/DeadLetterTopicTest.java new file mode 100644 index 0000000..f745a2f --- /dev/null +++ b/kafka-error-handling/src/test/java/com/ankurm/kafkaerrors/DeadLetterTopicTest.java @@ -0,0 +1,192 @@ +package com.ankurm.kafkaerrors; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.common.serialization.ByteArraySerializer; +import org.apache.kafka.common.serialization.StringSerializer; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer; +import org.springframework.kafka.test.EmbeddedKafkaBroker; +import org.springframework.kafka.test.context.EmbeddedKafka; +import org.springframework.kafka.test.utils.KafkaTestUtils; +import org.springframework.test.context.ActiveProfiles; + +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.HashMap; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +/** + * Retries, classification, and the two very different reasons a record ends up on the DLT. + * + * @see docs/04-the-dlt.md + */ +@SpringBootTest +@ActiveProfiles({ "test", "dlt" }) +@EmbeddedKafka(topics = { "payments", "payments-dlt" }, partitions = 1) +class DeadLetterTopicTest { + + @Autowired + KafkaTemplate 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. + * + *

Note the topic name: {@code payments-dlt}. {@code RetryTopicConstants.DEFAULT_DLT_SUFFIX} + * is {@code "-dlt"}, not {@code ".DLT"} — and if the topic does not exist, the + * recoverer logs a WARN and the record is gone. + */ + private Consumer dltConsumer() { + Map 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 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 consumer = dltConsumer()) { + ConsumerRecord 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 consumer = dltConsumer()) { + ConsumerRecord 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 props = new HashMap<>(KafkaTestUtils.producerProps(this.broker.getBrokersAsString())); + props.put("key.serializer", StringSerializer.class); + props.put("value.serializer", ByteArraySerializer.class); + DefaultKafkaProducerFactory 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 consumer = dltConsumer()) { + ConsumerRecord 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 pollFor(Consumer consumer, String key) { + long deadline = System.currentTimeMillis() + 30_000; + while (System.currentTimeMillis() < deadline) { + ConsumerRecords records = consumer.poll(Duration.ofMillis(500)); + for (ConsumerRecord record : records) { + if (key.equals(record.key())) { + return record; + } + } + } + return null; + } + + private static String header(ConsumerRecord 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 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); + }); + } + +} diff --git a/kafka-error-handling/src/test/java/com/ankurm/kafkaerrors/DefaultBackOffTest.java b/kafka-error-handling/src/test/java/com/ankurm/kafkaerrors/DefaultBackOffTest.java new file mode 100644 index 0000000..89a7abd --- /dev/null +++ b/kafka-error-handling/src/test/java/com/ankurm/kafkaerrors/DefaultBackOffTest.java @@ -0,0 +1,48 @@ +package com.ankurm.kafkaerrors; + +import org.junit.jupiter.api.Test; +import org.springframework.kafka.listener.SeekUtils; +import org.springframework.util.backoff.BackOffExecution; +import org.springframework.util.backoff.FixedBackOff; + +import java.util.ArrayList; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * What the stock {@code DefaultErrorHandler} actually does, read out of its own back-off rather + * than out of the documentation. No broker and no context: a {@code BackOff} is a state machine + * you can just run. + * + * @see docs/02-default-error-handler.md + */ +class DefaultBackOffTest { + + @Test + void theDefaultIsTenAttemptsWithNoDelayAtAll() { + FixedBackOff backOff = SeekUtils.DEFAULT_BACK_OFF; + BackOffExecution execution = backOff.start(); + + List 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(); + } + +} diff --git a/kafka-error-handling/src/test/java/com/ankurm/kafkaerrors/RetryableTopicTest.java b/kafka-error-handling/src/test/java/com/ankurm/kafkaerrors/RetryableTopicTest.java new file mode 100644 index 0000000..a8d34db --- /dev/null +++ b/kafka-error-handling/src/test/java/com/ankurm/kafkaerrors/RetryableTopicTest.java @@ -0,0 +1,82 @@ +package com.ankurm.kafkaerrors; + +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.test.EmbeddedKafkaBroker; +import org.springframework.kafka.test.context.EmbeddedKafka; +import org.springframework.test.context.ActiveProfiles; + +import java.time.Duration; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +/** + * Non-blocking retries, and the topics they create. + * + * @see docs/05-retryable-topic.md + */ +@SpringBootTest +@ActiveProfiles({ "test", "retrytopic" }) +@EmbeddedKafka(topics = { "invoices" }, partitions = 1) +class RetryableTopicTest { + + @Autowired + KafkaTemplate 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 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"); + } + +} diff --git a/kafka-error-handling/src/test/resources/application-test.yaml b/kafka-error-handling/src/test/resources/application-test.yaml new file mode 100644 index 0000000..6e3d04b --- /dev/null +++ b/kafka-error-handling/src/test/resources/application-test.yaml @@ -0,0 +1,3 @@ +spring: + kafka: + bootstrap-servers: ${spring.embedded.kafka.brokers}