如何为spring-kafka创建的所有MessageListenerContainers设置RecordInterceptor

import org.apache.kafka.clients.consumer.ConsumerInterceptor;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.listener.MessageListenerContainer;
import org.springframework.kafka.listener.RecordInterceptor;
import org.springframework.stereotype.Component;

import java.util.Map;

@Component
public class CustomKafkaListenerConfigurer implements KafkaListenerConfigurer {

    private final KafkaListenerEndpointRegistry endpointRegistry;

    public CustomKafkaListenerConfigurer(KafkaListenerEndpointRegistry endpointRegistry) {
        this.endpointRegistry = endpointRegistry;
    }

    @Override
    public void configureKafkaListeners(KafkaListenerEndpointRegistrar registrar) {
        Map<String, MessageListenerContainer> containers = endpointRegistry.getListenerContainers();
        for (MessageListenerContainer container : containers.values()) {
            container.setRecordInterceptor(new CustomRecordInterceptor());
        }
    }

    private static class CustomRecordInterceptor<K, V> implements RecordInterceptor<K, V> {

        @Override
        public ConsumerRecords<K, V> onConsume(ConsumerRecords<K, V> records) {
            // 在消息被消费之前的处理逻辑
            // 可以对消息进行修改或者添加额外的处理逻辑
            return records;
        }

        @Override
        public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {
            // 在消息被提交之前的处理逻辑
            // 可以添加额外的提交逻辑
        }
    }
}