Spring Boot Kafka消费者Prometheus指标注册失败问题排查

问题背景 #

Spring Boot 2.7.5应用中包含query-routing和query-tracking两个Kafka消费者，均通过@KafkaListener配置，各自使用独立的ConcurrentKafkaListenerContainerFactory实例，并都添加了MicrometerConsumerListener用于指标采集。

应用启动初期，访问actuator/prometheus可以看到两个消费者的kafka.consumer.fetch.manager.records.consumed.total指标；但几秒后，query-tracking的该指标被移除，日志提示同名指标标签键不一致（新指标多了topic标签），触发点为io.micrometer.core.instrument.binder.kafka.KafkaMetrics#checkAndBindMetrics。

使用依赖版本：micrometer-registry-prometheus1.9.5、spring-kafka（适配Spring Boot 2.7.5版本）。

原因分析 #

1. 初始指标注册：通过MicrometerConsumerListener注册的kafka.consumer.fetch.manager.records.consumed.total指标，默认只携带client_id、group_id等标签，不包含topic标签。
2. 后续指标冲突：Kafka客户端的KafkaMetrics（Micrometer的Kafka绑定器）会定期执行checkAndBindMetrics方法，从Kafka原生MBean中采集指标，此时该指标会携带topic标签。
3. Micrometer指标规则：Micrometer要求同名的指标必须拥有完全一致的标签键集合，否则会拒绝冲突指标的注册，并移除已存在的冲突指标。query-tracking消费者因批量监听+手动ACK的配置，更早触发了带topic标签的指标采集，导致与初始注册的无topic标签指标冲突，最终被移除。

修复方案 #

方案1：统一初始指标的标签集合（推荐，保留topic维度监控） #

修改两个消费者配置类中MicrometerConsumerListener的初始化逻辑，显式指定包含topic标签，让初始注册的指标与后续KafkaMetrics采集的指标标签键一致：

// 在QueryRoutingConfiguration和QueryTrackingConfiguration中修改
consumerFactory.addListener(new MicrometerConsumerListener<>(meterRegistry,
    Collections.singletonList(KafkaConsumerMetricName.TOPIC)));

这样初始注册的指标会携带topic标签，后续KafkaMetrics注册的同名指标标签键完全匹配，不会产生冲突。

方案2：禁用Kafka客户端原生指标自动绑定（简单，丢失topic维度） #

通过配置关闭Micrometer的Kafka绑定器自动采集，只保留MicrometerConsumerListener的指标：

在application.yml或application.properties中添加：

management.metrics.binders.kafka.enabled: false

此方法会阻止KafkaMetrics#checkAndBindMetrics的执行，避免产生带topic标签的冲突指标，但会丢失按topic维度拆分的监控数据。

方案3：自定义MeterFilter兼容标签差异（不推荐） #

修改MeterRegistryConfigurator中的MeterFilter，强制统一冲突指标的标签键集合，但此方法会导致指标维度不准确，仅作为临时应急方案：

@Bean
MeterRegistryCustomizer<MeterRegistry> meterRegistryCustomizer() {
    return registry -> registry.config()
        // 保留原有配置...
        .meterFilter(new MeterFilter() {
            @Override
            public Meter.Id map(Meter.Id id) {
                String targetMetric = "dps.kafka.consumer.fetch.manager.records.consumed.total";
                if (id.getName().equals(targetMetric)) {
                    // 确保所有实例都包含topic标签，不存在则补充默认值
                    boolean hasTopicTag = id.getTags().stream().anyMatch(tag -> tag.getKey().equals("topic"));
                    if (!hasTopicTag) {
                        return id.withTags(Tag.of("topic", "unidentified"));
                    }
                }
                return id;
            }
        });
}

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