Kafka消费者抛出异常时如何将自定义Header传入死信队列(DLQ)

我来帮你搞定这个问题——你遇到的核心问题是默认的死信处理逻辑并不会自动复用你修改后的Spring Message头，而是基于原始的Kafka ConsumerRecord来构建死信消息，再添加那些内置的异常相关Header，所以你手动设置的x-ecode没被传过去。结合你的场景，给你几个可行的解决方案：

1. 先修正配置的拼写错误！ #

你提到配置了spring.cloud.stream.kafka.binder.header=x-ecode，这里有个容易忽略的小错误：正确的配置项是复数的headers，而非单数header。修改后的配置应该是：

spring:
  cloud:
    stream:
      kafka:
        binder:
          headers: x-ecode # 注意是headers，不是header

这个配置用来告诉Kafka绑定器要传递哪些自定义Header，但仅靠这个还不够，还需要配合下面的逻辑调整。

2. 自定义DeadLetterPublishingRecoverer（最推荐方案） #

默认的死信发布器只会复制有限的Header，我们可以自定义一个DeadLetterPublishingRecoverer的Bean，完全控制死信消息的Header构建，把需要的自定义Header加进去。

示例代码如下：

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.ConsumerAwareListenerErrorHandler;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.messaging.Message;
import org.springframework.messaging.converter.MessagingMessageConverter;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.kafka.support.Bytes;

import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;

@Bean
public ConsumerAwareListenerErrorHandler customDlqErrorHandler(KafkaTemplate<Object, Object> kafkaTemplate) {
    // 定义死信主题和分区的映射逻辑（这里沿用原始消息的分区，可按需调整）
    DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(kafkaTemplate,
            (record, exception) -> new TopicPartition("你的死信主题名称", record.partition()));

    // 自定义Header处理逻辑，复制需要的自定义Header
    recoverer.setHeaderFunction((record, exception) -> {
        Map<String, Object> dlqHeaders = new HashMap<>();

        // 从原始ConsumerRecord中复制自定义Header（比如x-ecode）
        for (Header header : record.headers()) {
            if ("x-ecode".equals(header.key())) {
                dlqHeaders.put(header.key(), Bytes.wrap(header.value()));
            }
        }

        // 保留内置的异常相关Header（默认会自动添加，也可手动控制）
        dlqHeaders.put("x-exception-message", Bytes.wrap(exception.getMessage().getBytes()));
        dlqHeaders.put("x-original-partition", Bytes.of(ByteBuffer.allocate(4).putInt(record.partition()).array()));

        return dlqHeaders;
    });

    // 将错误处理器绑定到监听逻辑
    return (message, exception) -> {
        ConsumerRecord<?, ?> consumerRecord = ((MessagingMessageConverter) message.getHeaders().get(KafkaHeaders.CONSUMER_RECORD_CONVERTER))
                .getRecord(message);
        recoverer.accept(consumerRecord, exception);
        return null;
    };
}

这个方案的优势是完全可控，你可以根据需求添加任意自定义Header到死信消息中。

3. 修改原始ConsumerRecord的Header（另一种思路） #

死信处理器是基于原始的ConsumerRecord构建死信消息的，所以如果你能把自定义Header添加到原始的ConsumerRecord中，而不是只修改Spring的Message对象，死信处理器就能自动带上这个Header。

示例代码调整如下：

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.support.MessageBuilder;

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;

// 从接收的Message中获取原始ConsumerRecord
ConsumerRecord<?, ?> originalRecord = consumedMessage.getHeaders().get(KafkaHeaders.RECEIVED_RECORD, ConsumerRecord.class);

// 构建新的Header列表，添加自定义x-ecode
List<Header> newHeaders = new ArrayList<>(Arrays.asList(originalRecord.headers().toArray()));
newHeaders.add(new RecordHeader("x-ecode", ByteBuffer.allocate(4).putInt(100).array()));

// 创建修改后的ConsumerRecord
ConsumerRecord<?, ?> modifiedRecord = new ConsumerRecord<>(
        originalRecord.topic(),
        originalRecord.partition(),
        originalRecord.offset(),
        originalRecord.timestamp(),
        originalRecord.timestampType(),
        originalRecord.serializedKeySize(),
        originalRecord.serializedValueSize(),
        originalRecord.key(),
        originalRecord.value(),
        newHeaders,
        originalRecord.checksum()
);

// 将修改后的ConsumerRecord放回Message中
modifiedMessage = MessageBuilder.fromMessage(consumedMessage)
        .setHeader(KafkaHeaders.RECEIVED_RECORD, modifiedRecord)
        .setHeader("x-ecode", 100)
        .setHeader(BinderHeaders.PARTITION_OVERRIDE, consumedMessage.getHeaders().get(KafkaHeaders.RECEIVED_PARTITION_ID))
        .build();

// 抛出异常触发死信逻辑
throw new RuntimeException(modifiedMessage.toString());

最后检查点 #

确保你的消费者绑定配置开启了Header模式：

spring:
  cloud:
    stream:
      kafka:
        bindings:
          input: # 你的消费者绑定名称
            consumer:
              header-mode: headers # 必须是headers，不能是none或raw
              dead-letter-topic: 你的死信主题名称

内容的提问来源于stack exchange，提问作者Rajasekar P