Spring Cloud Stream Kafka的DLQ消息丢失原始字段问题咨询

在Spring Cloud Stream Kafka应用中，消费JSON格式的字符串消息时，我们通常会通过自定义反序列化器将消息映射为POJO。但触发DLQ（死信队列）机制时，默认行为是发送映射后的POJO到DLQ，而非原始的JSON字符串。这会导致DLQ中的消息丢失未被POJO映射的字段，后续修复POJO（新增字段映射）后，无法正常处理这些DLQ消息。

举个例子：

• 原始消息：

{
  "id": "abc",
  "name": "John",
  "age": 39
}

• 映射用的POJO：

class Person {
    private String id;
    private String name;
    // getter/setter
}

• DLQ中最终的失败消息：

{
  "id": "abc",
  "name": "John"
}

age字段因未被POJO映射而丢失。

方案1：直接消费原始字符串，手动反序列化 #

这种方式最直接，将输入通道的消息类型设为String，在业务代码中手动完成反序列化，出错时直接将原始字符串发送到DLQ。

实现步骤： #

1. 定义输入通道，接收String类型消息：

public interface MySink {
    String INPUT = "my-input";

    @Input(INPUT)
    SubscribableChannel input();
}

2. 消费方法中处理原始字符串，手动反序列化：

@Autowired
private MessageChannel dlqOutput;

@StreamListener(MySink.INPUT)
public void processRawMessage(String rawJson) {
    ObjectMapper objectMapper = new ObjectMapper();
    try {
        Person person = objectMapper.readValue(rawJson, Person.class);
        // 执行业务逻辑
    } catch (Exception e) {
        // 发送原始JSON到DLQ
        dlqOutput.send(MessageBuilder.withPayload(rawJson).build());
    }
}

3. 配置DLQ输出通道：

public interface MyDlqSink {
    String DLQ_OUTPUT = "my-dlq-output";

    @Output(DLQ_OUTPUT)
    MessageChannel dlqOutput();
}

方案2：自定义错误处理器，保留原始消息 #

如果不想修改消费逻辑，可以通过自定义错误处理机制，在反序列化阶段保留原始消息，并在出错时提取原始内容发送到DLQ。

实现步骤： #

1. 自定义反序列化器，将原始消息存入消息头：

public class RawMessagePreservingDeserializer extends JsonDeserializer<Person> {

    public RawMessagePreservingDeserializer() {
        super(Person.class);
    }

    @Override
    public Person deserialize(String topic, Headers headers, byte[] data) throws SerializationException {
        // 将原始字节数组转为字符串存入headers
        headers.add(new RecordHeader("rawPayload", data));
        return super.deserialize(topic, headers, data);
    }
}

2. 配置消费端使用自定义反序列化器：

spring:
  cloud:
    stream:
      kafka:
        bindings:
          my-input:
            consumer:
              configuration:
                value.deserializer: com.example.RawMessagePreservingDeserializer
      bindings:
        my-input:
          destination: input-topic
          group: my-group
        my-dlq-output:
          destination: dlq-topic

3. 编写错误处理器，从消息头中提取原始消息并发送到DLQ：

@ServiceActivator(inputChannel = "errorChannel")
public void handleError(ErrorMessage errorMessage) {
    Message<?> originalMsg = errorMessage.getOriginalMessage();
    if (originalMsg != null) {
        // 从headers获取原始消息
        Header rawPayloadHeader = originalMsg.getHeaders().get("rawPayload", Header.class);
        if (rawPayloadHeader != null) {
            String rawJson = new String(rawPayloadHeader.value());
            dlqOutput.send(MessageBuilder.withPayload(rawJson).build());
            return;
        }
    }
    // 兜底处理：发送错误信息到DLQ
    dlqOutput.send(MessageBuilder.withPayload(errorMessage.getPayload().toString()).build());
}

核心原理 #

默认情况下，Spring Cloud Stream会先完成消息的反序列化（转为POJO），再交给业务逻辑处理。当处理出错时，DLQ发送的是已经经过反序列化的对象，因此丢失了未被POJO映射的字段。上述方案的核心都是在反序列化阶段保留原始消息内容，或直接消费原始消息，确保DLQ中存储的是完整的原始输入。

内容的提问来源于stack exchange，提问作者Yosi Bronsberg