Storm1.1.0 + kafka0.10.2.1：NoSuchMethodError&ExceptionInInitializerError

Question

我是storm和kafka的新手。我要运行一个开源项目：github:cestella:streaming_outliers

本项目使用storm0.10.0。但是，我的风暴集群是storm1.0.3，它不适用于storm0.10.0。

所以，我必须更改storm的版本和一些相关的包。

然后，当我运行这个拓扑时，会出现以下错误。

java.lang.NoSuchMethodError: org.apache.kafka.common.network.NetworkSend.(Ljava/lang/String;Ljava/nio/ByteBuffer;)V at kafka.network.RequestOrResponseSend.(RequestOrResponseSend.scala:41)在 kafka.network.RequestOrResponseSend.(RequestOrResponseSend.scala:44) 在 kafka.network.BlockingChannel.send(BlockingChannel.scala:112) 在 kafka.consumer.SimpleConsumer.liftedTree1$1(SimpleConsumer.scala:98) 在 kafka.consumer。 SimpleConsumer.kafka$consumer$SimpleConsumer$$sendRequest(SimpleConsumer.scala:83) at kafka.consumer.SimpleConsumer.getOffsetsBefore(SimpleConsumer.scala:149) at kafka.javaapi.consumer.SimpleConsumer.getOffsetsBefore(SimpleConsumer.scala:79) at org.apache.storm.kafka.KafkaUtils.getOffset(KafkaUtils.java:75) 在 org.apache.storm.kafka.KafkaUtils.getOffset(KafkaUtils.java:65) 在 org.apache.storm.kafka.PartitionManager.(PartitionManager .java:103) 在 org.apache.storm.kafka.ZkCoordinator.refresh(ZkCoordinator.java:98) 在 org.apache.storm.kafka.ZkCoordinator.getMyManagedPartitions(Z kCoordinator.java:69) at org.apache.storm.kafka.KafkaSpout.nextTuple(KafkaSpout.java:129) at org.apache.storm.daemon.executor$fn__4905$fn__4920$fn__4951.invoke(executor.clj:644)在 org.apache.storm.util$async_loop$fn__557.invoke(util.clj:484) 在 clojure.lang.AFn.run(AFn.java:22) 在 java.lang.Thread.run(Thread.java:745 )

和

java.lang.ExceptionInInitializerError at kafka.metrics.KafkaMetricsGroup$class.newTimer(KafkaMetricsGroup.scala:90) at kafka.consumer.FetchRequestAndResponseMetrics.newTimer(FetchRequestAndResponseStats.scala:26) at kafka.consumer.FetchRequestAndResponseMetrics.(FetchRequestAndResponseStats. scala:35) at kafka.consumer.FetchRequestAndResponseStats.(FetchRequestAndResponseStats.scala:47) at kafka.consumer.FetchRequestAndResponseStatsRegistry$$anonfun$2.apply(FetchRequestAndResponseStats.scala:60) at kafka.consumer.FetchRequestAndResponseStatsRegistry$$anonfun$2.apply( FetchRequestAndResponseStats.scala:60) at kafka.utils.Pool$$anonfun$getAndMaybePut$1.apply(Pool.scala:52) at kafka.utils.Pool.getAndMaybePut(Pool.scala:70) at kafka.utils.Pool.getAndMaybePut (Pool.scala:52) 在 kafka.consumer.FetchRequestAndResponseStatsRegistry$.getFetchRequestAndResponseStats(FetchRequestAndResponseStats.scala:64) 在 kafka.consumer.SimpleConsumer.(SimpleConsumer.scala:44) 在 kafka.javaapi.consumer.SimpleC onsumer.(SimpleConsumer.scala:34) at org.apache.storm.kafka.DynamicPartitionConnections.register(DynamicPartitionConnections.java:60) at org.apache.storm.kafka.PartitionManager.(PartitionManager.java:74) at org.apache .storm.kafka.ZkCoordinator.refresh(ZkCoordinator.java:98) at org.apache.storm.kafka.ZkCoordinator.getMyManagedPartitions(ZkCoordinator.java:69) at org.apache.storm.kafka.KafkaSpout.nextTuple(KafkaSpout.java :129) at org.apache.storm.daemon.executor$fn__4905$fn__4920$fn__4951.invoke(executor.clj:644) at org.apache.storm.util$async_loop$fn__557.invoke(util.clj:484) at clojure.lang.AFn.run(AFn.java:22) at java.lang.Thread.run(Thread.java:745) 原因：java.lang.IllegalStateException: java.lang.ApplicationShutdownHooks.add( ApplicationShutdownHooks.java:66) 在 java.lang.Runtime.addShutdownHook(Runtime.java:211) 在 com.yammer.metrics.Metrics.(Metrics.java:21) ... 21 更多

和

java.lang.NullPointerException 在 net.opentsdb.utils.Config.getBoolean(Config.java:354) ~[stormjar.jar:1.6.6] 在 net.opentsdb.utils.Config.loadStaticVariables(Config.java:620) ~[stormjar.jar:1.6.6] 在 net.opentsdb.utils.Config.setDefaults(Config.java:532) ~[stormjar.jar:1.6.6] 在 net.opentsdb.utils.Config.(Config.java:131) ~[stormjar.jar:1.6.6] 在 com.caseystella.analytics.timeseries.tsdb.TSDBHandler$TSDBConfig.(TSDBHandler.java:127) ~[stormjar.jar:1.6.6] 在 com.caseystella.analytics.timeseries.tsdb.TSDBHandler$TSDBConfig.(TSDBHandler.java:124) ~[stormjar.jar:1.6.6] 在 com.caseystella.analytics.timeseries.tsdb.TSDBHandler.configure(TSDBHandler.java:144) ~[stormjar.jar:1.6.6] 在 com.caseystella.analytics.outlier.OutlierBolt.prepare(OutlierBolt.java:63) ~[stormjar.jar:1.6.6] 在 org.apache.storm.daemon.executor$fn__4973$fn__4986.invoke(executor.clj:791) ~[storm-core-1.0.3.jar:1.0.3] 在 org.apache.storm.util$async_loop$fn__557.invoke(util.clj:482) [storm-core-1.0.3.jar:1.0.3] java.lang.NullPointerException 在 net.opentsdb.utils.Config.getBoolean(Config.java:354) ~[stormjar.jar:1.6.6] 在 net.opentsdb.utils.Config.loadStaticVariables(Config.java:620) ~[stormjar.jar:1.6.6] 在 net.opentsdb.utils.Config.setDefaults(Config.java:532) ~[stormjar.jar:1.6.6] 在 net.opentsdb.utils.Config.(Config.java:131) ~[stormjar.jar:1.6.6] 在 com.caseystella.analytics.timeseries.tsdb.TSDBHandler$TSDBConfig.(TSDBHandler.java:127) ~[stormjar.jar:1.6.6] 在 com.caseystella.analytics.timeseries.tsdb.TSDBHandler$TSDBConfig.(TSDBHandler.java:124) ~[stormjar.jar:1.6.6] 在 com.caseystella.analytics.timeseries.tsdb.TSDBHandler.configure(TSDBHandler.java:144) ~[stormjar.jar:1.6.6] 在 com.caseystella.analytics.outlier.OutlierBolt.prepare(OutlierBolt.java:63) ~[stormjar.jar:1.6.6] 在 org.apache.storm.daemon.executor$fn__4973$fn__4986.invoke(executor.clj:791) ~[storm-core-1.0.3.jar:1.0.3] 在 org.apache.storm.util$async_loop$fn__557.invoke(util.clj:482) [storm-core-1.0.3.jar:1.0.3]

以下是包的版本。

kafka_version: 0.10.2.1

kafka-clients 版本：0.10.2.1

storm_version：1.1.0

storm-kafka 版本：1.1.0

flux_version: 1.1.0

tsdb_version: 2.3.0

网上有一些方法说附加kafka-clients的依赖可以解决第一个错误。但是，它不起作用。

非常感谢。

Answer 1

当您的应用程序包含同一个 jar 的多个版本（例如 kafka 0.10.2.1 和 0.10.0.1）时，不会出现此类方法错误。检查您的项目依赖关系并删除旧版本的 jar。

Answer 2

pom.xml 中的 kafka 和 kafka-clients 版本应该相互匹配。

<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka_2.12</artifactId>
  <version>1.0.0</version>
</dependency>

<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-clients</artifactId>
  <version>1.0.0</version>
</dependency>