无法将代理列表参数从Scala传递到Kafka：属性bootstrap.servers无效

Question

我需要从使用Scala和Spark的远程Kafka队列主题中消费消息。默认情况下，远程计算机上的Kafka端口设置为7072，而不是9092。此外，远程机器上有安装以下版本：

1. 卡夫卡0.10.1.0
2. 斯卡拉2.11

这意味着我应该通过经纪人列表（与该端口7072）从斯卡拉远程卡夫卡，因为否则它会尝试使用默认端口。

问题是根据日志，参数bootstrap.servers无法被远程机器识别。我还尝试将此参数重命名为metadata.broker.list,broker.list和listeners，但始终在日志中出现相同的错误Property bootstrap.servers is not valid，然后默认使用端口9092（并且显然不会消耗这些消息）。

在POM文件我用卡夫卡以下依赖和星火：

<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的2.10，而不是2.11。

这是我的Scala代码（它工作绝对没问题，如果我使用安装在亚马逊云我自己卡夫卡在那里我有EMR机（那里我已经用了卡夫卡的港口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) 

我正在阅读this Documentation，但看起来我所做的一切都是正确的。我应该使用其他一些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. 

更新＃2：

我发现了这个相关的话题。建议的解决方案说Fixed it by setting the property 'advertised.host.name' as instructed by the comments in the kafka configuration (config/server.properties)。我是否正确理解config/server.properties应该在安装有Kafka的远程机器上更改？

Kafka : How to connect kafka-console-consumer to fetch remote broker topic content?

Answer 1

我觉得我碰到了同样的问题，最近（EOFException类），原因是卡夫卡的版本不匹配。

如果我看这里https://mvnrepository.com/artifact/org.apache.spark/spark-streaming-kafka_2.10/1.6.2 kafka streaming版本的编译时间依赖性是0.8而您使用0.10。

就我所知0.9已经不兼容0.8。你可以尝试设置一个本地的0.8或0.9代理并尝试连接？