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

本文大致理一下checkpoint出现超时问题的排查思路：（本文基于flink-1.4.2）

超时判断逻辑

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

那么超时的原因会是什么呢？ 时间要么花费在barrier对齐，要么花费在异步状态遍历和写hdfs 第二种类型会很好看，因为状态很大的时候就会出现这 个现象，下面分析第一类出现的原因

Barrier处理流程

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

Barrier发送流程

RecordWriter#broadcastEvent

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

PipelinedSubpartition#add

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

server netty handler

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

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上的消费。

小结

从以上可以看出，其实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看到该机器流入的数据明显比别的流入量大 因此是因为数据倾斜导致了这个问题，追根溯源还是下游消费能力不足的问题

指标

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

• inPoolUsage
• OutPoolUsage
• OutputQueueLength
• inputQueueLength

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

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