Kafka Type Headers未移除求助：如何清除类型信息并保留指定Header

1. 彻底关闭类型信息Header #

你之前的配置未生效，大概率是配置层级或转换器默认行为导致的，需要从两方面修正：

修正配置文件

在application.properties中针对输出绑定单独配置关闭类型头，替换原有的全局绑定器配置：

# 针对test-out-vne输出绑定，关闭类型信息头
spring.cloud.stream.bindings.test-out-vne.producer.add-type-info-headers=false

# 同时确保Kafka绑定器的生产者配置也关闭类型头
spring.cloud.stream.kafka.bindings.test-out-vne.producer.configuration.spring.json.add.type.headers=false

自定义消息转换器

ByteArrayJsonMessageConverter默认可能仍会添加类型头，需要自定义Bean明确关闭：

import org.springframework.cloud.stream.converter.ByteArrayJsonMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.converter.MessageConverter;

@Configuration
public class StreamConverterConfig {
    @Bean
    public MessageConverter byteArrayJsonMessageConverter() {
        ByteArrayJsonMessageConverter converter = new ByteArrayJsonMessageConverter();
        converter.setAddTypeInfoHeaders(false); // 强制关闭类型头添加
        return converter;
    }
}

2. 过滤保留指定Header #

要仅保留jmsId和UUID，可以选择以下两种方式：

方式一：通过配置文件过滤（推荐，无需代码修改）

利用Spring Cloud Stream的Header过滤功能，在application.properties中添加：

# 设置输出绑定的Header模式为raw，仅保留指定Header
spring.cloud.stream.bindings.test-out-vne.producer.header-mode=raw
spring.cloud.stream.bindings.test-out-vne.producer.allowed-headers=jmsId,UUID

方式二：在函数中手动处理Header

如果需要更灵活的控制，可在你的test函数中手动构建新消息，只保留目标Header：

import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import java.util.HashMap;
import java.util.Map;
import java.util.function.Function;

@Bean
public Function<Message<Object>, Message<Object>> test() {
    return inputMsg -> {
        Map<String, Object> filteredHeaders = new HashMap<>();
        // 仅保留需要的Header
        if (inputMsg.getHeaders().containsKey("jmsId")) {
            filteredHeaders.put("jmsId", inputMsg.getHeaders().get("jmsId"));
        }
        if (inputMsg.getHeaders().containsKey("UUID")) {
            filteredHeaders.put("UUID", inputMsg.getHeaders().get("UUID"));
        }
        // 构建新消息返回
        return MessageBuilder.createMessage(
                inputMsg.getPayload(),
                MessageBuilder.withHeaders(filteredHeaders).build().getHeaders()
        );
    };
}

验证 #

重启应用后，输出的消息Header将仅包含jmsId和UUID，且不再出现spring_json_header_types等多余类型信息。

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