前言

呵呵也是最近有一些搭建 kafka 的环境的需求

然后从新看了一下一部分的配置情况, 这里大致理一下一些我这里比较关心的配置

那些配置关联了 kafka 服务器绑定服务

绑定 tcp 服务的配置来自于这里, 读取的是 config.dataPlaneListeners

config.dataPlaneListeners 是取自 listeners, 然后过滤掉了 controller 相关的配置

listeners 的配置来自于 "listeners" / "host.name + port" 的配置

那些配置关联了 kafka 服务器暴露给客户端的地址信息

和如下问题关联, 暴露给客户端的地址信息只要是来自于 advertisedListeners 相关

当前机器的虚拟机[NAT网络]映射出 9200 到宿主机, 局域网的其他机器访问宿主机+9200 访问不到 kafka 的服务

引用自上面的链接

FindCoordinatorRequest 获取的数据来自于 metadataSnapshot
kafkaController.sendUpdateMetadataRequest 发生事件的时候更新 metadataSnapshot 的请求
aliveNodes 来自于 broker 的相关信息, 这个 brokerInfo 来自于 KafkaController.initializeControllerContext
KafkaController.initializeControllerContext 的 aliveNodes 来自于 zkClient 向 "/brokers/ids/*" 获取 brokerId 列表, 然后在依次获取 "brokers/ids/$id" 的数据信息, 作为 Broker 的元数据信息
broker 注册到 zk 的信息来自于 KafkaServer 启动的时候向 zk 注册的, endpoints 的相关信息来自于 "advertised.listeners", "advertised.host.name:advertised.host.port", "listeners", "host.name:port"

具体的获取 advertisedListeners 相关的处理如下

kafka 启动之后还会连接 broker ?

可以发现一个问题, 就是假设我 advertised.listeners 配置了一个存在问题的配置, ip 不存在, 或者服务不存在

服务器启动大致会报错如下

java.net.SocketTimeoutException: Failed to connect within 30000 ms
	at kafka.controller.RequestSendThread.brokerReady(ControllerChannelManager.scala:297)
	at kafka.controller.RequestSendThread.doWork(ControllerChannelManager.scala:250)
	at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:96)

这个错误是什么呢?, 影响到的又有哪些呢?

这个请求的处理, 主要是来自于 controller

kafkaServer - kafkaController - controllerChannelManager - requestSendThread, 再到具体的 requestSendThread 就是最原始导致上面报错的地方了,

在各种事件触发的时候, controller 需要连接 broker, 发送请求[由 requestSendThread 来发送请求], 比如 RegisterBrokerAndReelect 的时候, 需要选举 controller, 之后需要发送 UpdateMetadataRequest

然后客户端尝试连接 kafka 服务, 会得到报错大致如下

[10:44:05.239] WARN  org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater 1023 handleCompletedMetadataResponse - [Consumer clientId=consumer-1, groupId=test3] Error while fetching metadata with correlation id 29 : {test20220528=LEADER_NOT_AVAILABLE}
[10:44:05.365] WARN  org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater 1023 handleCompletedMetadataResponse - [Consumer clientId=consumer-1, groupId=test3] Error while fetching metadata with correlation id 30 : {test20220528=LEADER_NOT_AVAILABLE}
[10:44:05.484] WARN  org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater 1023 handleCompletedMetadataResponse - [Consumer clientId=consumer-1, groupId=test3] Error while fetching metadata with correlation id 31 : {test20220528=LEADER_NOT_AVAILABLE}
[10:44:05.604] WARN  org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater 1023 handleCompletedMetadataResponse - [Consumer clientId=consumer-1, groupId=test3] Error while fetching metadata with correlation id 32 : {test20220528=LEADER_NOT_AVAILABLE}
[10:44:05.721] WARN  org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater 1023 handleCompletedMetadataResponse - [Consumer clientId=consumer-1, groupId=test3] Error while fetching metadata with correlation id 33 : {test20220528=LEADER_NOT_AVAILABLE}
[10:44:05.840] WARN  org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater 1023 handleCompletedMetadataResponse - [Consumer clientId=consumer-1, groupId=test3] Error while fetching metadata with correlation id 34 : {test20220528=LEADER_NOT_AVAILABLE}
[10:44:05.953] WARN  org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater 1023 handleCompletedMetadataResponse - [Consumer clientId=consumer-1, groupId=test3] Error while fetching metadata with correlation id 35 : {test20220528=LEADER_NOT_AVAILABLE}
[10:44:06.065] WARN  org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater 1023 handleCompletedMetadataResponse - [Consumer clientId=consumer-1, groupId=test3] Error while fetching metadata with correlation id 36 : {test20220528=LEADER_NOT_AVAILABLE}
[10:44:06.187] WARN  org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater 1023 handleCompletedMetadataResponse - [Consumer clientId=consumer-1, groupId=test3] Error while fetching metadata with correlation id 37 : {test20220528=LEADER_NOT_AVAILABLE}