Flink精确一次语义疑问：Checkpoint与Sink端Barrier确认问题

针对你基于Flink 1.3的RichSinkFunction实现MongoDB写入作业的几个疑问，我结合Flink 1.3的机制来逐个拆解：

疑问1：Sink何时确认Checkpoint Barrier？是在invoke函数开始时还是执行完成后？是否会等待MongoDB持久化响应后再确认？ #

在Flink 1.3中，对于RichSinkFunction来说，Checkpoint Barrier的确认逻辑是这样的：

• 当Barrier到达Sink算子时，Sink会先处理完所有在Barrier之前到达的数据（也就是这些数据的invoke()方法都执行完毕），之后才会向JobManager发送Checkpoint完成的确认信号。
• 如果你在invoke()里是同步写入MongoDB（代码里等待MongoDB返回写入成功的响应后，invoke()才结束），那么Barrier的确认就会在数据确实持久化到MongoDB之后才触发；但如果是异步写入（比如把数据丢到线程池就立刻返回invoke()），那Barrier可能在数据还没真正写入MongoDB时就完成确认了。

疑问2：若Checkpoint提交由异步线程执行，作业故障时Flink如何保证精确一次语义？若Sink已将数据写入MongoDB但Checkpoint未提交，重启后是否会导致重复数据？ #

首先要明确：Flink 1.3没有TwoPhaseCommitSinkFunction，所以仅靠RichSinkFunction无法原生实现端到端的EXACTLY_ONCE语义，这里的风险点要注意：

• 如果Checkpoint的提交是异步执行的，而你的Sink又是异步写入MongoDB，那么很可能出现两种异常场景：
  1. 数据还没写入MongoDB，但Checkpoint已经确认完成：此时作业故障重启，Flink会从该Checkpoint恢复，认为这些数据已经处理过，导致数据丢失。
  2. 数据已经写入MongoDB，但Checkpoint还没完成确认：此时作业故障重启，Flink会从最近一次成功的Checkpoint重新处理数据，这就会导致已经写入MongoDB的数据被重复写入，破坏EXACTLY_ONCE。
• 要缓解这个问题，你要么让MongoDB的写入具备幂等性（比如给每条数据分配唯一主键，重复写入时直接覆盖旧数据），要么改为同步写入模式（让invoke()等待MongoDB写入成功再返回）——但即使是同步写入，也无法完全避免极端情况（比如写入成功但Checkpoint确认信号在网络中丢失），这时候还是会出现重复，所以幂等性是更可靠的补充方案。

疑问3：从Flink控制台取消作业时，Flink会等待异步Checkpoint线程完成还是直接执行强制kill -9？ #

当你从Flink控制台取消作业时，Flink会先尝试优雅关闭流程：

1. 首先向所有TaskManager发送取消信号，Task会先处理完当前正在处理的批次数据，然后调用RichSinkFunction的close()方法，释放相关资源。
2. 如果优雅关闭在超时时间内（默认是1分钟，可配置）完成，作业就正常停止；如果超时还没完成，Flink才会强制终止进程（类似kill -9）。
3. 对于正在执行的异步Checkpoint线程：当作业被取消时，Flink会直接中止当前的Checkpoint流程，不会等待它完成——因为取消作业后，这个Checkpoint已经没有保留的意义了，Flink会清理相关的Checkpoint资源。

内容的提问来源于stack exchange，提问作者Mudit bhaintwal