flinkFlink常见Checkpoint超时问题排查思路

Posted 2021-12-16 九师兄

1.概述

转载：Flink常见Checkpoint超时问题排查思路 这里仅仅是自己学习。

在日常flink应用中，相信大家经常会遇到checkpoint超时失败这类的问题，遇到这种情况的时候仅仅只会在jobmanager处打一个超时abort的日志，往往一脸懵逼不知道时间花在什么地方了，本文就基于flink1.4.2版本理一下checkpoint出现超时问题的排查思路

2.超时判断逻辑

jobmanager定时trigger checkpoint，给source处发送trigger信号，同时会启动一个异步线程，在checkpoint timeout时长之后停止本轮checkpoint，cancel动作执行之后本轮的checkpoint就为超时，如果在超时之前收到了最后一个sink算子的ack信号，那么checkpoint就是成功的。

那么超时的原因会是什么呢？主要是一下两种:

Barrier对齐
异步状态遍历和写hdfs

第二种类型会很好看，因为状态很大的时候就会出现这个现象，下面分析第一类出现的原因。

3.Barrier处理流程

3.1 Barrier处理流程

StreamTask收集到相应的inputChannel的barrier，收集齐之后就将barrier下发，并开始自己task的checkpoint逻辑，如果上下游是rescale或者forward的形式，下游只需要等待1个并发的barrier，因为是point-to-point的形式，如果是hash或者rebalance，下游的每一个task开始checkpoint的前提就是要收集齐上游所有并发的barrier。

3.2 Barrier发送流程

3.2.1 RecordWriter#broadcastEvent

这个方法是专门用于barrier的下发，首先会将serializer中还没下发的部分已经写入数据的buffer下发防止barrier越过数据下发的行为，保证一致性，同时会将barrier包装成buffer对象，此处不申请堆外内存，直接将堆内内存包装成一个buffer下发。

这一点和【Flink】Flink 1.14 版本 新特性 Barrier 在流经算子做 checkpoint Barrier跳过 unaligned checkpoint 恰恰相反呀。

3.2.2 PipelinedSubpartition#add

将要发送给下游的数据add到subpartition中，同时通知监听的subpartitionView，notifybufferAvailable事件，这里notify操作会区分本地和远程两种channel不同处理，我们看到的都是远程消费的延迟问题，本地其实就立马执行下游的barrier收集动作了，远程的需要有网络传输过程。

3.2.3 server netty handler

LengthFieldBasedFrameDecoder => messageDecode => PartitionRequestServerHandler => PartitionRequestQueue => messageEncoder notifybufferAvailable事件最终会触发PartitionRequestQueue去将数据writeAndFlush到netty client中，在flush之前会判断channel是否可写，在flush成功后，会执行相应的barrier的处理逻辑，这里可以hack代码使其在发送barrier的时候,监听flush成功回调打印barrier发送成功的消息，使我们确认barrier已经从上游发送成功。

3.3 barrie接收流程

LengthFieldBasedFrameDecoder => messageDecode => PartitionRequestClientHandler => messageEncoder
首先读入消息时候会判断是否有堆积的stagedMessages，如果有则不处理，add到堆积消息中，如果没有将数据从netty buffer中拷贝至localbuffer中，这时候需要requestBuffer，这个方法并不block消费流程，但是如果request不到buffer那么会将数据丢入stagedMessages，同时监听bufferpool，等到buffer有recyle的时候就会开始buffer的转化，并且这个是时候会将channel的auto read标志置为false，因此这个通道就不再读入数据，barrier也是无法读入的，并且每个taskmanager共享一个channel，因此taskmanager上只要阻塞了就会影响这个taskmanager上的消费。

4.小结

从以上可以看出，其实barrier下游无法对齐的主要原因还是在于下游消费能力不足，会导致buffer堆积一段时间，但这时并不足以造成上游反压，因为反压需要下游channel持续无法写入，导致tcp阻塞，导致上游的outputbuffer占满才会引起反压。

一般的排查方式

查看ui上的失败checkpoint的detail，可以看到失败的是Pending -> EventEmit xxx这个算子的10个subtask。

一般由于barrier对齐原因没开始的subtask会是开始时间都是n/a，因为在汇报CheckpointStats的时候根本还没开始checkpoint，注意这里的11，12并不是subtaskIndex，这里千万不要被误导了去查看这个subtask的问题。

接着去jobmanager上查看这个checkpoint的一些延迟信息

2018-11-29 18:43:04,624 INFO  org.apache.flink.runtime.checkpoint.CheckpointCoordinator     - Checkpoint 224 expired before completing.
2018-11-29 18:43:26,763 WARN  org.apache.flink.runtime.checkpoint.CheckpointCoordinator     - Received late message for now expired checkpoint attempt 224 from f2862289958b430bc3dc20f39794ca2c of job 7c7769847d333438dd9ce845d5a2d980.
2018-11-29 18:43:26,766 WARN  org.apache.flink.runtime.checkpoint.CheckpointCoordinator     - Received late message for now expired checkpoint attempt 224 from 8f569166274106f22e49ed2ce919c930 of job 7c7769847d333438dd9ce845d5a2d980.
2018-11-29 18:43:26,770 WARN  org.apache.flink.runtime.checkpoint.CheckpointCoordinator     - Received late message for now expired checkpoint attempt 224 from a29e34c210b39104004af7f067c1a5d0 of job 7c7769847d333438dd9ce845d5a2d980.
2018-11-29 18:43:26,771 WARN  org.apache.flink.runtime.checkpoint.CheckpointCoordinator     - Received late message for now expired checkpoint attempt 224 from 7d4914521fd53fca56a4050d6f191ae9 of job 7c7769847d333438dd9ce845d5a2d980.
2018-11-29 18:43:26,771 WARN  org.apache.flink.runtime.checkpoint.CheckpointCoordinator     - Received late message for now expired checkpoint attempt 224 from 618c78d0008d0d525728ff9824339229 of job 7c7769847d333438dd9ce845d5a2d980.
2018-11-29 18:43:26,773 WARN  org.apache.flink.runtime.checkpoint.CheckpointCoordinator     - Received late message for now expired checkpoint attempt 224 from c8ba24a328234dc7f2f271db4a8eb1e3 of job 7c7769847d333438dd9ce845d5a2d980.
2018-11-29 18:43:26,777 WARN  org.apache.flink.runtime.checkpoint.CheckpointCoordinator     - Received late message for now expired checkpoint attempt 224 from 72af6c722fcc085dc8f7c46e9124d82e of job 7c7769847d333438dd9ce845d5a2d980.
2018-11-29 18:43:26,777 WARN  org.apache.flink.runtime.checkpoint.CheckpointCoordinator     - Received late message for now expired checkpoint attempt 224 from f824cb6920b04d19e05278ee362ec675 of job 7c7769847d333438dd9ce845d5a2d980.
2018-11-29 18:43:26,780 WARN  org.apache.flink.runtime.checkpoint.CheckpointCoordinator     - Received late message for now expired checkpoint attempt 224 from af6d867d2f12be23c7b23a938aba7c5e of job 7c7769847d333438dd9ce845d5a2d980.
2018-11-29 18:43:27,265 WARN  org.apache.flink.runtime.checkpoint.CheckpointCoordinator     - Received late message for now expired checkpoint attempt 224 from cee0205fe9a85e3e89e023a1166ed1e6 of job 7c7769847d333438dd9ce845d5a2d980.
2018-11-29 18:43:44,624 INFO  org.apache.flink.runtime.checkpoint.CheckpointCoordinator     - Triggering checkpoint 225 @ 1543488224622

可以根据这些失败的task的id去查询这些任务落在哪一个taskmanager上，经过排查发现，是同一台机器，通过ui看到该机器流入的数据明显比别的流入量大因此是因为数据倾斜导致了这个问题，追根溯源还是下游消费能力不足的问题

4.1 指标

可以反映一个算子的指标有几个

inPoolUsage
OutPoolUsage
OutputQueueLength
inputQueueLength

首先前3个值都是未加锁获得的非准确数据，为了数据消费本身的性能，参考意义不大。而inputQueueLength可以加以参考，为什么这个加锁了，可能是社区遗漏了。。这个值最大应该是（分区如果是hash，其他分区方式会更小一点）上游并发 * 2 + 8 + 上游并发，如果到达这个值左右，此时发送barrier到下游会无法反序列化并进行正确的checkpoint操作，至于为什么最后一个上游并发单独拎开，是因为这个含义表示的barrier数量，barrier的数量也会算在inputQueueLength内。

在flink1.5之后的@zhijiangW网络栈优化后的版本中据说checkpointBarrier是可以越过数据优先发送，需要确认这种情况下如何保障处理的exactly once语义呢？

这一点请看【Flink】Flink 1.14 版本 新特性 Barrier 在流经算子做 checkpoint Barrier跳过 unaligned checkpoint