问题1：阐述 Flink 提供的三种数据处理语义，解释 Checkpoint 机制如何保证 Flink 程序结果的 Exactly-Once 语义？

问题2：结合 Kafka 分析，如何通过两阶段提交协议，提供端到端的 Exactly-Once 处理？

1.Flink 提供的三种数据处理语义

① At-Most-Once：最多一次，如果产生故障，可能丢失数据。
② At-Least-Once：最少一次，如果产生故障，可能有重复数据。
③ Exactly-Once：精确一次，如果产生故障，也能保证数据不丢失不重复。

// At-Least-Once 最多一次
CheckpointingMode.AT_LEAST_ONCE

// Exactly-Once 精确一次
CheckpointingMode.AT_LEAST_ONCE

flink 新版本已经不提供 At-Most-Once 语义。

2.Checkpoint 机制保证 Exactly-Once 语义

结论：Checkpoint Barrier 对齐机制实现 Exactly-Once 语义。如果 Barrier 不对齐，即 At Least Once 语义。

Flink 分布式异步快照的核心是 Checkpoint 机制，其关键是采用标记信号 Barrier，使得数据流被切分成微批，进行 Checkpoint 保存状态数据，如下图所示。

Barrier工作流程.JPG

Barrier对齐机制.JPG

说明：Barrier 不对齐可能导致有重复数据。作业算子的全部并行度均为1，即默认 Barrier 对齐。

3.两阶段提交协议保证端到端 Exactly-Once 语义

3.1 Flink 两阶段提交协议-2PC

（1）两阶段提交协议原理

分布式系统利用两阶段提交协议实现事务性，保证数据的一致性。两阶段提交分为：预提交阶段 per-commit 与提交阶段 commit。通常包含两个角色：协调者和执行者，协调者管理所有执行者的操作，执行者用于执行具体的提交操作。

工作流程：
① 协调者向执行者发送 pre-commit 命令；

② 执行者执行预提交后，向协调者发送 ack；
③ 协调者收到 ack 后，向执行者发送 commit 命令；
④ 执行者执行提交操作；

说明：如果有执行者的预提交操作失败，返回失败给协调者，则协调者向所有执行者发送 rollback，执行回滚操作，保证数据一致性。

两阶段提交原理.JPG

两阶段提交 2PC 的缺点：
①单点故障导致阻塞：执行者向协调者发送 ack，然后协调者发送故障，导致执行者会一直阻塞下去。
② 数据不一致：当协调者向执行者发送部分 commit 请求之后，发生故障导致只有一部分执行者接受到了 commit 请求。

（2）Flink 的两阶段提交流程。

工作流程：
① JobMaster 即 CheckpointCoordinator 会定期向每个 source task发送命令 start checkpoint（trigger checkpoint）；
② 当 source task 收到 trigger checkpoint 指令后，产生 barrier 并通过广播的方式发送到下游。source task 同时会执行本地 checkpoint n，当 checkpoint n 完成后，向 JobMaster 发送 ack；
③ 当流图的所有节点都完成 checkpoint n，JobMaster 会收到所有节点的 ack，那么就表示完成 checkpoint n；

说明：checkpoint 机制的调用流程实质是 2PC。JobMaster 是协调者，所有operator task 是执行者。start checkpoint 是 pre-commit 的开始信号，而每个 operator task 的 checkpoint 是 pre-commit 过程，ack 是执行者 operator task 反馈给协调者 JobMaster ，最后 callback 是 commit。

Flink两阶段提交.JPG

3.2 Flink + Kafka 实现端到端 Exactly-Once 语义

结论：端到端 Exactly-Once = Flink 内部 Exactly-Once + 第三方存储支持事务（flink-connector 支持事务）

3.2.1 分析 TwoPhaseCommitSinkFunction

Sink 输出支持两阶段提交，其流程抽象为 TwoPhaseCommitSinkFunction。

public abstract class TwoPhaseCommitSinkFunction<IN, TXN, CONTEXT>
        extends RichSinkFunction<IN>
        implements CheckpointedFunction, CheckpointListener {
    /**
     * Method that starts a new transaction.
     *
     * @return newly created transaction.
     */
    protected abstract TXN beginTransaction() throws Exception;

    /**
     * Pre commit previously created transaction. Pre commit must make all of the necessary steps to prepare the
     * transaction for a commit that might happen in the future. After this point the transaction might still be
     * aborted, but underlying implementation must ensure that commit calls on already pre committed transactions
     * will always succeed.
     *
     * <p>Usually implementation involves flushing the data.
     */
    protected abstract void preCommit(TXN transaction) throws Exception;

    /**
     * Commit a pre-committed transaction. If this method fail, Flink application will be
     * restarted and {@link TwoPhaseCommitSinkFunction#recoverAndCommit(Object)} will be called again for the
     * same transaction.
     */
    protected abstract void commit(TXN transaction);

    /**
     * Abort a transaction.
     */
    protected abstract void abort(TXN transaction);
}

具体流程：
① 初始化快照 initializeState 或者执行快照 snapshotState 的时候，都会创建事务 beginTransaction；
② 执行快照 snapshotState 的时候，预提交事务 preCommit；
③ 完成快照 notifyCheckpointComplete 的时候，提交事务 commit；
④ 初始化快照 initializeState 或者 算子 Function 关闭（异常故障），丢弃事务 abort；

3.2.2 Flink + Kafka 的两阶段提交流程

（1）Start Checkpoint

JobManager（CheckpointCoordinator）向所有 kafka source 节点 trigger Checkpoint，即注入 checkpoint barrier。

start checkpoint.png

（2）Pre-Commit

kafka source task 把 checkpoint barrier 广播到下游，同时会执行本地 checkpoint。以此此类推，流图的所有节点都收到 barrier 和执行 checkpoint。最后当 kafka sink task 执行checkpoint 的时候，向 kafka 预提交事务 pre-commit。

pre-commit.png

（3）Commit

当流图的所有节点都完成 checkpoint，JobManager（CheckpointCoordinator）通知所有operator task，已经完成本次 checkpoint。此时 kafka sink task 向 kafka 提交事务 commit。

commit.png