apache-kafka - 发布无法反序列化为 DLT 主题的消息

Question

我不明白如何使用 spring kafka 将无法反序列化的消息写入 DLT 主题。

我根据 spring kafka docs 配置了消费者这对于在消息反序列化后发生的异常很有效。

但是当消息不可反序列化时，org.apache.kafka.common.errors.SerializationException在轮询消息时抛出。

随后，SeekToCurrentErrorHandler.handle(Exception thrownException, List<ConsumerRecord<?, ?>> records, ...)使用此异常调用，但记录列表为空，因此无法向 DLT 主题写入内容。

我如何将这些消息也写入 DLT 主题？

Answer 1

问题是异常是由 Kafka 客户端本身抛出的，因此 Spring 无法看到失败的实际记录。

这就是我们添加 ErrorHandlingDeserializer2 的原因可用于包装实际的解串器；失败被传递到监听器容器并作为 DeserializationException 重新抛出.

见 the documentation .

When a deserializer fails to deserialize a message, Spring has no way to handle the problem, because it occurs before the poll() returns. To solve this problem, version 2.2 introduced the ErrorHandlingDeserializer2. This deserializer delegates to a real deserializer (key or value). If the delegate fails to deserialize the record content, the ErrorHandlingDeserializer2 returns a null value and a DeserializationException in a header that contains the cause and the raw bytes. When you use a record-level MessageListener, if the ConsumerRecord contains a DeserializationException header for either the key or value, the container’s ErrorHandler is called with the failed ConsumerRecord. The record is not passed to the listener.

DeadLetterPublishingRecoverer具有检测异常并发布失败记录的逻辑。