Spring Kafka：如何通过KafkaAutoConfiguration配置RecordInterceptor实现集中日志？

你遇到的核心问题是：Spring Kafka的自动配置不会自动将Spring上下文中的RecordInterceptor Bean绑定到ConcurrentKafkaListenerContainerFactory，因此拦截器无法生效。以下是具体的配置方案：

1. 手动配置容器工厂绑定拦截器 #

创建配置类，借助ConcurrentKafkaListenerContainerFactoryConfigurer复用自动配置的基础属性，同时将自定义拦截器注入到容器工厂中：

@Configuration
public class KafkaListenerConfig {

    @Bean
    public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
            ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
            ConsumerFactory<Object, Object> kafkaConsumerFactory,
            LoggingRecordInterceptor loggingRecordInterceptor) {
        ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
        // 复用自动配置的consumer核心属性（如bootstrap-servers、group-id等）
        configurer.configure(factory, kafkaConsumerFactory);
        // 绑定自定义RecordInterceptor到容器工厂
        factory.setRecordInterceptor(loggingRecordInterceptor);
        return factory;
    }
}

2. 修正拦截器代码的潜在问题 #

你的拦截器中第一个intercept方法返回null，这会导致消息被直接丢弃（Spring Kafka会判定该消息无需进入后续处理流程）。若需保留正常消息处理，应返回原ConsumerRecord实例：

@Slf4j
@Component
public class LoggingRecordInterceptor implements RecordInterceptor<String, Object> {

    @Override
    public ConsumerRecord<String, Object> intercept(ConsumerRecord<String, Object> record) {
        log.info("消息到达，topic: {}, partition: {}, offset: {}", record.topic(), record.partition(), record.offset());
        return record; // 返回原消息，避免被丢弃
    }

    @Override
    public ConsumerRecord<String, Object> intercept(ConsumerRecord<String, Object> record, Consumer<?, ?> consumer) {
        log.info("消息到达（带Consumer），topic: {}, partition: {}, offset: {}", record.topic(), record.partition(), record.offset());
        return record;
    }

    @Override
    public void success(ConsumerRecord<String, Object> record, Consumer<?, ?> consumer) {
        log.info("消息处理成功，topic: {}, offset: {}", record.topic(), record.offset());
    }

    @Override
    public void failure(ConsumerRecord<String, Object> record, Exception exception, Consumer<?, ?> consumer) {
        log.error("消息处理失败，topic: {}, offset: {}, 异常信息: {}", record.topic(), record.offset(), exception.getMessage(), exception);
    }
}

3. 额外说明 #

Spring Boot 2.7+支持通过spring.kafka.consumer.interceptor.classes属性配置原生Kafka拦截器，但这属于Kafka原生的ConsumerInterceptor范畴，无法覆盖success、failure这类Spring Kafka特有的拦截场景，因此更推荐第一种容器工厂绑定的方案。

内容的提问来源于stack exchange，提问作者Lorenzo Panetta