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