无法使用strom集群从kafka读取

Question

我正在运行Strom集群，其中2个主管和1个nimbus正在运行。我在哪里阅读主题为“topic1”的kafka。但在UI上，我得到的错误如下

java.lang.RuntimeException：java.lang.RuntimeException：org.apache.zookeeper.KeeperException $ NoNodeException：KeeperErrorCode = NoNode for/brokers/topics/topic1/partitions at storm.kafka .DynamicBrok

和工人_ *。日志显示以下错误。

java.io.FileNotFoundException：文件 '/app/storm/supervisor/stormdist/airpushTop-1-1431081661/stormconf.ser' 不存在 在org.apache.commons.io.FileUtils.openInputStream（文件实用程序。 java：299）〜[commons-io-2.4.jar：2.4] at org.apache.commons.io.FileUtils.readFileToByteArray（FileUtils.java:1763）〜[commons-io-2.4.jar：2.4] at backtype.storm.config $ read_supervisor_storm_conf.invoke（config.clj：212）〜[storm-core-0.9.4.jar：0.9.4] at backtype.storm.daemon.worker $ worker_data.invoke（worker.clj： 182）〜[storm-core-0.9.4.jar：0.9.4] at backtype.storm.daemon.worker $ fn__5033 $ exec_fn__1754__auto ____ 5034.invoke（worker.clj：398）〜[storm-core-0.9.4。 jar：0.9.4] at clojure.lang.AFn.applyToH elper（AFn.java:185）[clojure-1.5.1.jar：na] at clojure.lang.AFn.applyTo（AFn.java:151）[clojure-1.5.1.jar：na] at clojure。核心$ apply.invoke（core.clj：617）〜[clojure-1.5.1.jar：na] at backtype.storm.daemon.worker $ fn__5033 $ mk_worker__5089.doInvoke（worker.clj：389）[storm-core -0.9.4.jar：0.9.4] at backjure.lang.RestFn.invoke（RestFn.java:512）[clojure-1.5.1.jar：na] at backtype.storm.daemon.worker $ _main。 invoke（worker.clj：500）[storm-core-0.9.4.jar：0.9.4] at clojure.lang.AFn.applyToHelper（AFn.java:172）[clojure-1.5.1.jar：na] at clojure.lang.AFn.applyTo（AFn.java:151）[clojure-1.5.1.jar：na] at backtype.storm.daemon.worker.main（Unknown Source）[storm-core-0.9.4 .jar：0.9.4] 2015-05-08T17：09：19.209 + 0530 bs .util [ERROR]暂停过程：（“初始化时出错”） java.lang.RuntimeException：（“初始化时出错”） at backtype.storm.util $ exit_process_BANG_.doInvoke（util.clj：325）[storm- core-0.9.4.jar：0.9.4] at clojure.lang.RestFn.invoke（RestFn.java:423）[clojure-1.5.1.jar：na] at backtype.storm.daemon.worker $ fn__5033 $ mk_worker__5089.doInvoke（worker.clj：389）[storm-core-0.9.4.jar：0.9.4] at clojure.lang.RestFn.invoke（RestFn.java:512）[clojure-1.5.1.jar ：na] at backtype.storm.daemon.worker $ _main.invoke（worker.clj：500）[storm-core-0.9.4.jar：0.9.4] at clojure.lang.AFn.applyToHelper（AFn。 java：172）[clojure-1.5.1.jar：na] at clojure.lang.AFn.applyTo（AFn.java:151）[clojure-1.5.1.jar：na] ava.lang.R unnyException：java.lang.RuntimeException：org.apache.zookeeper.KeeperException $ NoNodeException：KeeperErrorCode = NoNode for/brokers/topics/topic1/partitions at storm.kafka.DynamicBrokt backtype.storm.daemon.worker.main（Unknown Source）[ storm-core-0.9.4.jar：0.9.4]

我使用的是kafka_2.11-0.8.2.1，apache-storm-0.9.4和zookeeper-3.4.6。

kafka和cluster中没有不匹配的zookeeper jar版本。

请帮我这里。

非常感谢提前。

Answer 1

您是否确定了kafka中的topic1，并且您可以使用控制台使用者从中读取它，我已经看过org.apache.zookeeper。KeeperException $ NoNodeException：KeeperErrorCode = NoNode for ...当部署风暴拓扑时该主题不存在时。

你也可以发布一些关于如何配置卡夫卡喷嘴的代码？

这个错误是零星的还是一致的？