NoClassDefFoundError：Storm jar 的 kafka/api/OffsetRequest

Question

我正在尝试将 Storm 拓扑提交到集群，但我不断收到相同的错误：

Exception in thread "main" java.lang.NoClassDefFoundError: kafka/api/OffsetRequest
    at org.apache.storm.kafka.KafkaConfig.<init>(KafkaConfig.java:48)
    at org.apache.storm.kafka.trident.TridentKafkaConfig.<init>(TridentKafkaConfig.java:30)
    at storm.StormStreaming$.main(StormStreaming.scala:41)
    at storm.StormStreaming.main(StormStreaming.scala)
Caused by: java.lang.ClassNotFoundException: kafka.api.OffsetRequest
    at java.net.URLClassLoader.findClass(URLClassLoader.java:382)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
    at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:349)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
    ... 4 more

我使用

提交 jar 文件

./storm jar /patht/storm-app.jar storm.StormStreaming

如何解决？我尝试调整资源版本 - Kafka 和 Storm - 但它似乎不起作用。 我的 build.sbt 文件：

scalaVersion := "2.12.8"
val sparkVersion = "2.4.4"
val flinkVersion = "1.9.1"
val stormVersion = "2.1.0"
val kafkaVersion = "2.4.0"
    dependencyOverrides += "com.fasterxml.jackson.core" % "jackson-core" % "2.9.6"
    dependencyOverrides += "com.fasterxml.jackson.core" % "jackson-databind" % "2.9.6"
    dependencyOverrides += "com.fasterxml.jackson.module" % "jackson-module-scala_2.11" % "2.9.6"

    libraryDependencies ++= Seq(
      "org.apache.kafka" %% "kafka" % kafkaVersion excludeAll(
        ExclusionRule("org.slf4j", "slf4j-log4j12"),
        ExclusionRule("log4j", "log4j"),
        ExclusionRule("org.apache.zookeeper", "zookeeper")
        ),
      "org.apache.spark" %% "spark-sql-kafka-0-10" % sparkVersion,
      "org.apache.spark" %% "spark-core" % sparkVersion excludeAll("org.slf4j", "slf4j-log4j12"),
      "org.apache.spark" %% "spark-streaming" % sparkVersion,
      "org.apache.spark" %% "spark-sql" % sparkVersion excludeAll("org.slf4j", "slf4j-log4j12"),
      "org.apache.spark" %% "spark-streaming-kafka-0-10" % sparkVersion,
      "com.typesafe" % "config" % "1.3.3",
      "org.twitter4j" % "twitter4j-core" % "4.0.7",
      "org.twitter4j" % "twitter4j-stream" % "4.0.7",
      "org.apache.flink" % "flink-core" % flinkVersion % "provided",
      "org.apache.flink" %% "flink-streaming-scala" % flinkVersion excludeAll("org.slf4j", "slf4j-log4j12"),
      "org.apache.flink" %% "flink-scala" % flinkVersion excludeAll("org.slf4j", "slf4j-log4j12"),
      "org.apache.flink" %% "flink-clients" % flinkVersion excludeAll("org.slf4j", "slf4j-log4j12"),
      "org.apache.flink" %% "flink-connector-kafka" % flinkVersion,
      "org.apache.flink" %% "flink-runtime-web" % flinkVersion,
      "org.apache.flink" % "flink-avro-confluent-registry" % flinkVersion,
      "org.apache.storm" % "storm-core" % stormVersion % "provided" excludeAll(
        ExclusionRule("org.slf4j", "slf4j-log4j12"),
        ExclusionRule("org.slf4j", "log4j-over-slf4j")
        ),
      "org.apache.storm" % "storm-kafka-client" % stormVersion excludeAll("org.slf4j", "slf4j-log4j12"),
      "org.apache.storm" % "storm-sql-core" % stormVersion excludeAll("org.slf4j", "slf4j-log4j12"),
      "org.apache.storm" % "storm-sql-runtime" % stormVersion excludeAll("org.slf4j", "slf4j-log4j12"),
      "org.apache.storm" % "storm-kafka" % "1.2.3" excludeAll("org.slf4j", "slf4j-log4j12")
    )

Answer 1

您使用了错误的 Kafka jar。您应该依赖org.apache.kafka:kafka-clients 而不是org.apache.kafka:kafka_2.xx，后者是Kafka 服务器端的jar。

对kafka/api/OffsetRequest的依赖来自storm-kafka，不应该使用。它使用的是旧的 Kafka 客户端 API，它不再存在于 Kafka 中。请改用storm-kafka-client。