通过 Java 创建 Kafka 主题时出错

Question

我已尝试通过 Java 在 Kafka 中创建主题，如回答此问题所述： How Can we create a topic in Kafka from the IDE using API

我收到以下异常：

kafka.common.KafkaException: Failed to parse the broker info from zookeeper: {"jmx_port":-1,"timestamp":"1495085433520","endpoints":["PLAINTEXT://0.0.0.0:9092"],"host":"0.0.0.0","version":3,"port":9092}
        at kafka.cluster.Broker$.createBroker(Broker.scala:125)
        at kafka.utils.ZkUtils.getBrokerInfo(ZkUtils.scala:799)
        at kafka.utils.ZkUtils$$anonfun$getAllBrokersInCluster$2.apply(ZkUtils.scala:253)
        at kafka.utils.ZkUtils$$anonfun$getAllBrokersInCluster$2.apply(ZkUtils.scala:253)
        at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
        at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
        at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
        at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
        at scala.collection.TraversableLike$class.map(TraversableLike.scala:244)
        at scala.collection.AbstractTraversable.map(Traversable.scala:105)
        at kafka.utils.ZkUtils.getAllBrokersInCluster(ZkUtils.scala:253)
        at kafka.admin.AdminUtils$.getBrokerMetadatas(AdminUtils.scala:393)
        at kafka.admin.AdminUtils$.createTopic(AdminUtils.scala:415)
        at kafka.admin.AdminUtils.createTopic(AdminUtils.scala)
        at com.freecharge.payments.kafka.Test.main(Test.java:41)
Caused by: java.lang.NoSuchMethodError: scala.Predef$.ArrowAssoc(Ljava/lang/Object;)Ljava/lang/Object;
        at kafka.cluster.EndPoint$$anonfun$1.apply(EndPoint.scala:32)
        at kafka.cluster.EndPoint$$anonfun$1.apply(EndPoint.scala:32)
        at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
        at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
        at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
        at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:108)
        at scala.collection.TraversableLike$class.map(TraversableLike.scala:244)
        at scala.collection.mutable.ArrayOps$ofRef.map(ArrayOps.scala:108)
        at kafka.cluster.EndPoint$.<init>(EndPoint.scala:32)
        at kafka.cluster.EndPoint$.<clinit>(EndPoint.scala)
        at kafka.cluster.Broker$$anonfun$2.apply(Broker.scala:115)
        at kafka.cluster.Broker$$anonfun$2.apply(Broker.scala:115)
        at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
        at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
        at scala.collection.immutable.List.foreach(List.scala:318)
        at scala.collection.TraversableLike$class.map(TraversableLike.scala:244)
        at scala.collection.AbstractTraversable.map(Traversable.scala:105)
        at kafka.cluster.Broker$.createBroker(Broker.scala:115)

我已经坚持了大约 3 天，请帮忙。

Maven 配置是：

<dependency>
    <groupId>io.confluent</groupId>
    <artifactId>common-config</artifactId>
    <version>3.0.0</version>
</dependency>
<dependency>
    <groupId>io.confluent</groupId>
    <artifactId>common-utils</artifactId>
    <version>3.0.0</version>
</dependency>
<dependency>
    <groupId>org.codehaus.janino</groupId>
    <artifactId>janino</artifactId>
    <version>2.6.1</version>
</dependency>
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka_2.11</artifactId>
    <version>0.10.2.0</version>
</dependency>
<dependency>
    <groupId>com.101tec</groupId>
    <artifactId>zkclient</artifactId>
    <version>0.10</version>
</dependency>
<dependency>
    <groupId>io.confluent</groupId>
    <artifactId>kafka-schema-registry-client</artifactId>
    <version>3.0.0</version>
</dependency>
<dependency>
    <groupId>io.confluent</groupId>
    <artifactId>kafka-avro-serializer</artifactId>
    <version>3.0.0</version>
</dependency>

Answer 1

Zookeeper 客户端和 Kafka 的版本很可能不兼容。尝试使用zkclient 0.10版本：

<!-- https://mvnrepository.com/artifact/com.101tec/zkclient -->
<dependency>
    <groupId>com.101tec</groupId>
    <artifactId>zkclient</artifactId>
    <version>0.10</version>
</dependency>