如何将RecordInterceptor设置为ConcurrentKafkaListenerContainerFactory

Question

我使用的是Spring Kafka 2.2.7，我已经将@EnableKafka配置为kafkaListenerContainerFactory，并使用@KafkaListener消费消息，一切都按预期运行。

我想添加一个RecordInterceptor来记录所有消耗的消息，但发现很难配置它。留档指出可以在容器上设置RecordInterceptor，但我不确定如何获取容器的实例。

从版本2.2.7开始，您可以向侦听器容器添加RecordInterceptor；它将在调用允许检查或修改记录的侦听器之前被调用。

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

我查阅了Spring文档，但没有找到解决方案，这似乎是一件简单的事情，但也许我遗漏了一些东西。

在此方面的任何帮助都将不胜感激。

提前感谢。

Answer 1

由于2.2.7

factory.setRecordInterceptor(new RecordInterceptor);

另一个信息记录器拦截器无法用于批处理侦听器

从版本2.2.7开始，您可以将RecordInterceptor添加到侦听器容器中；它将在调用侦听器之前被调用，允许检查或修改记录。如果拦截器返回null，则不调用侦听器。当侦听器是批处理侦听器时，不会调用拦截器。