Kafka远程制作人-广告.收听者

Question

第一步-除了Kafka代理的标准属性之外，我还配置了以下内容：

listeners=SASL_PLAINTEXT://0.0.0.0:9092

advertised.listeners=SASL_PLAINTEXT://<hostname>:9092

我能够启动控制台使用者

kafka-console-consumer --new consumer --topic <topicname> --from-beginning --bootstrap-server <hostname>:9092 --consumer.config consumer.properties

我可以从集群内的另一台计算机上使用自定义生成器。《生产者属性》相关节选：

security.protocol=SASL_PLAINTEXT 

bootstrap.servers=<hostname>:9092

无法从远程计算机使用自定义生成器：

Exception org.apache.kafka.common.errors.TimeoutException: Batch containing 1 record(s) expired due to timeout while requesting metadata from brokers for <topicname>-<partition>

listeners=SASL_PLAINTEXT://0.0.0.0:9092
advertised.listeners=SASL_PLAINTEXT://<kafkaBrokerInternalIP>:9092

null

listeners=SASL_PLAINTEXT://0.0.0.0:9092
advertised.listeners=SASL_PLAINTEXT://<kafkaBrokerPublicIP>:9092

从我的消费者开始使用

kafka-console-consumer --new-consumer --topic <topicname> --from-beginning --bootstrap-server <hostname>:9092 --consumer.config consumer.properties

给了我一个警告，但我不认为这是对的...

WARN clients.NetworkClient: Error while fetching metadata with correlation id 1 : {<topicname>=LEADER_NOT_AVAILABLE}

从我的消费者开始使用

kafka-console-consumer --new-consumer --topic <topicname> --from-beginning --bootstrap-server <KafkaBrokerPublicIP>:9092 --consumer.config consumer.properties

INFO utils.AppInfoParser: Kafka version : 0.10.0-kafka-2.1.0
INFO utils.AppInfoParser: Kafka commitId : unknown

INFO internals.AbstractCoordinator: Discovered coordinator <hostname>:9092 (id: <someNumber> rack: null) for group console-consumer-<someNumber>.

WARN NetworkClient:600 - Error while fetching metadata with correlation id 0 : {<topicname>=LEADER_NOT_AVAILABLE}

org.apache.kafka.common.errors.TimeoutException: Failed to update metadata after 60000 ms.

NetworkClient:600 - Error while fetching metadata with correlation id 0 : {<topicname>=LEADER_NOT_AVAILABLE}

在过去的三天里，我一直在努力使它工作，但是我没有想法：/我的理解是广告.hostnames正是为了这个目的，但是要么是我做错了什么，要么是机器设置有问题。

任何提示都非常感谢！

Answer 1

我最近遇到了这个问题。在我的例子中，我启用了Kafka ACL，在通过注释this 2配置禁用它之后，问题得以解决。

authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
super.users=User:kafka

我想一个线程可能会对您有所帮助:https://gist.github.com/jorisdevrede/a7933a99251452bb1867

它在最后提到：