Unable to get offset lags for kafka. Reason: java.lang.NullPointerException at org.apache.storm.kafk

问题信息

Unable to get offset lags for kafka. Reason: java.lang.NullPointerException at org.apache.storm.kafka.monitor.KafkaOffsetLagUtil.getOffsetLags (KafkaOffsetLagUtil.java:272) at org.apache.storm.kafka.monitor.KafkaOffsetLagUtil.main (KafkaOffsetLagUtil.java:124)

我的情况

当数据量慢慢上来的时候，这个问题就会自动解决，网上有如下的解决方案，请参考

后知后觉（问题原因）

storm 消费 kafka 队列时，没有读到或者解析到 zk 中存储的 kafka 的偏移量 offset，可能原因：队列刚刚创建，storm 在 zk 中还未存放 offset 信息，手动在 kafka 命令行中生产几条消息（bin/kafka-console-producer.sh），即可帮助 storm 完成在 zk 中存储 offset 信息。

问题描述:

storm 起动 topology 的在 stormui 中可以看到类似错误

storm 的 KafkaSpout 接收 kafka 消息时报错，错误如下

Unable to get offset lags for kafka. Reason: java.lang.IllegalArgumentException: zk-node ‘/kafka_storm/webTraceId’ dose not exists. at org.apache.storm.kafka.monitor.KafkaOffsetLagUtil.getOldConsumerOffsetsFromZk (KafkaOffsetLagUtil.java:387) at org.apache.storm.kafka.monitor.KafkaOffsetLagUtil.getOffsetLags (KafkaOffsetLagUtil.java:268) at org.apache.storm.kafka.monitor.KafkaOffsetLagUtil.main (KafkaOffsetLagUtil.java:124)

或者如下

Unable to get offset lags for kafka. Reason: java.lang.NullPointerException at org.apache.storm.kafka.monitor.KafkaOffsetLagUtil.getOffsetLags (KafkaOffsetLagUtil.java:272) at org.apache.storm.kafka.monitor.KafkaOffsetLagUtil.main (KafkaOffsetLagUtil.java:124)

原因:

storm 读取自身存储的 kafka 信息时，没读到或者解析不正确。这个消息存储在 zookeeper 中。在 org.apache.storm.kafka.SpoutConfig 类中的 public SpoutConfig (BrokerHosts hosts, String topic, String zkRoot, String id) 中的 zkRoot 和 id, storm 会在 zookeepr 的中创建如下目录 /zkRoot/id，在这个目录下面会存放一些 kafka 相关信息。大致如下:

{"topology":{"id":"alarmTopology-11-1528251428","name":"alarmTopology"},"offset":397,"partition":0,"broker":{"host":"172.16.125.134","port":9092},"topic":"alarm-fe"}

zookeeper 路径如下:

[zk: localhost:2181 (CONNECTED) 81] ls /kafka_storm/webTraceId
[partition_0, partition_1, partition_2]

解决办法:

分 2 步:

delete /kafka_storm/webTraceId/partition_0
delete /kafka_storm/webTraceId/partition_1
delete /kafka_storm/webTraceId/partition_2
delete /kafka_storm/webTraceId

往 kafka 中生产数据

等数据被 storm 的 kafkaSpout 消费之后，/kafka_storm/webTraceId/partition_0 这些目录会被重新建立

问题解决