storm集成kafka报错【KeeperErrorCode = NoNode for /kafka/brokers/topics/test/partitions】

时间：2022-01-01 22:02:32

问题重现：

前置条件：使用的是CDH的KAFKA-0.8.2.0-1.kafka1.4.0.p0.56版本

最近在集成kafka-storm demo的时候，zookeeper总是报错。如下：

7382 [Thread-17-word-reader] INFO  b.s.d.executor - Opening spout word-reader:(6)
7382 [Thread-15-__acker] INFO  b.s.d.executor - Preparing bolt __acker:(2)
7388 [Thread-15-__acker] INFO  b.s.d.executor - Prepared bolt __acker:(2)
7399 [Thread-19-__system] INFO  b.s.d.executor - Preparing bolt __system:(-1)
7402 [Thread-19-__system] INFO  b.s.d.executor - Prepared bolt __system:(-1)
7404 [Thread-21-__system] INFO  b.s.d.executor - Preparing bolt __system:(-1)
7405 [Thread-21-__system] INFO  b.s.d.executor - Prepared bolt __system:(-1)
7411 [Thread-24-word-normalizer] INFO  b.s.d.executor - Preparing bolt word-normalizer:(5)
7411 [Thread-24-word-normalizer] INFO  b.s.d.executor - Prepared bolt word-normalizer:(5)
7448 [Thread-17-word-reader] INFO  o.a.c.f.i.CuratorFrameworkImpl - Starting
7459 [Thread-12-word-reader] INFO  b.s.d.executor - Opening spout word-reader:(8)
7460 [Thread-12-word-reader] INFO  o.a.c.f.i.CuratorFrameworkImpl - Starting
7461 [Thread-13-__acker] INFO  b.s.d.executor - Preparing bolt __acker:(3)
7461 [Thread-13-__acker] INFO  b.s.d.executor - Prepared bolt __acker:(3)
7473 [Thread-17-word-reader] INFO  o.a.c.f.i.CuratorFrameworkImpl - Starting
7473 [Thread-12-word-reader] INFO  o.a.c.f.i.CuratorFrameworkImpl - Starting
7475 [Thread-17-word-reader-EventThread] INFO  o.a.c.f.s.ConnectionStateManager - State change: CONNECTED
7475 [Thread-12-word-reader-EventThread] INFO  o.a.c.f.s.ConnectionStateManager - State change: CONNECTED
7475 [Thread-12-word-reader-EventThread] INFO  o.a.c.f.s.ConnectionStateManager - State change: CONNECTED
7476 [Thread-17-word-reader-EventThread] INFO  o.a.c.f.s.ConnectionStateManager - State change: CONNECTED
7500 [Thread-17-word-reader] ERROR b.s.util - Async loop died!
java.lang.RuntimeException: java.lang.RuntimeException: org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /kafka/brokers/topics/test/partitions
        at storm.kafka.DynamicBrokersReader.getBrokerInfo(DynamicBrokersReader.java:92) ~[storm-kafka-0.10.0.jar:0.10.0]
        at storm.kafka.trident.ZkBrokerReader.<init>(ZkBrokerReader.java:42) ~[storm-kafka-0.10.0.jar:0.10.0]
        at storm.kafka.KafkaUtils.makeBrokerReader(KafkaUtils.java:57) ~[storm-kafka-0.10.0.jar:0.10.0]
        at storm.kafka.KafkaSpout.open(KafkaSpout.java:87) ~[storm-kafka-0.10.0.jar:0.10.0]
        at backtype.storm.daemon.executor$fn__5624$fn__5639.invoke(executor.clj:564) ~[storm-core-0.10.0.jar:0.10.0]
        at backtype.storm.util$async_loop$fn__545.invoke(util.clj:477) [storm-core-0.10.0.jar:0.10.0]
        at clojure.lang.AFn.run(AFn.java:22) [clojure-1.6.0.jar:?]
        at java.lang.Thread.run(Thread.java:745) [?:1.7.0_79]
Caused by: java.lang.RuntimeException: org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /kafka/brokers/topics/test/partitions
        at storm.kafka.DynamicBrokersReader.getNumPartitions(DynamicBrokersReader.java:105) ~[storm-kafka-0.10.0.jar:0.10.0]
        at storm.kafka.DynamicBrokersReader.getBrokerInfo(DynamicBrokersReader.java:76) ~[storm-kafka-0.10.0.jar:0.10.0]
        ... 7 more
Caused by: org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /kafka/brokers/topics/test/partitions
        at org.apache.zookeeper.KeeperException.create(KeeperException.java:111) ~[zookeeper-3.4.5-cdh5.5.0.jar:3.4.5-cdh5.5.0--1]
        at org.apache.zookeeper.KeeperException.create(KeeperException.java:51) ~[zookeeper-3.4.5-cdh5.5.0.jar:3.4.5-cdh5.5.0--1]
        at org.apache.zookeeper.ZooKeeper.getChildren(ZooKeeper.java:1586) ~[zookeeper-3.4.5-cdh5.5.0.jar:3.4.5-cdh5.5.0--1]
        at org.apache.curator.framework.imps.GetChildrenBuilderImpl$3.call(GetChildrenBuilderImpl.java:214) ~[avro-tools-1.7.6-cdh5.5.0.jar:1.7.6-cdh5.5.0]
        at org.apache.curator.framework.imps.GetChildrenBuilderImpl$3.call(GetChildrenBuilderImpl.java:203) ~[avro-tools-1.7.6-cdh5.5.0.jar:1.7.6-cdh5.5.0]
        at org.apache.curator.RetryLoop.callWithRetry(RetryLoop.java:107) ~[avro-tools-1.7.6-cdh5.5.0.jar:1.7.6-cdh5.5.0]
        at org.apache.curator.framework.imps.GetChildrenBuilderImpl.pathInForeground(GetChildrenBuilderImpl.java:200) ~[avro-tools-1.7.6-cdh5.5.0.jar:1.7.6-cdh5.5.0]
        at org.apache.curator.framework.imps.GetChildrenBuilderImpl.forPath(GetChildrenBuilderImpl.java:191) ~[avro-tools-1.7.6-cdh5.5.0.jar:1.7.6-cdh5.5.0]
        at org.apache.curator.framework.imps.GetChildrenBuilderImpl.forPath(GetChildrenBuilderImpl.java:38) ~[avro-tools-1.7.6-cdh5.5.0.jar:1.7.6-cdh5.5.0]
        at storm.kafka.DynamicBrokersReader.getNumPartitions(DynamicBrokersReader.java:102) ~[storm-kafka-0.10.0.jar:0.10.0]
        at storm.kafka.DynamicBrokersReader.getBrokerInfo(DynamicBrokersReader.java:76) ~[storm-kafka-0.10.0.jar:0.10.0]
        ... 7 more
7500 [Thread-12-word-reader] ERROR b.s.util - Async loop died!
java.lang.RuntimeException: java.lang.RuntimeException: org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /kafka/brokers/topics/test/partitions
        at storm.kafka.DynamicBrokersReader.getBrokerInfo(DynamicBrokersReader.java:92) ~[storm-kafka-0.10.0.jar:0.10.0]
        at storm.kafka.trident.ZkBrokerReader.<init>(ZkBrokerReader.java:42) ~[storm-kafka-0.10.0.jar:0.10.0]
        at storm.kafka.KafkaUtils.makeBrokerReader(KafkaUtils.java:57) ~[storm-kafka-0.10.0.jar:0.10.0]
        at storm.kafka.KafkaSpout.open(KafkaSpout.java:87) ~[storm-kafka-0.10.0.jar:0.10.0]
        at backtype.storm.daemon.executor$fn__5624$fn__5639.invoke(executor.clj:564) ~[storm-core-0.10.0.jar:0.10.0]
        at backtype.storm.util$async_loop$fn__545.invoke(util.clj:477) [storm-core-0.10.0.jar:0.10.0]
        at clojure.lang.AFn.run(AFn.java:22) [clojure-1.6.0.jar:?]
        at java.lang.Thread.run(Thread.java:745) [?:1.7.0_79]
Caused by: java.lang.RuntimeException: org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /kafka/brokers/topics/test/partitions
        at storm.kafka.DynamicBrokersReader.getNumPartitions(DynamicBrokersReader.java:105) ~[storm-kafka-0.10.0.jar:0.10.0]
        at storm.kafka.DynamicBrokersReader.getBrokerInfo(DynamicBrokersReader.java:76) ~[storm-kafka-0.10.0.jar:0.10.0]
        ... 7 more
Caused by: org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /kafka/brokers/topics/test/partitions
        at org.apache.zookeeper.KeeperException.create(KeeperException.java:111) ~[zookeeper-3.4.5-cdh5.5.0.jar:3.4.5-cdh5.5.0--1]
        at org.apache.zookeeper.KeeperException.create(KeeperException.java:51) ~[zookeeper-3.4.5-cdh5.5.0.jar:3.4.5-cdh5.5.0--1]
        at org.apache.zookeeper.ZooKeeper.getChildren(ZooKeeper.java:1586) ~[zookeeper-3.4.5-cdh5.5.0.jar:3.4.5-cdh5.5.0--1]
        at org.apache.curator.framework.imps.GetChildrenBuilderImpl$3.call(GetChildrenBuilderImpl.java:214) ~[avro-tools-1.7.6-cdh5.5.0.jar:1.7.6-cdh5.5.0]
        at org.apache.curator.framework.imps.GetChildrenBuilderImpl$3.call(GetChildrenBuilderImpl.java:203) ~[avro-tools-1.7.6-cdh5.5.0.jar:1.7.6-cdh5.5.0]
        at org.apache.curator.RetryLoop.callWithRetry(RetryLoop.java:107) ~[avro-tools-1.7.6-cdh5.5.0.jar:1.7.6-cdh5.5.0]
        at org.apache.curator.framework.imps.GetChildrenBuilderImpl.pathInForeground(GetChildrenBuilderImpl.java:200) ~[avro-tools-1.7.6-cdh5.5.0.jar:1.7.6-cdh5.5.0]
        at org.apache.curator.framework.imps.GetChildrenBuilderImpl.forPath(GetChildrenBuilderImpl.java:191) ~[avro-tools-1.7.6-cdh5.5.0.jar:1.7.6-cdh5.5.0]
        at org.apache.curator.framework.imps.GetChildrenBuilderImpl.forPath(GetChildrenBuilderImpl.java:38) ~[avro-tools-1.7.6-cdh5.5.0.jar:1.7.6-cdh5.5.0]
        at storm.kafka.DynamicBrokersReader.getNumPartitions(DynamicBrokersReader.java:102) ~[storm-kafka-0.10.0.jar:0.10.0]
        at storm.kafka.DynamicBrokersReader.getBrokerInfo(DynamicBrokersReader.java:76) ~[storm-kafka-0.10.0.jar:0.10.0]
        ... 7 more
7506 [Thread-12-word-reader] ERROR b.s.d.executor - 
java.lang.RuntimeException: java.lang.RuntimeException: org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /kafka/brokers/topics/test/partitions
        at storm.kafka.DynamicBrokersReader.getBrokerInfo(DynamicBrokersReader.java:92) ~[storm-kafka-0.10.0.jar:0.10.0]
        at storm.kafka.trident.ZkBrokerReader.<init>(ZkBrokerReader.java:42) ~[storm-kafka-0.10.0.jar:0.10.0]
        at storm.kafka.KafkaUtils.makeBrokerReader(KafkaUtils.java:57) ~[storm-kafka-0.10.0.jar:0.10.0]
        at storm.kafka.KafkaSpout.open(KafkaSpout.java:87) ~[storm-kafka-0.10.0.jar:0.10.0]
        at backtype.storm.daemon.executor$fn__5624$fn__5639.invoke(executor.clj:564) ~[storm-core-0.10.0.jar:0.10.0]
        at backtype.storm.util$async_loop$fn__545.invoke(util.clj:477) [storm-core-0.10.0.jar:0.10.0]
        at clojure.lang.AFn.run(AFn.java:22) [clojure-1.6.0.jar:?]
        at java.lang.Thread.run(Thread.java:745) [?:1.7.0_79]
Caused by: java.lang.RuntimeException: org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /kafka/brokers/topics/test/partitions
        at storm.kafka.DynamicBrokersReader.getNumPartitions(DynamicBrokersReader.java:105) ~[storm-kafka-0.10.0.jar:0.10.0]
        at storm.kafka.DynamicBrokersReader.getBrokerInfo(DynamicBrokersReader.java:76) ~[storm-kafka-0.10.0.jar:0.10.0]
        ... 7 more
Caused by: org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /kafka/brokers/topics/test/partitions
        at org.apache.zookeeper.KeeperException.create(KeeperException.java:111) ~[zookeeper-3.4.5-cdh5.5.0.jar:3.4.5-cdh5.5.0--1]
        at org.apache.zookeeper.KeeperException.create(KeeperException.java:51) ~[zookeeper-3.4.5-cdh5.5.0.jar:3.4.5-cdh5.5.0--1]
        at org.apache.zookeeper.ZooKeeper.getChildren(ZooKeeper.java:1586) ~[zookeeper-3.4.5-cdh5.5.0.jar:3.4.5-cdh5.5.0--1]
        at org.apache.curator.framework.imps.GetChildrenBuilderImpl$3.call(GetChildrenBuilderImpl.java:214) ~[avro-tools-1.7.6-cdh5.5.0.jar:1.7.6-cdh5.5.0]
        at org.apache.curator.framework.imps.GetChildrenBuilderImpl$3.call(GetChildrenBuilderImpl.java:203) ~[avro-tools-1.7.6-cdh5.5.0.jar:1.7.6-cdh5.5.0]
        at org.apache.curator.RetryLoop.callWithRetry(RetryLoop.java:107) ~[avro-tools-1.7.6-cdh5.5.0.jar:1.7.6-cdh5.5.0]
        at org.apache.curator.framework.imps.GetChildrenBuilderImpl.pathInForeground(GetChildrenBuilderImpl.java:200) ~[avro-tools-1.7.6-cdh5.5.0.jar:1.7.6-cdh5.5.0]
        at org.apache.curator.framework.imps.GetChildrenBuilderImpl.forPath(GetChildrenBuilderImpl.java:191) ~[avro-tools-1.7.6-cdh5.5.0.jar:1.7.6-cdh5.5.0]
        at org.apache.curator.framework.imps.GetChildrenBuilderImpl.forPath(GetChildrenBuilderImpl.java:38) ~[avro-tools-1.7.6-cdh5.5.0.jar:1.7.6-cdh5.5.0]
        at storm.kafka.DynamicBrokersReader.getNumPartitions(DynamicBrokersReader.java:102) ~[storm-kafka-0.10.0.jar:0.10.0]
        at storm.kafka.DynamicBrokersReader.getBrokerInfo(DynamicBrokersReader.java:76) ~[storm-kafka-0.10.0.jar:0.10.0]
        ... 7 more
7506 [Thread-17-word-reader] ERROR b.s.d.executor - 
java.lang.RuntimeException: java.lang.RuntimeException: org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /kafka/brokers/topics/test/partitions
        at storm.kafka.DynamicBrokersReader.getBrokerInfo(DynamicBrokersReader.java:92) ~[storm-kafka-0.10.0.jar:0.10.0]
        at storm.kafka.trident.ZkBrokerReader.<init>(ZkBrokerReader.java:42) ~[storm-kafka-0.10.0.jar:0.10.0]
        at storm.kafka.KafkaUtils.makeBrokerReader(KafkaUtils.java:57) ~[storm-kafka-0.10.0.jar:0.10.0]
        at storm.kafka.KafkaSpout.open(KafkaSpout.java:87) ~[storm-kafka-0.10.0.jar:0.10.0]
        at backtype.storm.daemon.executor$fn__5624$fn__5639.invoke(executor.clj:564) ~[storm-core-0.10.0.jar:0.10.0]
        at backtype.storm.util$async_loop$fn__545.invoke(util.clj:477) [storm-core-0.10.0.jar:0.10.0]
        at clojure.lang.AFn.run(AFn.java:22) [clojure-1.6.0.jar:?]
        at java.lang.Thread.run(Thread.java:745) [?:1.7.0_79]
Caused by: java.lang.RuntimeException: org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /kafka/brokers/topics/test/partitions
        at storm.kafka.DynamicBrokersReader.getNumPartitions(DynamicBrokersReader.java:105) ~[storm-kafka-0.10.0.jar:0.10.0]
        at storm.kafka.DynamicBrokersReader.getBrokerInfo(DynamicBrokersReader.java:76) ~[storm-kafka-0.10.0.jar:0.10.0]
        ... 7 more
Caused by: org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /kafka/brokers/topics/test/partitions
        at org.apache.zookeeper.KeeperException.create(KeeperException.java:111) ~[zookeeper-3.4.5-cdh5.5.0.jar:3.4.5-cdh5.5.0--1]
        at org.apache.zookeeper.KeeperException.create(KeeperException.java:51) ~[zookeeper-3.4.5-cdh5.5.0.jar:3.4.5-cdh5.5.0--1]
        at org.apache.zookeeper.ZooKeeper.getChildren(ZooKeeper.java:1586) ~[zookeeper-3.4.5-cdh5.5.0.jar:3.4.5-cdh5.5.0--1]
        at org.apache.curator.framework.imps.GetChildrenBuilderImpl$3.call(GetChildrenBuilderImpl.java:214) ~[avro-tools-1.7.6-cdh5.5.0.jar:1.7.6-cdh5.5.0]
        at org.apache.curator.framework.imps.GetChildrenBuilderImpl$3.call(GetChildrenBuilderImpl.java:203) ~[avro-tools-1.7.6-cdh5.5.0.jar:1.7.6-cdh5.5.0]
        at org.apache.curator.RetryLoop.callWithRetry(RetryLoop.java:107) ~[avro-tools-1.7.6-cdh5.5.0.jar:1.7.6-cdh5.5.0]
        at org.apache.curator.framework.imps.GetChildrenBuilderImpl.pathInForeground(GetChildrenBuilderImpl.java:200) ~[avro-tools-1.7.6-cdh5.5.0.jar:1.7.6-cdh5.5.0]
        at org.apache.curator.framework.imps.GetChildrenBuilderImpl.forPath(GetChildrenBuilderImpl.java:191) ~[avro-tools-1.7.6-cdh5.5.0.jar:1.7.6-cdh5.5.0]
        at org.apache.curator.framework.imps.GetChildrenBuilderImpl.forPath(GetChildrenBuilderImpl.java:38) ~[avro-tools-1.7.6-cdh5.5.0.jar:1.7.6-cdh5.5.0]
        at storm.kafka.DynamicBrokersReader.getNumPartitions(DynamicBrokersReader.java:102) ~[storm-kafka-0.10.0.jar:0.10.0]
        at storm.kafka.DynamicBrokersReader.getBrokerInfo(DynamicBrokersReader.java:76) ~[storm-kafka-0.10.0.jar:0.10.0]
        ... 7 more
7512 [ProcessThread(sid:0 cport:-1):] INFO  o.a.s.s.o.a.z.s.PrepRequestProcessor - Got user-level KeeperException when processing sessionid:0x1525ee04dcb000c type:create cxid:0x30 zxid:0x2b txntype:-1 reqpath:n/a Error Path:/storm/errors/local-host-1-1453290705 Error:KeeperErrorCode = NodeExists for /storm/errors/local-host-1-1453290705
7513 [ProcessThread(sid:0 cport:-1):] INFO  o.a.s.s.o.a.z.s.PrepRequestProcessor - Got user-level KeeperException when processing sessionid:0x1525ee04dcb000c type:create cxid:0x31 zxid:0x2d txntype:-1 reqpath:n/a Error Path:/storm/errors/local-host-1-1453290705/word-reader Error:KeeperErrorCode = NodeExists for /storm/errors/local-host-1-1453290705/word-reader
7517 [ProcessThread(sid:0 cport:-1):] INFO  o.a.s.s.o.a.z.s.PrepRequestProcessor - Got user-level KeeperException when processing sessionid:0x1525ee04dcb000c type:create cxid:0x36 zxid:0x31 txntype:-1 reqpath:n/a Error Path:/storm/errors/local-host-1-1453290705/word-reader-last-error Error:KeeperErrorCode = NodeExists for /storm/errors/local-host-1-1453290705/word-reader-last-error
7518 [Thread-12-word-reader] ERROR o.a.s.s.o.a.z.s.NIOServerCnxnFactory - Thread Thread[Thread-12-word-reader,5,main] died
java.lang.RuntimeException: org.apache.storm.shade.org.apache.zookeeper.KeeperException$NodeExistsException: KeeperErrorCode = NodeExists for /errors/local-host-1-1453290705/word-reader-last-error
        at backtype.storm.util$wrap_in_runtime.invoke(util.clj:49) ~[storm-core-0.10.0.jar:0.10.0]
        at backtype.storm.zookeeper$create_node.invoke(zookeeper.clj:92) ~[storm-core-0.10.0.jar:0.10.0]
        at backtype.storm.cluster$mk_distributed_cluster_state$reify__4580.set_data(cluster.clj:106) ~[storm-core-0.10.0.jar:0.10.0]
        at backtype.storm.cluster$mk_storm_cluster_state$reify__5120.report_error(cluster.clj:465) ~[storm-core-0.10.0.jar:0.10.0]
        at backtype.storm.daemon.executor$throttled_report_error_fn$fn__5469.invoke(executor.clj:193) ~[storm-core-0.10.0.jar:0.10.0]
        at backtype.storm.daemon.executor$mk_executor_data$fn__5523$fn__5524.invoke(executor.clj:256) ~[storm-core-0.10.0.jar:0.10.0]
        at backtype.storm.util$async_loop$fn__545.invoke(util.clj:489) ~[storm-core-0.10.0.jar:0.10.0]
        at clojure.lang.AFn.run(AFn.java:22) ~[clojure-1.6.0.jar:?]
        at java.lang.Thread.run(Thread.java:745) [?:1.7.0_79]
Caused by: org.apache.storm.shade.org.apache.zookeeper.KeeperException$NodeExistsException: KeeperErrorCode = NodeExists for /errors/local-host-1-1453290705/word-reader-last-error
        at org.apache.storm.shade.org.apache.zookeeper.KeeperException.create(KeeperException.java:119) ~[storm-core-0.10.0.jar:0.10.0]
        at org.apache.storm.shade.org.apache.zookeeper.KeeperException.create(KeeperException.java:51) ~[storm-core-0.10.0.jar:0.10.0]
        at org.apache.storm.shade.org.apache.zookeeper.ZooKeeper.create(ZooKeeper.java:783) ~[storm-core-0.10.0.jar:0.10.0]
        at org.apache.storm.shade.org.apache.curator.framework.imps.CreateBuilderImpl$11.call(CreateBuilderImpl.java:676) ~[storm-core-0.10.0.jar:0.10.0]
        at org.apache.storm.shade.org.apache.curator.framework.imps.CreateBuilderImpl$11.call(CreateBuilderImpl.java:660) ~[storm-core-0.10.0.jar:0.10.0]
        at org.apache.storm.shade.org.apache.curator.RetryLoop.callWithRetry(RetryLoop.java:107) ~[storm-core-0.10.0.jar:0.10.0]
        at org.apache.storm.shade.org.apache.curator.framework.imps.CreateBuilderImpl.pathInForeground(CreateBuilderImpl.java:656) ~[storm-core-0.10.0.jar:0.10.0]
        at org.apache.storm.shade.org.apache.curator.framework.imps.CreateBuilderImpl.protectedPathInForeground(CreateBuilderImpl.java:441) ~[storm-core-0.10.0.jar:0.10.0]
        at org.apache.storm.shade.org.apache.curator.framework.imps.CreateBuilderImpl.forPath(CreateBuilderImpl.java:431) ~[storm-core-0.10.0.jar:0.10.0]
        at org.apache.storm.shade.org.apache.curator.framework.imps.CreateBuilderImpl$3.forPath(CreateBuilderImpl.java:239) ~[storm-core-0.10.0.jar:0.10.0]
        at org.apache.storm.shade.org.apache.curator.framework.imps.CreateBuilderImpl$3.forPath(CreateBuilderImpl.java:193) ~[storm-core-0.10.0.jar:0.10.0]
        at sun.reflect.GeneratedMethodAccessor40.invoke(Unknown Source) ~[?:?]
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[?:1.7.0_79]
        at java.lang.reflect.Method.invoke(Method.java:606) ~[?:1.7.0_79]
        at clojure.lang.Reflector.invokeMatchingMethod(Reflector.java:93) ~[clojure-1.6.0.jar:?]
        at clojure.lang.Reflector.invokeInstanceMethod(Reflector.java:28) ~[clojure-1.6.0.jar:?]
        at backtype.storm.zookeeper$create_node.invoke(zookeeper.clj:91) ~[storm-core-0.10.0.jar:0.10.0]
        ... 7 more
7526 [Thread-17-word-reader] ERROR b.s.util - Halting process: ("Worker died")
java.lang.RuntimeException: ("Worker died")
        at backtype.storm.util$exit_process_BANG_.doInvoke(util.clj:336) [storm-core-0.10.0.jar:0.10.0]
        at clojure.lang.RestFn.invoke(RestFn.java:423) [clojure-1.6.0.jar:?]
        at backtype.storm.daemon.worker$fn__7184$fn__7185.invoke(worker.clj:532) [storm-core-0.10.0.jar:0.10.0]
        at backtype.storm.daemon.executor$mk_executor_data$fn__5523$fn__5524.invoke(executor.clj:261) [storm-core-0.10.0.jar:0.10.0]
        at backtype.storm.util$async_loop$fn__545.invoke(util.clj:489) [storm-core-0.10.0.jar:0.10.0]
        at clojure.lang.AFn.run(AFn.java:22) [clojure-1.6.0.jar:?]
        at java.lang.Thread.run(Thread.java:745) [?:1.7.0_79]


相关代码： 

public static void main(String[] args) throws InterruptedException, FileNotFoundException, AuthorizationException, AlreadyAliveException, InvalidTopologyException {
		BrokerHosts brokerHosts = new ZkHosts("wycoldb01:2181,wycoldb02:2181,wycoldb05:2181", "/kafka/brokers");
		SpoutConfig spoutConf = new SpoutConfig(brokerHosts, "test", "/kafkastorm", "word");

		spoutConf.scheme = new SchemeAsMultiScheme(new StringScheme());
//		spoutConf.forceStartOffsetTime(-2);

		spoutConf.zkServers = new ArrayList<String>() {
			{
				add("wycoldb01");
				add("wycoldb02");
				add("wycoldb05");
			}
		};
		spoutConf.zkPort = 2181;


原因分析：

查看storm官网关于kafka的文档上看到：
storm集成kafka报错【KeeperErrorCode = NoNode for /kafka/brokers/topics/test/partitions】

 
ZkHosts

ZkHosts is what you should use if you want to dynamically track Kafka broker to partition mapping. This class uses Kafka's ZooKeeper entries to track brokerHost -> partition mapping. You can instantiate an object by calling java public ZkHosts(String brokerZkStr, String brokerZkPath) public ZkHosts(String brokerZkStr) Where brokerZkStr is just ip:port (e.g. localhost:2181). brokerZkPath is the root directory under which all the topics and partition information is stored. By default this is /brokers which is what the default Kafka implementation uses.

By default, the broker-partition mapping is refreshed every 60 seconds from ZooKeeper. If you want to change it, you should set host.refreshFreqSecs to your chosen value.
在从网上拷贝过来的代码上，使用了第一种构造方法。于是改成第二中构造，即使用默认的构造方法

但是自己却不是特别理解这个brokerZkPath路径，下来抽点时间把框架原理整整，如果有知道的大牛大神，请不吝赐教，谢谢。

解决方案：

BrokerHosts brokerHosts = new ZkHosts("wycoldb01:2181,wycoldb02:2181,wycoldb05:2181", "/kafka/brokers");

修改为： 
BrokerHosts brokerHosts = new ZkHosts("wycoldb01:2181,wycoldb02:2181,wycoldb05:2181");

问题解决！
标签：

相关文章