在Flink 1.4.0之前，Flink只能做到应用程序内的精确一次处理（exactly-once semantic），而无法做到端到端的精确一次（end-to-end exactly-once semantics）处理。 要提供端到端的精确一次语义，还需要sink端的外部存储系统提供提交和回滚的功能，然后与Flink Checkpoint机制结合使用才能实现端到端的精确一次语义。

在分布式系统中，协调提交和回滚的一种常见方法就是两阶段提交协议。在Flink 1.4.0版本中，Flink实现了一个新功能：TwoPhaseCommitSinkFunction。这是一个抽象类，它提取了两阶段提交协议中的通用逻辑，并提供了一个抽象层，对用户来说，我们只需实现少量的几个方法便可以实现端到端精确一次的Flink应用程序。

我们首先来了解一下什么是两阶段提交协议，以及其内部的工作原理。

两阶段提交协议

两阶段提交（Two-phase Commit）是为了使基于分布式系统架构的所有节点在进行事务提交时保持一致性而设计的一种算法，通常也被称为两阶段提交协议。

在分布式系统中，虽然每个节点可以知道自己的操作是否成功，但是却无法知道其他节点的操作结果。当一个事务跨越多个节点时，为了保持事务的ACID特性，需要引入一个作为协调者（Coordinator）的组件来统一掌控所有节点（通常被称为参与者）的操作结果并最终指示这些参与者节点是否要把操作结果进行真正的提交（比如，将更新后的数据写入外部存储系统等等）。

总的来说，两阶段提交协议的算法思路可以概括为：参与者将操作结果通知协调者，协调者根据所有参与者的反馈结果决定各参与者是要执行提交操作还是终止操作。

1. 两阶段提交的前提条件

两阶段提交的成立要基于以下假设：

• 该分布式系统中，存在一个节点作为协调者，其他节点作为参与者，且节点之间可以进行网络通信。
• 所有节点都采用预写式日志，且日志被写入后即被保存在可靠的存储设备上，即使节点损坏也不会导致日志数据的丢失。
• 所有节点不会永久性损坏，即使损坏后也可以恢复。

2. 两阶段提交的基本算法

下面通过分阶段来对两阶段提交进行说明。

a. 第一阶段（提交请求阶段）

1. 协调者节点向所有参与者节点询问是否可以执行提交操作，并开始等待各参与者节点的响应。
2. 参与者节点执行协调者节点询问发起为止的所有事务操作，并将undo信息和redo信息写入日志。
3. 各参与者节点响应协调者节点发起的询问，如果参与者节点的事务操作执行成功，则它返回一个"ack"同意消息；如果参与者节点的事务操作执行失败，则它返回一个"abort"中止消息。

b. 第二阶段（提交执行阶段）

第二阶段要执行怎样的操作是由第一阶段各参与者返回的消息来决定的。

当所有参与者节点都返回"ack"同意消息时：

1. 协调者节点向所有参与者节点发出"正式提交"的请求。
2. 参与者节点正式完成操作，并释放在整个事务期间占用的资源。
3. 参与者节点向协调者节点发送"完成"消息。
4. 协调者节点收到所有参与者节点反馈的"完成"消息后，事务完成。

如果任意一个参与者节点返回了"abort"中止消息，或者协调者节点在第一阶段的询问超时之前无法获取所有参与者节点的响应消息时：

1. 协调者节点向所有参与者节点发出"回滚操作（callback）"的请求。
2. 参与者节点利用之前写入的Undo信息执行回滚，并释放在整个事务期间占用的资源。
3. 参与者节点向协调者节点发送"回滚完成"消息。
4. 协调者节点收到所有参与者节点反馈的"回滚完成"消息后，取消事务。

3. 两阶段提交的缺点

两阶段提交算法的最大缺点就在于它在执行过程中，节点会因为在等待对方的响应消息而处于阻塞状态。

另外，协调者节点通知参与者节点进行提交操作时，如有参与者节点出现了崩溃等情况而导致协调者始终无法获取所有参与者的响应消息，这时协调者将只能依赖其自身的超时机制来生效。

Flink-两阶段提交协议

1. Flink-Kafka构建端到端Exactly-once应用

假设，我们有一个Flink应用程序，source端为Kafka，中间做一些聚合操作，最后写回Kafka。Kafka在0.11版本开始支持事务，这就可以使用Flink和Kafka构建端到端Exactly-once的应用程序。

我们在介绍两阶段提交的时候了解到，在分布式系统中，事务的提交或回滚是需要多个所有参与者都同意才能确保数据的一致性。那么，Flink就是用两阶段提交来保证数据的一致性的。

Checkpoint的开始表示两阶段提交协议的"pre-commit"阶段，当触发Checkpoint时，Flink JobManager会向数据流注入一个barrier（它将数据流中的记录划分为进入当前Checkpoint的部分和进入下一个Checkpoint的部分）。Barrier会随着数据流在operator之间传递，对于每一个operator，都会触发它的状态后端来保存其状态数据。

Flink应用程序的source端会保存Kafka的offsets，之后会将barrier传递给下一个operator。如果operator只有内部状态，这是没有问题的，因为内部状态是由Flink的状态后端来管理和存储的；如果operator有外部状态，比如sink端要写入外部存储系统（比如Kafka），那么为了确保exactly-once的语义，外部存储系统必须提供整合两阶段提交协议的事务机制。

预提交阶段在Checkpoint成功完成之后结束。下一步是通知所有operatorCheckpoint已经成功，这是两阶段提价协议的提交阶段，JobManager会为应用程序中的每个operator发出Checkpoint完成的回调通知。

2. Flink实现两阶段提交

Flink将两阶段提交协议中的通用逻辑抽象为了一个类——TwoPhaseCommitSinkFunction。

我们在实现端到端exactly-once的应用程序时，只需实现这个类的4个方法即可：

• beginTransaction：开始事务时，会在目标文件系统上的临时目录中创建一个临时文件，之后将处理数据写入该文件。
• preCommit：在预提交时，我们会刷新文件，关闭它并不再写入数据。我们还将为下一个Checkpoint的写操作启动一个新事务。
• commit：在提交事务时，我们自动将预提交的文件移动到实际的目标目录。
• abort：中止时，将临时文件删除。

如果出现任何故障，Flink将应用程序的状态恢复到最近一次成功的Checkpoint。如果故障发生在预提交成功之后，但还没来得及通知JobManager之前，在这种情况下，Flink会将operator恢复到已经预提交但尚未提交的状态。

参考

• https://flink.apache.org/features/2018/03/01/end-to-end-exactly-once-apache-flink.html
• https://zh.wikipedia.org/wiki
• https://courses.cs.vt.edu/the-two-phase-commit-protocol