求助：通过BeanPostProcessor添加Micrometer监听器后Kafka Producer指标缺失

背景 #

我需要为应用添加监控指标，统计Kafka消息的发送与消费数量，因此尝试通过自定义库的BeanPostProcessor，给所有ConsumerFactory和ProducerFactory自动添加MicrometerConsumerListener与MicrometerProducerListener。

现有配置代码 #

ProducerFactory配置 #

@Bean
public ProducerFactory<String, MyDto> producerFactory() {
    Map<String, Object> configProps = new HashMap<>();
    configProps.put(
            ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
            props.getBootstrapAddress());
    configProps.put(
            ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
            JsonSerializer.class);
    configProps.put(
            ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
            JsonSerializer.class);

    return new DefaultKafkaProducerFactory<>(configProps);
}

@Bean(name = "producerTestFactory")
public ProducerFactory<String, String> producerTestFactory() {
    Map<String, Object> configProps = new HashMap<>();
    configProps.put(
            ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
            props.getBootstrapAddress());
    configProps.put(
            ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
            StringSerializer.class);
    configProps.put(
            ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
            StringSerializer.class);

    return new DefaultKafkaProducerFactory<>(configProps);
}

ConsumerFactory配置 #

@Bean
public ConsumerFactory<String, MyDto> consumerFactory() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, prop.getBootstrapAddress());
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "1");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
    props.put(ErrorHandlingDeserializer.KEY_DESERIALIZER_CLASS, StringDeserializer.class);
    props.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, StringDeserializer.class);
    props.put(JsonDeserializer.TRUSTED_PACKAGES, "*");
    return new DefaultKafkaConsumerFactory<>(props, new StringDeserializer(),
            new ErrorHandlingDeserializer(new StringDeserializer()));
}

KafkaTemplate配置 #

@Bean
public KafkaTemplate<String, MyDto> kafkaTemplate() {
    return new KafkaTemplate<>(producerFactory());
}

@Bean(name = "testKafkaTemplate")
public KafkaTemplate<String, String> testKafkaTemplate(
        @Qualifier("producerTestFactory") ProducerFactory<String, String> testKafkaProducer
) {
    return new KafkaTemplate<>(testKafkaProducer);
}

自定义监控BeanPostProcessor #

@RequiredArgsConstructor
public class KafkaMonitoringPostProcessor implements BeanPostProcessor {

    private final MeterRegistry meterRegistry;

    @Override
    public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
        if (bean instanceof ConsumerFactory cf) {
            cf.addListener(new MicrometerConsumerListener(meterRegistry));
        } else if (bean instanceof ProducerFactory pf) {
            pf.addListener(new MicrometerProducerListener<>(meterRegistry));
        }
        return bean;
    }

}

问题现象 #

• Consumer的监控指标正常显示
• Producer的监控指标完全看不到
• 直接在producerFactory()方法内手动添加MicrometerProducerListener可以正常采集指标，但为了方案可扩展性，必须通过自定义库的BeanPostProcessor实现

补充说明 #

已通过调试确认：BeanPostProcessor工作正常，MicrometerProducerListener已成功添加到所有ProducerFactory实例中

内容的提问来源于stack exchange，提问作者Ytug Ilya