Conditional (content-based) routing on Kafka headers in Spring Cloud Stream

Question

I'm using Spring Cloud Stream 3.x in Spring Boot 2.x application to consume messages from a Kafka topic.

I want to have a listener that consumes messages conditionally on some custom header value, as per doc:

@StreamListener(value = "someTopic", condition = "headers['SomeHeader']=='SomeHeaderValue'")
public void onMessage(Message<?> message) {
  LOGGER.info("Received: {}", message);
}

However listener never gets notified, and if condition is removed I see the following in the log:

Received: ... SomeHeader: [B@1055e4af ...

It turns out that custom headers are left in Kafka byte array raw format, making them not eligible for condition evaluation.

Is some additional configuration needed or am I missing something?

score 2 · Accepted Answer · answered Dec 17 '20 at 08:19

After some digging in sources and stackoveflow I have found the following:

Spring Cloud Stream delegates to Spring Kafka message and headers conversion (KafkaMessageChannelBinder ~ getHeaderMapper)
Headers are left in raw format by default headers conversion implementation (BinderHeaderMapper)
Spring Cloud Stream allows customization of headers mapping and particularly conversion of headers from byte array to String (How can I map incoming headers as String instead of byte[] in my Spring Cloud Stream project?)

So I added my custom header mapper bean (bean name is important, it allows to omit additional configuration property), which maps my custom header to String:

@Bean
public KafkaHeaderMapper kafkaBinderHeaderMapper() {
    SimpleKafkaHeaderMapper headerMapper = new SimpleKafkaHeaderMapper();
    headerMapper.setRawMappedHeaders(Map.of(
        "SomeHeader", true
    ));
    return headerMapper;
}

That fixed the problem:

Received: ... SomeHeader: SomeHeaderValue ...

P.S. It seems like a bug in Spring Cloud Stream:

It introduces its own implementation of header mapper (BinderHeaderMapper), but the latter doesn't respect conditional routing feature.
Header mapper is subclassed in KafkaMessageChannelBinder, this added behaviour is non-obvious and will be lost if custom header mapper is provided.

You should "accept" your own answer. You could also use `condition = "new String(headers['SomeHeader'])=='SomeHeaderValue'"` — Gary Russell, Dec 17 '20 at 14:48
Yep, I can accept it in two days :) As for `new String` in condition -- usable and shortest solution, but IMHO a bit ugly. — Oleg Efimov, Dec 17 '20 at 18:10
I don't remember why we had to reintroduce that mapper but `setRawMappedHeaders` is on the `AbstractKafkaHeaderMapper` so it's available on the `BinderHeaderMapper` too. — Gary Russell, Dec 17 '20 at 18:42

Conditional (content-based) routing on Kafka headers in Spring Cloud Stream

1 Answers1