Spring Cloud Stream Kafka消费者timestamp header值被替换问题求助

问题根源 #

Spring Cloud Stream Kafka绑定器默认会将Kafka Broker分配的系统时间戳自动映射到Spring Message对象的timestamp头字段，这会直接覆盖你通过KafkaTemplate添加的自定义timestamp头。而@KafkaListener直接操作原生ConsumerRecord，不会触发这种自动映射逻辑，所以能正确获取自定义值。

解决方案（保留原timestamp键） #

以下两种方案均可实现保留自定义timestamp头的需求：

方案1：消费者端禁用自动头映射

在Spring Cloud Stream消费者的配置文件中，给目标绑定设置header-mode=raw，让绑定器完全原生传递Kafka Headers，不做任何自动转换：

properties配置：

# 替换<your-consumer-binding-name>为实际的消费者绑定名（比如input）
spring.cloud.stream.kafka.bindings.<your-consumer-binding-name>.consumer.header-mode=raw

yaml配置：

spring:
  cloud:
    stream:
      kafka:
        bindings:
          <your-consumer-binding-name>:
            consumer:
              header-mode: raw

配置生效后，Message的Headers会完全保留生产者传入的自定义timestamp头，不会被Broker的系统时间戳覆盖。

方案2：自定义头映射规则

如果不想全局禁用头转换，可以创建自定义的KafkaHeaderMapper Bean，禁用系统时间戳的自动映射：

import org.springframework.cloud.stream.binder.kafka.DefaultKafkaHeaderMapper;
import org.springframework.cloud.stream.binder.kafka.KafkaHeaderMapper;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class KafkaHeaderConfig {
    @Bean
    public KafkaHeaderMapper kafkaHeaderMapper() {
        DefaultKafkaHeaderMapper mapper = new DefaultKafkaHeaderMapper();
        // 禁用Broker时间戳到Message.timestamp的自动映射
        mapper.setMapTimestamp(false);
        // 确保字符串类型的头被正确映射
        mapper.setMapAllStringsOut(true);
        return mapper;
    }
}

验证生产者端头添加正确性 #

确保你在生产者端是将自定义时间戳添加到Kafka原生Headers中，示例代码：

// 方式1：通过ProducerRecord添加
ProducerRecord<String, String> record = new ProducerRecord<>("target-topic", "message-key", "message-content");
// 注意头值要转为字节数组
record.headers().add("timestamp", String.valueOf(System.currentTimeMillis()).getBytes());
kafkaTemplate.send(record);

// 方式2：通过KafkaTemplate重载方法直接传入头
Map<String, Object> headers = new HashMap<>();
headers.put("timestamp", System.currentTimeMillis());
kafkaTemplate.send("target-topic", "message-key", null, "message-content", headers);

消费者端获取自定义头示例 #

@Bean
public Consumer<Message<String>> customConsumer() {
    return message -> {
        // 从Headers中获取自定义timestamp（注意类型匹配，生产者传的是字节/字符串就对应解析）
        String timestampStr = message.getHeaders().get("timestamp", String.class);
        Long customTimestamp = Long.parseLong(timestampStr);
        // 业务逻辑处理
        System.out.println("自定义时间戳：" + customTimestamp);
    };
}

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