Apache Storm Trident中Kafka分区与偏移量存储及消费滞后查询

刚好对Trident+Kafka的这套机制比较熟悉，来给你拆解下这两个问题：

你在ZooKeeper的/transactional/<StreamName>/coordinator/meta路径看到的内容，确实和Trident的事务状态挂钩，但Trident存储Kafka偏移量的逻辑得分成两种场景来看：

• 非事务性Trident拓扑：如果你的拓扑没启用事务，偏移量的存储位置很直观，就在Storm的ZooKeeper根节点下的/storm/<TopologyName>/kafka/offsets路径里。这里会按Kafka的topic、分区层级来存储对应消费偏移量，直接用ZooKeeper的get命令就能看到某个分区的已消费偏移量，一眼就能对应上分区。
• 事务性Trident拓扑：这时候你看到的/transactional/<StreamName>/coordinator/meta是事务协调器的元数据，里面的偏移量是和Trident的事务批次绑定的，不会直接暴露分区对应关系。实际上，事务性场景下，Kafka的分区偏移量会被封装在Trident的事务状态中，存储在你配置的状态后端里——如果用的是默认ZooKeeper状态后端，会在/transactional/<StreamName>/state/partition-<X>这类路径下，但这些数据是序列化后的格式，直接用ls或get看是乱码，因为Trident会把每个事务批次对应的所有分区偏移量打包存储起来。

查询消费滞后的方法分几种，看你的拓扑类型和需求来选：

• Storm UI + Kafka原生工具组合法：
  • 先打开Storm UI，找到你的Trident拓扑，查看Spout的统计面板，里面会有已处理消息数的大致统计。
  • 再用Kafka自带的消费者组工具，比如旧版本的kafka-consumer-groups.sh或者新版本的kafka-consumer-groups，执行命令：kafka-consumer-groups.sh --bootstrap-server <你的KafkaBroker地址> --describe --group <你的Trident拓扑名>。这个命令会列出每个分区的最新偏移量、消费者已提交的偏移量，两者的差值就是消费滞后。注意：如果是事务性拓扑，这里显示的已提交偏移量是Trident最近完成的事务批次对应的偏移量，因为Trident只会在事务提交后才更新偏移量。
• 自定义监控埋点：
  • 可以在Trident的Spout或者自定义Bolt里，用Kafka的AdminClient API获取每个分区的最新偏移量，再和Trident当前消费的偏移量做对比。
  • 如果用的是Storm的状态后端，也可以通过Trident的StateQuery接口查询存储的偏移量，然后计算差值。比如加一个定时运行的监控Bolt，把计算出的滞后量输出到Prometheus这类监控系统里，方便实时查看。
• ZooKeeper直接查询（仅非事务场景）：
  • 对于非事务性拓扑，直接用ZooKeeper的get命令查看/storm/<TopologyName>/kafka/offsets/<TopicName>/<PartitionNumber>的内容，得到已消费偏移量；再用Kafka的kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list <Broker地址> --topic <TopicName> --time -1获取每个分区的最新偏移量，手动计算两者的差值就是消费滞后。

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