Checkpoint 失败之Checkpoint Expire应如何解决？

如果 Checkpoint 做的非常慢，超过了 timeout 还没有完成，则整个 Checkpoint 也会失败。当一个 Checkpoint 由于超时而失败是，会在 jobmanager.log 中看到如下的日志：

Checkpoint 1 of job 85d268e6fbc19411185f7e4868a44178 expired before completing.

表示 Chekpoint 1 由于超时而失败，这个时候可以可以看这个日志后面是否有类似下面的日志：

Received late message for now expired checkpoint attempt 1 from
0b60f08bf8984085b59f8d9bc74ce2e1 of job 85d268e6fbc19411185f7e4868a44178.

我们按照下面的日志把 TM 端的 snapshot 分为三个阶段，开始做 snapshot 前，同步阶段，异步阶段：

DEBUG
Starting checkpoint (6751) CHECKPOINT on task taskNameWithSubtasks (4/4)

这个日志表示 TM 端 barrier 对齐后，准备开始做 Checkpoint。

DEBUG
2019-08-06 13:43:02,613 DEBUG org.apache.flink.runtime.state.
AbstractSnapshotStrategy -
DefaultOperatorStateBackend snapshot (FsCheckpointStorageLocation
{fileSystem=org.
apache.flink.core.fs.SafetyNetWrapperFileSystem@70442baf,
checkpointDirectory=xxxxxxxx,
sharedStateDirectory=xxxxxxxx, taskOwnedStateDirectory=xxxxxx,
metadataFilePath=xxxxxx,
reference=(default), fileStateSizeThreshold=1024}, synchronous part) in
thread Thread[Async calls on
Source: xxxxxx
_source -> Filter (27/70),5,Flink Task Threads] took 0 ms.

上面的日志表示当前这个 backend 的同步阶段完成，共使用了 0 ms。

DEBUG
DefaultOperatorStateBackend snapshot (FsCheckpointStorageLocation
{fileSystem=org.
apache.flink.core.fs.SafetyNetWrapperFileSystem@7908affe,
checkpointDirectory=xxxxxx,
sharedStateDirectory=xxxxx, taskOwnedStateDirectory=xxxxx,
metadataFilePath=xxxxxx,
reference=(default), fileStateSizeThreshold=1024}, asynchronous part) in
thread Thread[pool-48-
thread-14,5,Flink Task Threads] took 369 ms

上面的日志表示异步阶段完成，异步阶段使用了 369 ms 在现有的日志情况下，我们通过上面三个日志，定位 snapshot 是开始晚，同步 阶段做的慢，还是异步阶段做的慢。然后再按照情况继续进一步排查问题。