您的位置: 首页  >  文章  >  kafka问题排查之 Java代码不进行消费

kafka问题排查之 Java代码不进行消费

分类: 文章 • 2024-10-30 16:53:28

发现问题

使用 kafka 在linux系统，通过命令测试消费正常, 但在Java 代码无法正常接收队列消息

控制台提示信息：

15:21:33.804 [main] INFO  org.apache.coyote.http11.Http11NioProtocol - Initializing ProtocolHandler ["http-nio-8886"]
15:21:33.836 [main] INFO  org.apache.coyote.http11.Http11NioProtocol - Starting ProtocolHandler ["http-nio-8886"]
15:21:33.841 [main] INFO  org.apache.tomcat.util.net.NioSelectorPool - Using a shared selector for servlet write/read
15:21:33.859 [concurrentMessageListenerContainer-0-kafka-consumer-1] INFO  org.apache.kafka.clients.consumer.internals.AbstractCoordinator - Discovered coordinator platform-010-030-050-167:16792 (id: 2147483647 rack: null) for group test-mengqa.
15:21:33.878 [main] INFO  org.springframework.boot.context.embedded.tomcat.TomcatEmbeddedServletContainer - Tomcat started on port(s): 8886 (http)
15:21:33.884 [main] INFO  com.navinfo.opentsp.qingqi.webpush.core.Application - Started Application in 49.808 seconds (JVM running for 53.243)
15:21:34.864 [concurrentMessageListenerContainer-0-kafka-consumer-1] INFO  org.apache.kafka.clients.consumer.internals.AbstractCoordinator - Marking the coordinator platform-010-030-050-167:16792 (id: 2147483647 rack: null) dead for group test-mengqa

最后一行 Marking the coordinator platform-010-030-050-167:16792 (id: 2147483647 rack: null) dead for group test-mengqa
被标记为死亡, 不能接收消息的原因可能就是消费者死亡导致的

分析过程

从 log 可以分析这句是 AbstractCoordinator 类打印的, 我们找到打印这行信息的代码

protected void coordinatorDead() {
    if (this.coordinator != null) {
         log.info("Marking the coordinator {} dead.", this.coordinator.id());
         this.coordinator = null;
     }
 }

显然coordinator 为null了
打断点追踪

kafka问题排查之 Java代码不进行消费

platform-010-030-050-167 是 Kafka 实例所在服务器的主机名,
16792 是 kafka 的端口,这玩意好像是 Kafka 的连接地址
乍一看, 觉得没什么问题

从堆栈追踪到上一调用方法处看看：
kafka问题排查之 Java代码不进行消费

既然 coordinator 不为空, 那进入代码块一定是因为 client.connectionFailed(coordinator) 这个方法的返回结果为true了。
从语义分析是因为客户端连接 coordinator 失败

进入该方法看看：
kafka问题排查之 Java代码不进行消费

看到形参名是 node, 这个对象应该就是 kafka 的节点信息,
点开查看一下对象的具体属性，知道问题原因在此。

问题定位

注意看上图的 host 属性, host 的意思一般是主机.
局域网内, 通过主机名是无法访问的。
一般是通过 IP 、域名、或者修改 hosts 文件把主机名和 IP 对应起来
定位后,我们尝试用最简单的方法解决问题.

解决

最简单的就是修改本机 hosts 文件

windows 系统 hosts 文件位于 C:\Windows\System32\drivers\etc\hosts
使用管理员权限打开, 追加 IP 和 主机名对应关系

10.30.50.167 platform-010-030-050-167

重启解决之。

方案总结

可能因为没有给 Kafka 设置监听地址导致的默认监听主机名

在配置中果然搜索到类似选项, 按照注释的意思就是会广播给消费者和生产者的地址.
我们按照要求改成 advertised.listeners=PLAINTEXT://10.30.50.167:16792
恢复本机 hosts 文件经测试同样解决了问题

最后

我们在 application.properties 中已经指定 spring.kafka.bootstrap-servers 为 IP, 为什么还会使用主机名链接呢?

推测客户端是先连接到 Kafka 实例后会从 zk 中获取配置
然后客户端 watch zk 节点得到配置地址后才开始监听队列。

相关推荐