批量监听器中无法处理反序列化异常

Question

我正在将 Spring boot 从版本 2 迁移到版本 3。

在我之前的实现中，我基于 v2.8.4 文档，一切工作正常。

目前，ListenerUtils.byteArrayToDeserializationException 已被删除，我必须使用 SerializationUtils.byteArrayToDeserializationException。

为此，我基于 v3.1.x 文档，但我遇到了问题。

文档中给出的示例，正是这部分，无法编译：

DeserializationException deserEx = SerializationUtils.byteArrayToDeserializationException(this.logger,
                    headers.get(i).get(SerializationUtils.VALUE_DESERIALIZER_EXCEPTION_HEADER));

理论上传递的是一个byte[]，但该方法只接受一个Header。所以我们需要一个演员阵容。使用 ListenerUtils，我将其转换为 byte[]。但现在，我转换为 Header 或 RecordHeader，出现以下错误 -> 忽略外部反序列化异常标头；可能的攻击？类似于以下内容：

DeserializationException exception = SerializationUtils.byteArrayToDeserializationException(
            new LogAccessor(IndexingService.class),
            new RecordHeader(SerializationUtils.KEY_DESERIALIZER_EXCEPTION_HEADER,
                (byte[]) headersOfNullItem.get(SerializationUtils.VALUE_DESERIALIZER_EXCEPTION_HEADER)));

我尝试创建一个 DeserializationExceptionHeader，但它是受包保护的。

我错过了什么吗？如何避免该问题以便处理反序列化中的错误？

Answer 1

考虑添加

org.apache.kafka.common.header.Headers

作为您的

@KafkaListener

方法参数，并从那里获取

SerializationUtils.VALUE_DESERIALIZER_EXCEPTION_HEADER

。

如果没问题，那么我们确实必须修复该文档。

当我们映射到

byte[]

时，我们可能也不能从 Kafka 标头中提取

MessageHeaders

- 而只是将其保留为

DeserializationExceptionHeader

的实例。所以，

headers.get(i).get(SerializationUtils.VALUE_DESERIALIZER_EXCEPTION_HEADER))

仍然有效。