LocalCluster提交storm KafkaSpout重复

问题描述:

通过LocalCluster提交storm KafkaSpout的应用,每次重新提交应用时,消息都会重复消费,而且在zookeeper中不能创建存储节点.

日志中一直出现KeeperErrorCode = NoNode for 节点path.

从网上查看:

1. 大多都是bolt需要继承BaseBasicBolt,也就是说可能没有ack.

2.SpoutConfig构造中id是要唯一的.

但是我bolt继承的就是BaseBasicBolt.而且我设置的id也是不没有改变的.

解决方法:

我使用的是BrokerHosts brokerHosts = new ZkHosts(zkConnect);

SpoutConfig spoutConfig = new SpoutConfig(brokerHosts, topic, zkRoot,id);

首先storm kafka 的offset在zookeeper中保存的路径方式为：${zkRoot}/${id}/${partitionId},但是在zookeeper中不能创建此路经,这就是导致没有记录下offset的进度,重复消息的根本原因.

但是zk相关的信息在BrokerHosts已经设置完成,为何找不到呢?

看上面SpoutConfig中的两个变量,就是问题的所在了,还需要设置这两个成员变量,赋值之后,问题就得到了解决.

总结:storm kafka 提供的KafkaSpout组件是将offset维护在上面设置的zk路径下.