将无法反序列化的消息发布到 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。

当反序列化器无法反序列化消息时，Spring 无法处理该问题，因为它发生在 poll() 返回之前。为了解决这个问题，2.2 版本引入了 ErrorHandlingDeserializer2。这个反序列化器委托给一个真正的反序列化器（键或值）。如果委托未能反序列化记录内容，则 ErrorHandlingDeserializer2 在包含原因和原始字节的标头中返回空值和 DeserializationException。当您使用记录级 MessageListener 时，如果 ConsumerRecord 包含键或值的 DeserializationException 标头，则使用失败的 ConsumerRecord 调用容器的 ErrorHandler。记录不会传递给监听器。

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