【问题标题】：add record level custom latency metric in kafka streams在 kafka 流中添加记录级自定义延迟指标
【发布时间】：2020-08-04 12:39:22
【问题描述】：
我尝试向我的 kafka-streams 应用程序添加一个特定指标，该指标将测量延迟并报告给 jmx。
我在 scala 中使用StreamsDSL，因此使用ProcessorAPI 度量指标（我知道这是可能的）对我不起作用。
我想了解的基本内容是：
- 如何提取特定记录属性（即标题）以用作度量计算的一部分
- 如何将新指标添加到报告给 jmx 的指标中
谢谢！
【问题讨论】：
标签：
apache-kafka
apache-kafka-streams
jmx
【解决方案1】：
您将需要回退到处理器 API 以访问记录元数据（如标头）并注册自定义指标。
请注意，您可以混合搭配 DSL 和处理器 API，因此没有必要离开 DSL。相反，您可以通过KStream.process() 或KStream.transform() 插入自定义Processors 或Transformers（请注意，您可能希望使用transform() 的多个“兄弟姐妹”而不是transform()）。