Kafka - Intercept incoming messages

[LeonardoFloriano]

Is there any way to intercept incoming messages and extract the headers before being consumed ?
Similar to using a consumer interceptor in spring boot.

We need to change tenants before persisting any data.

[ozangunalp]

You can use the standard Kafka client interceptor ConsumerInterceptor

You can configure your implementation class using mp.messaging.incoming.$channel.interceptor.classes. It'll be instantiated by the Kafka client.

If that doesn't suit you, you can provide a decorator, in which you can modify the reactive stream per channel :
PublisherDecorator

More info on decorators: http://smallrye.io/smallrye-reactive-messaging/3.22.1/concepts/decorators/

[ozangunalp]

Closed fully by the IncomingInterceptors in #2244