Kafka Streams - 从指标注册表访问数据

Question

我很难找到有关如何访问 Kafka Streams 指标注册表中数据的文档，我想我可能正在尝试在圆孔中安装方钉。我希望就以下方面得到一些建议：

目标

收集记录在 Kafka Streams metrics registry 中的指标并将这些值发送到任意端点

工作流程

这就是我认为需要完成的，并且我已经完成了除最后一步之外的所有步骤（因为指标注册表是私有的，所以最后一步有问题） .但我可能会以错误的方式解决这个问题：

• 定义一个实现MetricReporter接口的class。构建 Kafka 在 metricChange 方法中创建的指标列表（例如，每当调用此方法时，使用当前注册的指标更新 hashmap）。
• 在 metric.reporters 配置中指定此 class 属性
• 设置一个进程来轮询 Kafka Streams 指标注册表以获取当前数据，并将值发送到任意端点

无论如何，最后一步在 Kafka 0.10.0.1 中似乎是不可能的，因为指标注册表未公开。如果这是正确的工作流程（听起来不是..），或者我是否误解了提取 Kafka Streams 指标的过程，请告诉我一下？

Answer 1

尽管指标注册表未公开，您仍然可以通过其 KafkaMetric.value() / KafkaMetric.value(timestamp) 方法获取给定 KafkaMetric 的值。例如，正如您在 JMXRporter 中观察到的那样，它保留来自实例化 init() 和 metricChange/metricRemoval 方法的 KafkaMetrics 列表，然后在其 MBean 实现中，当 getAttribute 是被调用，就会调用对应的KafkaMetrics.value()函数。因此，对于您的自定义记者，您可以应用类似的模式，例如，定期轮询所有保留的 KafkaMetrics.value()，然后将结果通过管道传输到您的终点。

Answer 2

org.apache.kafka.common.metrics 中的 MetricReporter 接口已经使您能够管理报告器中的所有 Kafka 流指标。所以不需要kafka内部注册表。