1、在storm整合kafka后,运行storm程序,报错如下:
12:47:09.110 [Thread-26-kafkaspout-executor[3 4]] ERROR org.apache.storm.util - Async loop died!
java.lang.NoClassDefFoundError: Could not initialize class org.apache.log4j.Log4jLoggerFactory
at org.apache.log4j.Logger.getLogger(Logger.java:39) ~[log4j-over-slf4j-1.6.6.jar:1.6.6]
at kafka.utils.Logging$class.logger(Logging.scala:24) ~[kafka_2.11-0.10.0.1.jar:?]
at kafka.consumer.SimpleConsumer.logger$lzycompute(SimpleConsumer.scala:35) ~[kafka_2.11-0.10.0.1.jar:?]
at kafka.consumer.SimpleConsumer.logger(SimpleConsumer.scala:35) ~[kafka_2.11-0.10.0.1.jar:?]
at kafka.utils.Logging$class.info(Logging.scala:75) ~[kafka_2.11-0.10.0.1.jar:?]
at kafka.consumer.SimpleConsumer.info(SimpleConsumer.scala:35) ~[kafka_2.11-0.10.0.1.jar:?]
at kafka.consumer.SimpleConsumer.liftedTree1$1(SimpleConsumer.scala:94) ~[kafka_2.11-0.10.0.1.jar:?]
at kafka.consumer.SimpleConsumer.kafka$consumer$SimpleConsumer$$sendRequest(SimpleConsumer.scala:83) ~[kafka_2.11-0.10.0.1.jar:?]
at kafka.consumer.SimpleConsumer.getOffsetsBefore(SimpleConsumer.scala:149) ~[kafka_2.11-0.10.0.1.jar:?]
at kafka.javaapi.consumer.SimpleConsumer.getOffsetsBefore(SimpleConsumer.scala:79) ~[kafka_2.11-0.10.0.1.jar:?]
at org.apache.storm.kafka.KafkaUtils.getOffset(KafkaUtils.java:81) ~[storm-kafka-1.2.2.jar:1.2.2]
at org.apache.storm.kafka.KafkaUtils.getOffset(KafkaUtils.java:71) ~[storm-kafka-1.2.2.jar:1.2.2]
at org.apache.storm.kafka.PartitionManager.<init>(PartitionManager.java:135) ~[storm-kafka-1.2.2.jar:1.2.2]
at org.apache.storm.kafka.ZkCoordinator.refresh(ZkCoordinator.java:110) ~[storm-kafka-1.2.2.jar:1.2.2]
at org.apache.storm.kafka.ZkCoordinator.getMyManagedPartitions(ZkCoordinator.java:71) ~[storm-kafka-1.2.2.jar:1.2.2]
at org.apache.storm.kafka.KafkaSpout.nextTuple(KafkaSpout.java:135) ~[storm-kafka-1.2.2.jar:1.2.2]
at org.apache.storm.daemon.executor$fn__10727$fn__10742$fn__10773.invoke(executor.clj:654) ~[storm-core-1.2.2.jar:1.2.2]
at org.apache.storm.util$async_loop$fn__553.invoke(util.clj:484) [storm-core-1.2.2.jar:1.2.2]
at clojure.lang.AFn.run(AFn.java:22) [clojure-1.7.0.jar:?]
at java.lang.Thread.run(Thread.java:745) [?:1.8.0_45]
12:47:09.132 [Thread-26-kafkaspout-executor[3 4]] ERROR org.apache.storm.daemon.executor -
java.lang.NoClassDefFoundError: Could not initialize class org.apache.log4j.Log4jLoggerFactory
at org.apache.log4j.Logger.getLogger(Logger.java:39) ~[log4j-over-slf4j-1.6.6.jar:1.6.6]
at kafka.utils.Logging$class.logger(Logging.scala:24) ~[kafka_2.11-0.10.0.1.jar:?]
at kafka.consumer.SimpleConsumer.logger$lzycompute(SimpleConsumer.scala:35) ~[kafka_2.11-0.10.0.1.jar:?]
at kafka.consumer.SimpleConsumer.logger(SimpleConsumer.scala:35) ~[kafka_2.11-0.10.0.1.jar:?]
at kafka.utils.Logging$class.info(Logging.scala:75) ~[kafka_2.11-0.10.0.1.jar:?]
at kafka.consumer.SimpleConsumer.info(SimpleConsumer.scala:35) ~[kafka_2.11-0.10.0.1.jar:?]
at kafka.consumer.SimpleConsumer.liftedTree1$1(SimpleConsumer.scala:94) ~[kafka_2.11-0.10.0.1.jar:?]
at kafka.consumer.SimpleConsumer.kafka$consumer$SimpleConsumer$$sendRequest(SimpleConsumer.scala:83) ~[kafka_2.11-0.10.0.1.jar:?]
at kafka.consumer.SimpleConsumer.getOffsetsBefore(SimpleConsumer.scala:149) ~[kafka_2.11-0.10.0.1.jar:?]
at kafka.javaapi.consumer.SimpleConsumer.getOffsetsBefore(SimpleConsumer.scala:79) ~[kafka_2.11-0.10.0.1.jar:?]
at org.apache.storm.kafka.KafkaUtils.getOffset(KafkaUtils.java:81) ~[storm-kafka-1.2.2.jar:1.2.2]
at org.apache.storm.kafka.KafkaUtils.getOffset(KafkaUtils.java:71) ~[storm-kafka-1.2.2.jar:1.2.2]
at org.apache.storm.kafka.PartitionManager.<init>(PartitionManager.java:135) ~[storm-kafka-1.2.2.jar:1.2.2]
at org.apache.storm.kafka.ZkCoordinator.refresh(ZkCoordinator.java:110) ~[storm-kafka-1.2.2.jar:1.2.2]
at org.apache.storm.kafka.ZkCoordinator.getMyManagedPartitions(ZkCoordinator.java:71) ~[storm-kafka-1.2.2.jar:1.2.2]
at org.apache.storm.kafka.KafkaSpout.nextTuple(KafkaSpout.java:135) ~[storm-kafka-1.2.2.jar:1.2.2]
at org.apache.storm.daemon.executor$fn__10727$fn__10742$fn__10773.invoke(executor.clj:654) ~[storm-core-1.2.2.jar:1.2.2]
at org.apache.storm.util$async_loop$fn__553.invoke(util.clj:484) [storm-core-1.2.2.jar:1.2.2]
at clojure.lang.AFn.run(AFn.java:22) [clojure-1.7.0.jar:?]
at java.lang.Thread.run(Thread.java:745) [?:1.8.0_45]
12:47:09.191 [Thread-26-kafkaspout-executor[3 4]] ERROR org.apache.storm.util - Halting process: ("Worker died")
java.lang.RuntimeException: ("Worker died")
at org.apache.storm.util$exit_process_BANG_.doInvoke(util.clj:341) [storm-core-1.2.2.jar:1.2.2]
at clojure.lang.RestFn.invoke(RestFn.java:423) [clojure-1.7.0.jar:?]
at org.apache.storm.daemon.worker$fn__11400$fn__11401.invoke(worker.clj:788) [storm-core-1.2.2.jar:1.2.2]
at org.apache.storm.daemon.executor$mk_executor_data$fn__10612$fn__10613.invoke(executor.clj:281) [storm-core-1.2.2.jar:1.2.2]
at org.apache.storm.util$async_loop$fn__553.invoke(util.clj:494) [storm-core-1.2.2.jar:1.2.2]
at clojure.lang.AFn.run(AFn.java:22) [clojure-1.7.0.jar:?]
at java.lang.Thread.run(Thread.java:745) [?:1.8.0_45]
Process finished with exit code 1
上述问题是kafka中的包和storm中的jar冲突导致的,解决方法是如下:
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.11</artifactId>
<version>0.10.0.1</version>
<!-- 解决jog4j冲突 -->
<!-- 在kafka的依赖处加入如下 -->
<exclusions>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
</exclusion>
<exclusion>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
</exclusion>
</exclusions>
<!-- -->
</dependency>
参考文章:传送门