我正在Spring Boot 2. x应用程序中使用Spring Cloud Stream 3.x来消费来自Kafka主题的消息。
我想有一个侦听器，它有条件地在一些自定义标头值上使用消息，如文档所示：
@StreamListener(value = "someTopic", condition = "headers['SomeHeader']=='SomeHeaderValue'")
public void onMessage(Message<?> message) {
LOGGER.info("Received: {}", message);
}
但是侦听器永远不会收到通知，如果条件被删除，我在日志中会看到以下内容：
Received: ... SomeHeader: [B@1055e4af ...
事实证明，自定义标头保留为Kafka字节数组原始格式，使它们没有资格进行条件评估。
是否需要一些额外的配置，或者我遗漏了什么？
经过对源和stackoveflow的一些挖掘，我发现了以下内容：
所以我添加了我的自定义标头映射器bean（bean名称很重要，它允许省略额外的配置属性），它将我的自定义标头映射到String：
@Bean
public KafkaHeaderMapper kafkaBinderHeaderMapper() {
SimpleKafkaHeaderMapper headerMapper = new SimpleKafkaHeaderMapper();
headerMapper.setRawMappedHeaders(Map.of(
"SomeHeader", true
));
return headerMapper;
}
这解决了问题：
Received: ... SomeHeader: SomeHeaderValue ...
附言：这似乎是Spring Cloud Stream的bug：
