记一次Flink任务无限期INITIALIZING排查过程

1.前言

环境：Flink-1.16.1，部署模式：Flink On YARN，现象：Flink程序能正常提交到 YARN，Job状态是 RUNNING，而 Task状态一直处于 INITIALIZING，如下图：
在这里插入图片描述

通过界面可以看到 flink任务一直没有执行，处于初始化状态，观察到初始化状态最长可持续 10几个小时。

补充：Flink On YARN模式如何打开 Flink任务监控页面。首先，打开 YARN任务管理界面（默认 8088端口），找到运行中的 flink任务，点击下图中的 ApplicationMaster即可。
在这里插入图片描述

2.通过大模型获取问题排查思路

笔者：Source: kafkaReader_3 -> router_4 -> (Sink: rdbWriter_6, Sink: kafkaWriter_8) 状态 INITIALIZING的原因

GPT：

在 Flink 中，INITIALIZING 状态表示作业的一个任务（task）正在初始化。具体到你提供的例子，Source: kafkaReader_3 -> router_4 -> (Sink: rdbWriter_6, Sink: kafkaWriter_8) 是一个任务链，包括一个 Kafka 数据源（kafkaReader_3）、一个数据处理算子（router_4），以及两个数据汇（Sink: rdbWriter_6 和 Sink: kafkaWriter_8）。当任务启动时，它会经历一系列的状态转换，其中之一就是 INITIALIZING。这个状态表示任务正在进行初始化操作，例如创建连接、加载配置、准备资源等。在初始化完成后，任务会进入 RUNNING 状态，开始正常的数据处理和计算。需要注意的是，INITIALIZING 状态只是任务的一个瞬时状态，表示任务正在初始化过程中。如果任务长时间停留在 INITIALIZING 状态，可能存在初始化过程出现问题的情况，例如资源不足、连接失败等。在正常情况下，任务应该尽快从 INITIALIZING 状态转移到 RUNNING 状态以继续执行数据处理逻辑。因此，如果你的任务一直停留在 INITIALIZING 状态，建议检查相关的日志以了解初始化过程中是否发生了错误或异常。同时，还可以检查资源分配、网络连接等因素，以确保任务能够成功初始化并转移到 RUNNING 状态。

3.检查 YARN资源

打开 YARN任务监控页面，左侧 Nodes页面可以查看集群可用的计算节点个数，Scheduler页面可以看到队列资源占用情况，以及CPU和内存占用情况。

4.查看 JobManager和 TaskManager的日志

查看日志有两种方式：

通过 flink任务监控页面

通过登录计算节点所在服务器

ll /home/yarn_data/log/application_1708676775521_0024/container_e07_1708676775521_0024_01_000002-rw-r--r-- 1 yarn ocdp  8907 Mar  7 11:41 directory.info
-rw-r--r-- 1 yarn ocdp 18796 Mar  7 15:42 gc.log.0.current
-rw-r----- 1 yarn ocdp 15772 Mar  7 11:41 launch_container.sh
-rw-r--r-- 1 yarn ocdp     0 Mar  7 11:41 prelaunch.err
-rw-r--r-- 1 yarn ocdp   100 Mar  7 11:41 prelaunch.out
-rw-r--r-- 1 yarn ocdp  1270 Mar  7 11:41 taskmanager.err
-rw-r--r-- 1 yarn ocdp 71955 Mar  7 11:46 taskmanager.log
-rw-r--r-- 1 yarn ocdp 66686 Mar  7 11:46 taskmanager.out

5.查看 Flink程序堆栈信息

通过 flink任务监控页面查看 Thread Dump，如下图：

在这里插入图片描述

这里我把关键部分堆栈摘抄下来了：

{"threadName": "Source: kafkaReader_3 -> router_4 -> (Sink: rdbWriter_6, Sink: kafkaWriter_8) (1/1)#0","stringifiedThreadInfo": "\"Source: kafkaReader_3 -> router_4 -> (Sink: rdbWriter_6, Sink: kafkaWriter_8) (1/1)#0\" Id=72 WAITING on java.util.concurrent.CountDownLatch$Sync@6792895f\n\tat sun.misc.Unsafe.park(Native Method)\n\t-  waiting on java.util.concurrent.CountDownLatch$Sync@6792895f\n\tat java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)\n\tat java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:836)\n\tat java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:997)\n\tat java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304)\n\tat java.util.concurrent.CountDownLatch.await(CountDownLatch.java:231)\n\tat org.apache.kafka.clients.producer.internals.TransactionalRequestResult.await(TransactionalRequestResult.java:50)\n\tat org.apache.kafka.clients.producer.KafkaProducer.initTransactions(KafkaProducer.java:569)\n\t...\n\n"
}

这里主要就是看线程名称为任务名称的，比如这里就是：

Source: kafkaReader_3 -> router_4 -> (Sink: rdbWriter_6, Sink: kafkaWriter_8)

可以看到笔者编写的 flink程序阻塞的原因是：initTransactions，在初始化事务。