如何使用动态 DNS 设置公共 Kafka 代理？

Question

我配置了一个包含 3 个代理的 Kafka 集群，每个代理使用 3 个 Zookeeper。下图展示了我的集群的图形表示。

使用主机192.168.0.10 在同一网络中的生产者和消费者测试通过kafka-console-producer 和kafka-console-consumer 命令完美运行。

基于该上下文，当我尝试通过 Internet 通过kafka-console-producer.sh --broker-list DYNAMIC_DNS_ADDR:30192,DYNAMIC_DNS_ADDR:30292,DYNAMIC_DNS_ADDR:30392 --topic twitter_tweets 生成一些数据时，我收到以下错误：

[2018-12-10 09:59:20,772] ERROR 向 twitter_tweets 主题发送消息时出错，键为空，值：16 字节，错误：(org.apache.kafka.clients.producer.internals.ErrorLoggingCallback) org.apache.kafka.common.errors.TimeoutException：twitter_tweets-1 的 1 条记录到期：自批处理创建以来已过去 1505 毫秒加上延迟时间 [2018-12-10 09:59:22,273] WARN [Producer clientId=console-producer] 无法建立到节点 1 的连接。经纪人可能不可用。 (org.apache.kafka.clients.NetworkClient)

代理侦听器配置有以下属性：

listeners=PLAINTEXT://0.0.0.0:9092,SSL://0.0.0.0:9443
advertised.listeners=PLAINTEXT://192.168.0.241:9092,SSL://192.168.0.241:9443

显然，advertised.listeners 属性的每个代理中的 IP 地址都发生了变化。我正在使用CentOS 6.10 和Kafka 2.0.1 进行该设置。远程登录测试有效。另一个到 Kafka REST 代理端口的转发是通过 Internet 工作并列出所有主题。

Answer 1

见https://rmoff.net/2018/08/02/kafka-listeners-explained/

您需要两个监听器——一个响应和广告内部地址，一个用于外部地址。

关键是您的客户端连接到的侦听器将返回该侦听器的主机地址和端口。

目前，您正在将外部流量欺骗到内部流量，而您的外部流量因此会击中内部侦听器。

您需要这样的东西（根据每个代理的要求改变aws_internal_listener 的 IP/主机名）：

KAFKA_LISTENERS: aws_internal_listener://192.168.0.241:9092,external_listener://192.168.0.241:29092

KAFKA_ADVERTISED_LISTENERS: aws_internal_listener://192.168.0.241:9092,external_listener://DYNAMIC_DNS_ADDR:29092

KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: aws_internal_listener:PLAINTEXT,external_listener:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: aws_internal_listener

然后您的 DYNAMIC_DNS_ADDR 端口转发器应将连接重定向到 AWS 节点上的 29092。关键是 external 连接不应在与 internal 侦听器匹配的主机上的侦听器端口结束（它通告内部192.168.0 地址）

使用kafkacat -L -b DYNAMIC_DNS_ADDR:29092 调试和验证您的配置，如in the article here 所述。