How to set RecordInterceptor to ConcurrentKafkaListenerContainerFactory

user_2905565 0
Viewed 2561

I'm using Spring Kafka 2.2.7, I have configured @EnableKafka with kafkaListenerContainerFactory and using @KafkaListener to consume messages, everything is working as expected.

I want to add a RecordInterceptor to log all the consumed messages but finding it difficult to configure it. The documentation states that RecordInterceptor can be set on a container, however I'm not sure how to obtain an instance of the container.

Starting with version 2.2.7, you can add a RecordInterceptor to the listener container; it will be invoked before calling the listener allowing inspection or modification of the record.

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Bytes> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, Bytes> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(createConsumerFactory());
        factory.setConcurrency(consumerCount);
        return factory;
    }

I've looked through the Spring documentation but haven't found a solution, it seems to be a simple thing but perhaps I've missed something.

Any help in this regard would be appreciated.

Thanks in advance.

spring-kafka apache-kafka java
1 Answers

There is a method setRecordInterceptor since 2.2.7

factory.setRecordInterceptor(new RecordInterceptor);

And another information RecordInterceptor will not work for batch listener

Starting with version 2.2.7, you can add a RecordInterceptor to the listener container; it will be invoked before calling the listener allowing inspection or modification of the record. If the interceptor returns null, the listener is not called. The interceptor is not invoked when the listener is a batch listener.

user_99591520
Related