寻求Kafka Consumer中记录年龄（生产到消费时长）相关指标

你要的「记录从生产到消费的时间差」这个指标，Kafka原生并没有直接提供开箱即用的Consumer指标，但可以通过以下几种方式实现：

• 自定义埋点计算
  利用Kafka消息自带的CreateTime（生产者发送消息时的时间戳，默认启用），在消费者消费消息时，用当前时间减去消息的CreateTime得到时间差，然后将这个值上报到你的监控系统（比如Prometheus、Grafana）。
  示例代码片段（Java）：

  consumerRecords.forEach(record -> {
      long recordAge = System.currentTimeMillis() - record.timestamp();
      // 上报recordAge到监控，比如用Micrometer的timer或者gauge
  });
  

  注意：如果你的消息用的是LogAppendTime（Broker写入日志的时间戳），计算逻辑同理，只是时间基准换成Broker的写入时间。

• 流处理框架计算
  用ksqlDB或者Kafka Streams做流处理，直接在流中计算每条消息的消费年龄：

  • 在ksqlDB中，可以通过SELECT TIMESTAMP_DIFF(CURRENT_TIMESTAMP(), RECORD_TIMESTAMP(), MINUTES) AS record_age FROM your_topic EMIT CHANGES;这类语句计算，再将结果导出为指标。
  • Kafka Streams中可以通过ProcessorAPI或者KStream的转换操作计算时间差，然后通过Metrics接口暴露指标。

• 第三方监控扩展
  部分监控工具（比如Datadog、New Relic）支持通过自定义插件或者日志解析来提取这个指标，你可以在消费者应用中输出包含消息时间戳和消费时间的日志，让监控工具解析计算时间差并生成指标。

内容的提问来源于stack exchange，提问作者r-vanooyen