带有URI的Flink、Kafka和Zookeeper

Question

我正在尝试从本地机器连接到Kafka：

kafkaParams.setProperty("bootstrap.servers", Defaults.BROKER_URL)
kafkaParams.setProperty("metadata.broker.list", Defaults.BROKER_URL)
kafkaParams.setProperty("group.id", "group_id")
kafkaParams.setProperty("auto.offset.reset", "earliest")

非常好，但是我的BROKER_URI定义如下my-server.com:1234/my/subdirectory。

我发现这种现象被称为chroot路径。

它抛出以下错误：Caused by: org.apache.kafka.common.config.ConfigException: Invalid url in bootstrap.servers: my-server.com:1234/my/subdirectory

我该如何解决这个问题？

以下是我的依赖项：

val flinkVersion = "1.0.3"

"org.apache.flink" %% "flink-scala" % flinkVersion % "provided",
"org.apache.flink" %% "flink-streaming-scala" % flinkVersion % "provided",
"org.apache.flink" %% "flink-connector-kafka-0.9" % flinkVersion,

Answer 1

只需尝试不带路径上下文和斜杠的host:port格式。如果您有多个服务器，它将是一个列表host1:port1,host2:port2

参考：http://kafka.apache.org/documentation.html

Answer 2

bootstrap.servers应该是一个逗号分隔的列表，如下所示：address1:port1,address2:port2,...,addressn:portn。如果你只有一个Kafka代理，你应该输入像localhost:9092这样的东西(除非你配置Kafka在另一个端口上运行)。

有关如何让Flink和Kafka协同工作的更多详细信息，您可以参考this post from dataArtisans。

Answer 3

太愚蠢了。动物园管理员!=卡夫卡。正如您在代码中看到的，我使用了两次相同的URL，但事实证明它们应该是不同的。

我正在尝试从本地机器连接到Kafka：

kafkaParams.setProperty("bootstrap.servers", Defaults.KAFKA_URL)
kafkaParams.setProperty("metadata.broker.list", Defaults.ZOOKEEPER_URL)
kafkaParams.setProperty("group.id", "group_id")
kafkaParams.setProperty("auto.offset.reset", "earliest")