无法将broker list参数从Scala传递到Kafka：属性Bootstrap.Servers无效

Question

null

这意味着我应该将代理列表（端口7072)从Scala传递到远程Kafka，因为否则它将尝试使用默认端口。

问题是，根据日志，远程计算机无法识别参数bootstrap.servers。我还尝试将此参数重命名为metadata.broker.list、broker.list和listeners，但日志中始终出现相同的错误属性bootstrap.servers无效，然后默认使用端口9092（显然不使用消息）。

在POM文件中，我对Kafka和Spark使用了以下依赖关系：

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-streaming_2.10</artifactId>
    <version>1.6.2</version>
</dependency>

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-streaming-kafka_2.10</artifactId>
    <version>1.6.2</version>
</dependency>

这是我的Scala代码（如果我使用自己安装在Amazon Cloud中的Kafka，在那里我有EMR机器（在那里我有用于Kafka的端口9092))，它绝对可以工作）：

    val testTopicMap = testTopic.split(",").map((_, kafkaNumThreads.toInt)).toMap

   val kafkaParams = Map[String, String](
      "broker.list" -> "XXX.XX.XXX.XX:7072",
      "zookeeper.connect" -> "XXX.XX.XXX.XX:2181",
      "group.id" -> "test",
      "zookeeper.connection.timeout.ms" -> "10000",
      "auto.offset.reset" -> "smallest")

    val testEvents: DStream[String] =
      KafkaUtils
        .createStream[String, String, StringDecoder, StringDecoder](
        ssc,
        kafkaParams,
        testTopicMap,
        StorageLevel.MEMORY_AND_DISK_SER_2
      ).map(_._2)

我正在阅读这篇文档，但看起来我所做的一切都是正确的。我应该使用其他一些Kafka客户端API（其他Maven依赖项）吗？

更新1：

val testTopicMap = testTopic.split(",").toSet
val kafkaParams = Map[String, String]("metadata.broker.list" -> "XXX.XX.XXX.XX:7072,XXX.XX.XXX.XX:7072,XXX.XX.XXX.XX:7072","bootstrap.servers" -> "XXX.XX.XXX.XX:7072,XXX.XX.XXX.XX:7072,XXX.XX.XXX.XX:7072",
                                      "auto.offset.reset" -> "smallest")
val testEvents = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, testTopicMap).map(_._2)

testEvents.print()

17/01/02 12:23:15 ERROR ApplicationMaster: User class threw exception: org.apache.spark.SparkException: java.io.EOFException: Received -1 when reading from channel, socket has likely been closed.
java.io.EOFException: Received -1 when reading from channel, socket has likely been closed.
java.io.EOFException: Received -1 when reading from channel, socket has likely been closed.

Answer 1

我想我最近遇到了同样的问题(EOFException)，原因是Kafka版本不匹配。

如果我看一下https://mvnrepository.com/artifact/org.apache.spark/spark-streaming-kafka2.10/1.6.2，那么kafka流版本的编译时依赖关系是0.8，而您使用的是0.10。

据我所知，0.9已经与0.8不兼容了。您可以尝试设置本地0.8或0.9代理并尝试连接吗？