如何查看Storm消费者的Kafka主题偏移量及重启重放问题

我来帮你一步步理清这个问题：

为什么你的storm-kafka-group找不到？ #

你设置了ProcessingGuarantee.AT_MOST_ONCE，这是关键原因！在这种语义下，Storm-Kafka Client不会将偏移量提交到Kafka或ZooKeeper。

AT_MOST_ONCE的核心逻辑是：消息被发送到Bolt处理后，Storm直接认为处理完成，不会做任何偏移量的持久化操作——既不会往Kafka的消费者组偏移量存储里写数据，也不会再依赖ZooKeeper（毕竟新版本Storm已经默认用Kafka存储偏移量了）。所以不管是用zkCli.sh查ZK，还是用Kafka的消费者组命令，都找不到这个group的痕迹。

如何查看偏移量？ #

因为AT_MOST_ONCE模式下没有持久化的偏移量，你可以通过两种方式了解当前消费位置：

• 查看Storm UI指标：打开Storm的Web UI，找到你的拓扑，进入KafkaTridentSpoutOpaque组件的详情页，里面会有currentOffset或nextOffset这类指标，能实时看到每个Kafka分区的当前消费偏移量。
• 临时切换模式测试：把ProcessingGuarantee改成AT_LEAST_ONCE（或者EXACTLY_ONCE，如果你的拓扑支持），重启拓扑后，再用Kafka的消费者组描述命令查询：

  bin/kafka-consumer-groups.sh --describe --group storm-kafka-group --bootstrap-server localhost:9092
  

  这时候就能看到该消费者组的所有分区偏移量信息了，测试完成后再改回AT_MOST_ONCE即可。

重启拓扑后Kafka事件会被重放吗？ #

答案是：取决于Kafka消费者的auto.offset.reset配置，默认情况下不会重放旧消息。

因为AT_MOST_ONCE模式下没有提交过偏移量，每次拓扑重启时，Kafka Consumer会根据auto.offset.reset的默认值latest来决定消费起点：

• 如果是默认的latest：重启后会直接从每个Kafka分区的最新偏移量开始消费，之前未消费（或消费过但没提交偏移量）的旧消息不会被重放。
• 如果你手动设置了auto.offset.reset=earliest：重启后会从每个分区的起始位置开始消费，所有历史消息都会被重放。

你可以在创建KafkaSpoutConfig时通过setProp添加这个配置：

kafkaSpoutConfig = KafkaSpoutConfig.builder(brokerURL, kafkaTopic)
    .setProp(ConsumerConfig.GROUP_ID_CONFIG,"storm-kafka-group")
    .setProcessingGuarantee(ProcessingGuarantee.AT_MOST_ONCE)
    .setProp(ConsumerConfig.CLIENT_ID_CONFIG, InetAddress.getLocalHost().getHostName())
    .setProp(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest") // 手动设置偏移量重置策略

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