如何将Kafka的ConsumerRecord<K,V>或Spring Cloud Stream元数据传入反序列化器？

Spring Cloud Stream集成支持通过Function<Message<T>>处理Kafka等MQ接收的消息，还能配置反序列化器将键值的字节数组转成应用可用格式，反序列化工作由Kafka客户端完成。底层基于常规Kafka客户端，直接和ConsumerRecord<K, V>交互，会把主题、分区、偏移量、接收时间戳等字段映射为Message的Header传给Function实现。

现在需要实现把ConsumerRecord或接收时间戳传入反序列化器，让反序列化时就能用这个值（比如作为消息创建时间）。当前临时方案是在监听器里读取kafka_receivedTimestamp头，但这会把反序列化逻辑泄露到消费处理端，希望优化。

现有代码示例：

package foo;
import java.nio.ByteBuffer;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.serialization.Deserializer;
import org.springframework.messaging.Message;
import java.util.function.Consumer;

record Container(long id, long createdAt) {}

class ContainerDeserializer implements Deserializer<Container> {
    public Container deserialize(String topic, Headers headers, byte[] data) {
        // 希望在这里读取kafka_receivedTimestamp
        return new Container(ByteBuffer.wrap(data).longValue(), 0L); 
    }
}

class ContainerConsumer implements Consumer<Message<Container>> {
    public void accept(Message<Container> message) {
        Container original = message.getPayload();
        Container enriched = new Container(original.id(), message.getHeaders().get("kafka_receivedTimestamp", Long.class));
        // 处理enriched对象
    }
}

现有YAML配置：

spring:
  cloud:
    stream:
      kafka:
        bindings:
          containerConsumer-in-0:
            consumer:
              configuration:
                spring.deserializer.value.delegate.class: foo.ContainerDeserializer

解决方案 #

方案一：自定义反序列化器包装类，直接获取ConsumerRecord #

通过自定义包装类扩展Kafka的反序列化器，直接拿到ConsumerRecord的完整信息并传递给实际反序列化器：

1. 实现包装类，重写方法注入时间戳到Headers

package foo;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.Deserializer;
import org.springframework.kafka.support.serializer.DelegatingDeserializer;
import java.nio.ByteBuffer;

public class ConsumerRecordAwareDeserializer<T> extends DelegatingDeserializer {

    public ConsumerRecordAwareDeserializer(Deserializer<T> delegate) {
        super(delegate);
    }

    @Override
    public Object deserialize(String topic, ConsumerRecord<?, ?> record) {
        // 获取接收时间戳并写入Headers
        long receivedTimestamp = record.timestamp();
        record.headers().add("kafka_receivedTimestamp", ByteBuffer.allocate(Long.BYTES).putLong(receivedTimestamp).array());
        return super.deserialize(topic, record);
    }
}

2. 修改原反序列化器，从Headers读取时间戳

class ContainerDeserializer implements Deserializer<Container> {
    @Override
    public Container deserialize(String topic, Headers headers, byte[] data) {
        long createdAt = 0L;
        // 从Headers中读取传递的时间戳
        var timestampHeader = headers.lastHeader("kafka_receivedTimestamp");
        if (timestampHeader != null) {
            createdAt = ByteBuffer.wrap(timestampHeader.value()).getLong();
        }
        long id = ByteBuffer.wrap(data).longValue();
        return new Container(id, createdAt);
    }
}

3. 修改YAML配置，替换为自定义包装类

spring:
  cloud:
    stream:
      kafka:
        bindings:
          containerConsumer-in-0:
            consumer:
              configuration:
                value.deserializer: foo.ConsumerRecordAwareDeserializer
                spring.deserializer.value.delegate.class: foo.ContainerDeserializer

方案二：扩展消息转换器，统一注入时间戳 #

通过Spring Cloud Stream的消息转换器扩展，在消息转换前把时间戳注入到Kafka Headers：

1. 自定义消息转换器

package foo;

import org.springframework.kafka.support.converter.MessagingMessageConverter;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import java.nio.ByteBuffer;

public class TimestampInjectingMessageConverter extends MessagingMessageConverter {

    @Override
    protected org.springframework.messaging.Message<?> convert(ConsumerRecord<?, ?> record, String topic, Integer partition) {
        // 注入接收时间戳到Headers
        record.headers().add("kafka_receivedTimestamp", ByteBuffer.allocate(Long.BYTES).putLong(record.timestamp()).array());
        return super.convert(record, topic, partition);
    }
}

2. 注册自定义转换器到Spring容器

package foo;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.support.converter.ConsumerRecordMessageConverter;

@Configuration
public class KafkaConfig {

    @Bean
    public ConsumerRecordMessageConverter timestampInjectingMessageConverter() {
        return new TimestampInjectingMessageConverter();
    }
}

3. 保持ContainerDeserializer的修改（读取Headers时间戳），原有YAML配置无需调整。

注意事项 #

• 方案一直接操作ConsumerRecord，能拿到所有字段，适合需要自定义处理ConsumerRecord的场景；
• 方案二通过消息转换器统一处理，更贴合Spring Cloud Stream的扩展逻辑，适合全局统一注入字段的需求；
• 两种方案都把时间戳注入逻辑封装在反序列化环节，避免了消费端的逻辑泄露。

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