在日常flink应用中，相信大家经常会遇到checkpoint超时失败这类的问题，遇到这种情况的时候仅仅只会在jobmanager处打一个超时abort的日志，往往一脸懵逼不知道时间花在什么地方了，本文就基于flink1.4.2版本理一下checkpoint出现超时问题的排查思路
超时判断逻辑
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成功后，会执行相应的barrier的处理逻辑，这里可以hack代码使其在发送barrier的时候,监听flush成功回调打印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。
WX20181129-202331.png
一般由于barrier对齐原因没开始的subtask会是开始时间都是n/a，因为在汇报CheckpointStats的时候根本还没开始checkpoint，注意这里的11，12
并不是subtaskIndex，这里千万不要被误导了去查看这个subtask的问题。
WX20181129-201928.png
接着去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
语义呢？
网友评论