Spring Cloud Stream kafka 指标未显示在执行器指标中 | Spring Boot 2.2.2 | Spring Cloud Hoxton.SR4

Posted 2023-02-27

【中文标题】Spring Cloud Stream kafka 指标未显示在执行器指标中 | Spring Boot 2.2.2 | Spring Cloud Hoxton.SR4

【英文标题】：Spring cloud stream kafka metrics are not shown in actuator metrics | Springboot 2.2.2 | SpringCloud Hoxton.SR4

【发布时间】：2020-10-26 04:44:20

【问题描述】：

我没有看到使用执行器 /actuator/metrics 的 kafka 消费者/生产者指标。仅显示基本的 jvm 指标。

"names": [
"jvm.buffer.memory.used",
"jvm.threads.states",
"jvm.memory.committed",
"spring.integration.channels",
"process.uptime",
"jvm.memory.used",
"jvm.gc.pause",
"logback.events",
"http.server.requests",
"jvm.memory.max",
"tomcat.sessions.active.current",
"jvm.buffer.total.capacity",
"system.cpu.usage",
"jvm.threads.live",
"jvm.classes.unloaded",
"jvm.classes.loaded",
"jvm.threads.peak",
"tomcat.sessions.rejected",
"tomcat.sessions.alive.max",
"jvm.gc.memory.promoted",
"jvm.buffer.count",
"jvm.gc.memory.allocated",
"tomcat.sessions.expired",
"tomcat.sessions.created",
"jvm.gc.max.data.size",
"system.cpu.count",
"spring.integration.handlers",
"spring.integration.sources",
"process.start.time",
"jvm.threads.daemon",
"tomcat.sessions.active.max",
"jvm.gc.live.data.size",
"process.cpu.usage"
]

我的应用程序运行良好，绑定看起来也不错，并且在 /actuator/bindings 中可见。

[

"bindingName": "send-kafka-out",
"name": "test.topic",
"group": null,
"pausable": false,
"state": "running",
"extendedInfo": 
"bindingDestination": "test.topic",
"ExtendedProducerProperties": 
"autoStartup": true,
"partitionCount": 1,
"headerMode": "headers",
"extension": 
"bufferSize": 16384,
"compressionType": "none",
"sync": false,
"sendTimeoutExpression": null,
"batchTimeout": 0,
"messageKeyExpression": null,
"headerPatterns": null,
"configuration": ,
"topic": 
"replicationFactor": null,
"replicasAssignments": ,
"properties": 
,
"useTopicHeader": false,
"recordMetadataChannel": null
,
"validPartitionSelectorProperty": true,
"validPartitionKeyProperty": true

,
"input": false

]

下面是我的流配置

public interface KafkaStreams 
    String OUTPUT = "send-kafka-out";

    @Output(OUTPUT)
    MessageChannel getOutboundMessageChannel();



@SpringBootApplication
@EnableDiscoveryClient
@EnableJms
@EnableBinding(KafkaStreams.class)
@ComponentScan("com.test.connector.*")
public class ConnectorApplication 

    public static void main(String[] args) 
        SpringApplication.run(ConnectorApplication.class, args);
    

下面是我的 config.yml

spring:
  jms:
    listener:
      auto-startup: true
      acknowledge-mode: AUTO
      max-concurrency: 10
      concurrency: 10
  cloud:
    stream:
      default-binder: kafka-default
      function:
        definition: 
      kafka:
        binder:
          brokers:
            - localhost:9092
          configuration:
            default.key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
            default.value.serde: org.apache.kafka.common.serialization.Serdes$BytesSerde
            commit.interval.ms: 1000
            #security:
            #  protocol: SASL_PLAINTEXT

      bindings:
        send-kafka-out:
          destination: test.topic
          contentType: text/plain
          group: output-group-1
          binder: kafka-default
          producer:
            header-mode: headers

      binders:
        kafka-default:
         type: kafka
         environment:
           spring:
             cloud:
               stream:
                kafka:
                  binder:
                    brokers: localhost:9092

我正在使用 SpringBoot 2.2.2.RELEASE、SpringCloud Hoxton.SR4 和 spring-cloud-stream-binder-kafka 3.0.4.RELEASE。关于我缺少什么的任何指示都会有所帮助。

【问题讨论】：

@Gary Russell，你能帮我解决这个问题吗？我有点卡在这里。甚至尝试了具有上述版本的新应用程序并且可以重现它。

【参考方案1】：

您需要有输入绑定。 Kafka binder 指标仅支持消费者。看起来您在代码中只有一个输出绑定。我刚刚尝试了一个包含消费者的快速应用程序，并且能够检索指标：

/actuator/metrics/spring.cloud.stream.binder.kafka.offset

"name":"spring.cloud.stream.binder.kafka.offset","description":"Unconsumed messages for a particular group and topic","baseUnit":null,"measurements":["statistic":"VALUE","value":0.0],"availableTags":["tag":"topic","values":["uppercase-in-0"],"tag":"group","values":["anonymous.06730d87-0871-4b82-bd82-ae4958dbe2a3"]]

【讨论】：

感谢@sobychacko 在这里的帮助。让我试一试。此外，我正在寻找使用执行器+千分尺的详细生产者/消费者指标。 github.com/micrometer-metrics/micrometer/pull/1835 中提到的东西。知道是否支持？

是的，您应该看到所有这些 kafka 生产者/消费者指标。如果没有，请随时在 binder repo 中提交一个新问题。