使用Spring Cloud Stream时，如何为Kafka Consumer添加TYPE_ID头？

在Spring Cloud Stream中处理Kafka消息时，若消息缺少TYPE_ID头导致反序列化失败，可通过以下几种方式在消费前为消息添加类型标识头：

方式一：Spring Cloud Stream全局通道拦截器（ChannelInterceptor） #

通过全局通道拦截器，在消息进入消费者处理逻辑前修改消息头，适合Spring Cloud Stream生态内的统一处理。

import org.springframework.cloud.stream.config.GlobalChannelInterceptor;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

@Component
@GlobalChannelInterceptor(patterns = "your-input-channel") // 替换为你的输入通道名称
public class TypeIdHeaderInterceptor implements ChannelInterceptor {

    @Override
    public Message<?> postReceive(Message<?> message, MessageChannel channel) {
        if (message == null || message.getHeaders().containsKey("__TypeId__")) {
            return message;
        }
        // 为消息添加TYPE_ID头，替换为你的目标Payload类全限定名
        return MessageBuilder.fromMessage(message)
                .setHeader("__TypeId__", com.yourpackage.YourPayload.class.getName())
                .build();
    }
}

注意：若使用Spring Cloud Stream默认的JsonMessageConverter，需添加spring_json_header_types头，格式为JSON字符串，示例：

.setHeader("spring_json_header_types", "{\"__TypeId__\":\"com.yourpackage.YourPayload\"}")

方式二：自定义消息转换器（MessageConverter） #

自定义MessageConverter，在反序列化流程中主动补充TYPE_ID头，适合需要深度定制消息转换逻辑的场景。

import org.springframework.cloud.stream.converter.AbstractJsonMessageConverter;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

@Component
public class TypeIdAwareJsonConverter extends AbstractJsonMessageConverter {

    public TypeIdAwareJsonConverter() {
        super();
    }

    @Override
    protected Object convertFromInternal(Message<?> message, Class<?> targetClass, Object conversionHint) {
        // 仅当缺少TYPE_ID头时添加
        Message<?> modifiedMessage = message.getHeaders().containsKey("__TypeId__")
                ? message
                : MessageBuilder.fromMessage(message)
                        .setHeader("__TypeId__", targetClass.getName())
                        .build();
        return super.convertFromInternal(modifiedMessage, targetClass, conversionHint);
    }
}

方式三：Kafka客户端消费者拦截器（ConsumerInterceptor） #

在Kafka客户端层面拦截消费记录，添加TYPE_ID头，适合需要跨Spring Cloud Stream生态、统一处理所有Kafka消费消息的场景。

1. 实现拦截器 #

import org.apache.kafka.clients.consumer.ConsumerInterceptor;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.springframework.stereotype.Component;

import java.util.Map;

@Component
public class TypeIdConsumerInterceptor implements ConsumerInterceptor<String, Object> {

    @Override
    public ConsumerRecords<String, Object> onConsume(ConsumerRecords<String, Object> records) {
        for (TopicPartition partition : records.partitions()) {
            for (ConsumerRecord<String, Object> record : records.records(partition)) {
                if (record.headers().lastHeader("__TypeId__") == null) {
                    // 添加TYPE_ID头，替换为目标Payload类全限定名
                    record.headers().add("__TypeId__", com.yourpackage.YourPayload.class.getName().getBytes());
                }
            }
        }
        return records;
    }

    @Override
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {}
    @Override
    public void close() {}
    @Override
    public void configure(Map<String, ?> configs) {}
}

2. 配置拦截器 #

在application.yml中指定拦截器类：

spring:
  cloud:
    stream:
      kafka:
        binder:
          configuration:
            interceptor.classes: com.yourpackage.TypeIdConsumerInterceptor

注意事项 #

• TYPE_ID头名称：若使用Kafka官方JsonDeserializer，需添加__TypeId__头；若使用Spring Cloud Stream默认的JsonMessageConverter，需对应使用spring_json_header_types头。
• 反序列化配置：使用JsonDeserializer时，需在配置中开启use-native-decoding: true，并设置spring.json.trusted.packages: "*"以允许反序列化指定类。

内容的提问来源于stack exchange，提问作者sreekesh.s