看了 BucketSink 的相关源码。着重看了它的checkpoint以及故障恢复机制。
把大概的理解梳理如下：
BucketSink 大体的工作流程：
1.新建一个文件，不断的写入文件中，后缀命名为 .in-progress
2.判断文件写入完毕，关闭该文件时，后缀名命名为 .pending
3.checkpoint触发时，将上次ck到这次ck间的所有 .pending 文件变为 finish 状态
BucketSink 实现了 CheckpointedFunction 接口
有两个方法
void snapshotState(FunctionSnapshotContext context) throws Exception;
与
void initializeState(FunctionInitializationContext context) throws Exception;�其中：
initializeState 方法在每次新建 BucketSink 或者故障恢复时 会调用。
snapshotState 在每次触发 ck 时会被调用。
下面简单分析下这两个方法的逻辑：
initializeState 方法主要执行一些初始化操作，其中我认为关键的在于

1.png

restoredBucketStates = stateStore.getSerializableListState("bucket-states");
该方法获取一个叫做 bucket-states 的状态对象，从名称也可知，该对象用于重启。正常情况下，该对象无内容下面的for语句不会执行。但是若有故障重启的情况，则会从上次的ck中读取出内容，也就是上次ck的状态信息，然后执行回滚操作保证数据的一致性。这一点最后再做介绍。

snapshotState 方法用于触发 ck 操作。
这个方法做了如下几件事
1.获取当前正在写的 .pending 文件的大小，以便若下次 ck 前发生故障，可以获知本次ck时，该文件的大小，以便删除本次ck后到故障发生时写入的数据，或者显示该文件的有效数据大小。
2.将所有 .pending 状态的文件存储到list中，稍后ck结束后，方便修改其状态为 finish
3.将当前状态存入 restoredBucketStates 对象，以便若下次 ck 前发生故障，可以从这个状态处进行恢复。

同时，BucketSink也实现了 CheckpointListener 接口
void notifyCheckpointComplete(long checkpointId) throws Exception;
该方法会在 ck 完成后调用。

2

该方法，将 .pending 文件的状态转为 final 状态
并且移除writer已经处于close状态的bucket。

最后详细说一下故障恢复。
当程序因故障自动恢复时，initializeState 方法的 restoredBucketStates 就会从上次 ck 中获取到上次ck时的状态。进而进行恢复。

3

首先，将 .pending 状态的文件名列表清空即可，因为将 .pending 状态转为 finish 状态，可以在 notifyCheckpointComplete 方法中完成。故障恢复时，该方法对 .pending 的文件的做法是不做处理，等待故障恢复之后，第一次ck触发时，便会自动的将 .pending 的文件变为 finish 状态。
而之所以不处理 .pending 状态文件，是因为 .pending 状态文件说明该文件已经写入完毕，就差ck成功后修改文件状态（也就是文件名）而已，本质上，该文件已经不再写入数据，没有数据的变化。
接下来 handlePendingInProgressFile 就是处理 .in-progress 状态的文件。
我们设想一下，故障重启是指在上次成功的ck之后，下次ck之前，发生了故障，然后应用自动重启，使用的是上次成功的ck的状态信息。
这样的话，上次 ck 时状态为 .in-progress 的文件，可能在故障发生时，已经处于 .pending 状态，也就是写完的状态，也可能仍然处于 .in-progress 状态。
flink的做法是，不管处于什么状态 首先全部标注为 finish 状态。然后根据上次ck时状态中存储的文件的大小进行截断，这样，该文件就能回滚到上次ck成功时的状态。若 Hadoop 版本不支持截断操作，则新建一个后缀为 .valid-length 的文件，内容为文件的大小，单位 byte。
然后flink就可以从上次ck处重新拉取数据源，继续处理，写入sink。
最后，调用 handlePendingFilesForPreviousCheckpoints 将上次ck成功后，若故障发生的很快，没来得及调用 CheckpointListener 的 notifyCheckpointComplete 方法，则此处将文件状态置为 finish 。

BucketSink 是一个控制类，具体的写入操作可以自己实现 org.apache.flink.streaming.connectors.fs.Writer 接口。
其中 snappy 等压缩文件的追加，可以使用
Fs.append 的方式追加内容到同一文件中