企业官网建设流程全解析

1. 从一次数据重复说起：为什么 Flink CheckPoint 非要拉上两阶段提交协议

前两年接手一个 Kafka 到 Kafka 的数据管道，作业跑得挺稳，CheckPoint 也一直正常。结果有次集群磁盘打满触发了 TaskManager 重启，恢复之后下游业务方找上门，说同一批订单数据被消费了两遍，库里直接多出一堆脏数据。当时第一反应是去看 CheckPoint 配置，确认了是 EXACTLY_ONCE 模式，内部算子状态也做了快照恢复，理论上不该重复。折腾了半天才发现，问题出在 Sink 端：Kafka Producer 的数据在 CheckPoint 完成之前就已经发出去了，作业挂掉重启时，这批"已发出但未确认入 CheckPoint"的数据被重新消费、重新写入，而下游没有幂等去重能力，重复就产生了。

这个坑其实是很多 Flink 使用者都会踩的：大家默认以为开了 EXACTLY_ONCE 就万事大吉，但 Flink 的精确一次语义是分层的。内部状态的一致性靠 CheckPoint 的 Barrier 对齐和异步快照就能搞定，可一旦涉及外部系统（Kafka、数据库、文件系统、Doris、TiDB 这些），光靠内部 CheckPoint 根本管不住。要让数据从 Source 到 Sink 端到端真正只处理一次，就必须引入两阶段提交协议（Two-Phase Commit Protocol）。这套协议不是 Flink 发明的，它借用了分布式事务里经典的 2PC 思想，把它嫁接到 CheckPoint 机制上，让外部系统的写入动作和 Flink 的 CheckPoint 周期绑在一起，先"预提交占位"、等全局快照确认后再"真正提交"。

本文会从两阶段提交协议在 Flink CheckPoint 里的完整链路讲起，包括它的分阶段流程、TwoPhaseCommitSinkFunction这个抽象类的生命周期方法、Kafka Sink 的事务超时参数到底怎么算、文件 Sink 和数据库 Sink 为什么实现不一样、Sink V2 之后又发生了什么变化。如果你是刚上手 Flink 的，会搞明白"为什么我的作业恢复后会重复"；如果你已经写过自定义 Sink，会看到 2PC 落地时那些参数背后的取舍逻辑。整个过程我会尽量用大白话拆，配合实际配置和踩过的坑，让你能直接抄作业。

2. 两阶段提交协议在 Flink 里的核心机制拆解

2.1 内部一致性为什么撑不起端到端

先把 Flink 的 CheckPoint 机制捋清楚，才能理解 2PC 补的是哪块短板。Flink 的 CheckPoint 本质是一个分布式的异步快照：JobManager 里的 CheckpointCoordinator 定期往 Source 注入 Barrier，Barrier 像一道分界线顺着数据流往下游飘，每个算子收到 Barrier 后把当前状态做一次快照上传到状态后端（RocksDB、HashMapStateBackend 或远端文件系统），然后继续往下游传。等所有算子的快照都完成了，一次 CheckPoint 就算成功，JobManager 记录下这一次的元数据。

问题在于：算子的状态能被快照，但发给外部系统的请求没法被快照。比如 Kafka Sink 把一条消息producer.send()出去，这条消息在网络里飞着、在 Broker 的日志里躺着，Flink 的状态里根本没有它的存在。作业一旦崩溃恢复，Flink 只知道从上一个成功的 CheckPoint 重新消费数据，那些"已经发出去但属于下一个 CheckPoint 的数据"就会被重新处理。这就是至少一次（AT_LEAST_ONCE）场景下重复数据的来源。

要消除这种重复，就得让"发出数据"和"CheckPoint 成功"这两件事之间存在一个可回滚的中间态，让外部系统能配合 Flink 说："这批数据我先记下来但不生效，等你说 OK 我再真正提交；你说失败我就回滚。"

2.2 两阶段提交协议的两步分别干了什么

Flink 里的两阶段提交，把一次 CheckPoint 周期拆成两个动作：

第一阶段，预提交（preCommit / prepare）。Barrier 到达 Sink 时，Sink 触发snapshotState()，在这里它会把当前活跃的事务做一个"预备状态"的快照。以 Kafka 为例，这个动作是把 Producer 里缓冲的数据flush()出去，让数据落到 Broker 上但事务还处于打开状态（未 commit）。同时把当前事务的 ID、状态等信息写进 Flink 的算子状态里，跟着 CheckPoint 一起落盘。

第二阶段，提交（commit）。等到所有算子都报告 CheckPoint 成功、CheckpointCoordinator 确认这次全局快照完成，它会回调所有算子的notifyCheckpointComplete()。这时候 Sink 才真正调用外部系统的 commit 接口，比如 Kafka 的producer.commitTransaction()、Doris Stream Load 的 label 确认、数据库 XA 事务的commit。一旦提交成功，这批数据对外可见，事务生命周期结束。

中间还有个兜底动作：回滚（abort）。如果作业在这次 CheckPoint 完成之前挂了，恢复的时候 Flink 会从状态里找到那个未提交的事务，调用abort或recoverAndAbort把它撤销掉，这样 Broker 上那些"预提交但没 commit"的数据就不会被下游看到。Kafka 会把 abort 标记写入事务日志，下游消费者用isolation.level=read_committed读的时候会自动跳过这些被中止的事务数据。

用一句话概括：先占坑、后落锤、挂了就回滚。这正是经典 2PC 的 Prepare-Commit-Abort 三步，只不过 Flink 把协调者的角色交给了 CheckpointCoordinator。

2.3 CheckpointCoordinator 到底扮演了什么角色

很多人以为 2PC 的协调者是某个专门的组件，其实在 Flink 里它就是 CheckpointCoordinator，藏在 JobManager 里。它的职责可以拆成几块：

• 发起 CheckPoint：按配置的间隔往 Source 注入 Barrier，同时把这次 CheckPoint 的 ID 广播给所有算子。
• 收集 ACK：每个算子完成状态快照后会向 Coordinator 报告，Coordinator 统计是否所有 Task 都 ACK 了。
• 确认全局完成：当所有 ACK 到齐，标记这次 CheckPoint 为 completed，把元数据写入 CompletedCheckpointStore 持久化。
• 下发 commit 通知：回调每个实现了CheckpointListener的算子，触发notifyCheckpointComplete(checkpointId)。
• 处理超时与失败：如果某个算子迟迟不 ACK，或者某次 CheckPoint 里某个 Task 报错，Coordinator 会判定这次 CheckPoint 失败，并触发对应的事务 abort 逻辑。

理解这个角色分工很重要：Sink 端自己是不知道 CheckPoint 是否全局成功的，它只能被动等 Coordinator 的通知。这也是为什么notifyCheckpointComplete存在延迟——它必须等最慢的那个算子完成快照。如果作业里有个别算子 CheckPoint 特别慢（比如状态特别大、反压严重、磁盘 IO 抖动），整个 2PC 的第二阶段就会被拖后，进而直接影响事务的存活时间。

2.4 它和教科书上的 2PC 有什么不同

分布式系统课上讲的 2PC，是为了解决跨多个数据库的原子提交问题，参与者是多个资源管理器，存在协调者单点、同步阻塞、脑裂等经典问题。Flink 的 2PC 做了不少简化，不能完全等同：

最关键的区别是：Flink 的每个 Sink 并行子任务管自己的事务，协调者只是通知"可以提交了"，并不参与事务本身的 Prepare 决策。这让它避开了协调者宕机导致全局悬挂的死结，但代价是把一部分正确性责任交给了外部系统的事务超时机制——这部分我在下一章讲 Kafka Sink 时细说。

3. 源码视角：TwoPhaseCommitSinkFunction 是怎么跑起来的

3.1 生命周期方法逐个拆解

Flink 早期把 2PC 的公共逻辑封装成了TwoPhaseCommitSinkFunction抽象类，虽然新版已经标记为 deprecated，但绝大多数现成连接器（Kafka、部分文件 Sink）还在用它，而且它的方法设计是理解 2PC 最好的入口。它继承自RichSinkFunction并实现了CheckpointedFunction和CheckpointListener，把 2PC 拆成几个必须实现的方法：

• beginTransaction()：开启一个新事务。Kafka 里对应producer.beginTransaction()；文件 Sink 里可能是创建临时文件、拿到一个临时目录路径。
• invoke()：处理每条数据，把数据写进当前事务的上下文。
• preCommit(transaction)：第一阶段，返回一个"预提交状态"。Kafka 里是对当前事务flush()，让数据落到 Broker 但保持事务打开。
• commit(transaction)：第二阶段，真正提交。Kafka 里是commitTransaction()。
• abort(transaction)：回滚未提交的事务。
• recoverAndCommit(transaction)：作业恢复时对已经从 CheckPoint 里读到的事务做提交（这种事务理论上已经 preCommit 过）。
• recoverAndAbort(transaction)：恢复时把未提交的事务中止。

整个流程可以用一段伪代码串起来：

// 数据到达时的循环 public void invoke(IN value, Context context) throws Exception { if (currentTransaction == null) { currentTransaction = beginTransaction(); } invoke(currentTransaction, value, context); // 写入当前事务 } // CheckPoint 触发时的第一阶段 public void snapshotState(FunctionSnapshotContext context) throws Exception { preCommit(currentTransaction); // flush 但不提交 // 把当前事务写进 operator state state.clear(); state.add(new State<>(currentTransaction.toString(), userContext)); } // CheckPoint 全局完成后回调，第二阶段 public void notifyCheckpointComplete(long checkpointId) throws Exception { Iterator<State<TXN, CONTEXT>> pending = state.iterator(); while (pending.hasNext()) { State<TXN, CONTEXT> txn = pending.next(); commit(txn.transaction); // 真正提交 pending.remove(); } currentTransaction = beginTransaction(); // 开新事务 }

看到这里你就应该明白了：一次 CheckPoint 周期 = 一个事务。从上一个 CheckPoint 的 commit 之后到当前 CheckPoint 的 preCommit 之间收集的所有数据，被打包在同一个事务里。事务边界和 CheckPoint 边界严格对齐，这是整个 2PC 正确性的基础。

3.2 算子状态为什么必须存事务句柄

很多人会忽略一个细节：snapshotState里保存的到底是什么？保存的不是数据，而是事务句柄，也就是能标识这个未提交事务的信息。Kafka 里就是<transactionalId, topicPartition, offset>这一组；文件 Sink 里是临时文件的路径。

为什么必须存它？因为作业恢复时，TaskManager 内存里那些 Producer 对象、临时文件句柄全没了，Flink 只能从 CheckPoint 里恢复算子状态。如果状态里没有事务句柄，就没法知道"上次有个事务在预提交状态悬挂着"，也就没法去 commit 或 abort，最终会出现要么数据永久不可见（未 commit），要么下游看到半成品数据（部分写入未回滚）。

我见过一个自研 Sink 的实现，作者把事务状态存成了本地变量，CheckPoint 里只存了 offset，结果恢复后事务全丢，Broker 上的悬挂事务得等超时才被清理，下游 30 分钟内看不到数据——排了半天才定位到是状态没做快照。

3.3 故障恢复时的三条处理路径

作业重启后，每个 Sink 子任务会从恢复的算子状态里读出待处理的事务列表，然后走三条不同的分支：

1. recoverAndCommit：如果这个事务对应的 CheckPoint 已经全局完成了（也就是 Coordinator 确认过，但 notifyCheckpointComplete 没来得及回调就宕机），那就直接 commit。这种事务在外部系统看来是"可以提交的"，提交它是安全的。
2. recoverAndAbort：如果事务对应的 CheckPoint 没有完成就宕机了，那必须 abort，否则下游会读到脏数据。
3. 正常情况下：无待处理事务，直接从上一个成功 CheckPoint 的 offset 继续消费，开新事务。

哪条路径取决于 CheckPoint 元数据的记录。Flink 在恢复时会把最近一次成功 CheckPoint 的 ID 告诉算子，算子据此判断哪些事务该提交、哪些该中止。

3.4 Sink V2 之后：Committer 抽象登场

TwoPhaseCommitSinkFunction在 Flink 1.20 之后被标记为 deprecated，原因是它把事务的 preCommit 逻辑绑在snapshotState里，每个 CheckPoint 都必须 flush，对高吞吐场景不太友好。新版推荐基于 Sink V2 的接口自己实现，核心是把 2PC 的两阶段明确拆成三个角色：

• SinkWriter：负责写数据，产出需要提交的"committable"对象。
• Committer：负责真正的 commit 和 abort，被放在一个独立的算子（Committer Operator）里，可以单独并行度、单独资源。
• CommittableStateManager：负责把待提交的 committable 存进状态，JobManager 侧的CommittableCoordinator负责在 CheckPoint 完成后通知 Committer。

这套架构的好处是：写数据和提交数据解耦，IO 密集的提交动作不会阻塞主数据流；同时通过CommittableCoordinator做全局协调，比自定义notifyCheckpointComplete更清晰。现在 Kafka、FileSystem、Doris 这些连接器的新版实现都在往这个方向迁移。如果你要写新 Sink，建议直接从 Sink V2 入手，别再用 deprecated 的抽象类了。

4. 落地实战：Kafka Sink 的两阶段提交配置

4.1 事务超时参数到底怎么算

这是 2PC 里最容易搞错的地方。Kafka Sink 有个硬性约束：Kafka 的事务超时时间（transaction.timeout.ms）必须大于事务从开启到提交的最大可能时长，否则 Broker 会主动 abort 掉超时未提交的事务，导致数据丢失。而事务的最大存活时长和 CheckPoint 配置强相关。

Flink 官方给出的计算公式大致是：

transaction.timeout.ms > checkpointInterval * (maxConcurrentCheckpoints + 1) + tolerableRestartTime

其中各项含义：

举个例子，CheckPoint 间隔 3 分钟、最大并发 1、容忍重启 10 分钟，那 transaction.timeout.ms 至少要 3 × (1+1) + 10 = 16 分钟 = 960000 毫秒。实际配置时我一般还会留 20% 到 30% 的余量，避免边界抖动。

注意：transaction.timeout.ms是 Kafka Producer 的配置，默认是 1 分钟，非常容易踩坑。如果你 CheckPoint 间隔是 5 分钟，但没改这个参数，Broker 会在第 1 分钟就把事务超时 abort 掉，表现为 CheckPoint 频繁失败、下游数据断流。另外这个参数在运行时不允许修改，需要和transactional.id的配置一起在构造 Producer 时定好。

4.2 一份能直接抄的 Kafka Sink 配置

下面是一份我实际在用的 Kafka Sink 配置片段，CheckPoint 间隔设的是 3 分钟，事务超时设了 17 分钟，语义模式明确开启 EXACTLY_ONCE：

Properties props = new Properties(); props.setProperty("bootstrap.servers", "kafka-broker:9092"); props.setProperty("transaction.timeout.ms", String.valueOf(17 * 60 * 1000)); // 17 分钟 props.setProperty("enable.idempotence", "true"); // 幂等生产者，配合事务使用 props.setProperty("isolation.level", "read_committed"); // 下游消费者读已提交 KafkaSink<String> sink = KafkaSink.<String>builder() .setBootstrapServers("kafka-broker:9092") .setRecordSerializer(KafkaRecordSerializationSchema.builder() .setTopic("target-topic") .setValueSerializationSchema(new SimpleStringSchema()) .build()) .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE) .setTransactionalIdPrefix("my-job-") .setKafkaProducerConfig(props) // 关键：事务超时必须大于 CheckPoint 间隔 + 容忍重启时间 .setProperty("transaction.timeout.ms", String.valueOf(17 * 60 * 1000)) .build();

对应的 Flink 作业配置：

execution.checkpointing.interval: 3min execution.checkpointing.timeout: 10min execution.checkpointing.mode: EXACTLY_ONCE execution.checkpointing.min-pause: 1min state.backend: rocksdb state.checkpoints.dir: hdfs:///flink/checkpoints

这里min-pause我配了 1 分钟，意思是两次 CheckPoint 之间至少间隔 1 分钟，避免 CheckPoint 挤在一起抢资源。对高吞吐作业来说这个参数很关键，否则 CheckPoint 一密集，Kafka 的 flush 动作频繁触发，吞吐掉得厉害。

4.3 文件系统 Sink 和数据库 Sink 的不同打法

同样是 2PC，不同外部系统的实现差异很大，不能一套配置走天下。我把常见的三类对比一下：

文件系统的 2PC 特别值得说：它的"提交"其实是一次重命名操作。第一次写数据时创建一个隐藏的临时文件（比如.part-xxx.inprogress），等 commit 时把它重命名成正式文件。重命名操作在大多数文件系统上是原子的，所以数据要么完整可见，要么完全不可见，不存在中间状态。但这里有个坑：临时文件和目标文件必须在同一个文件系统上，跨挂载点 or 跨对象存储的重命名不是原子操作，会导致可见性错误。我在一次 HDFS 到 OSS 的迁移中就遇到过，重命名变成了拷贝加删除，作业挂了之后出现半截文件。

Doris Sink 走的是 Stream Load 的 label 机制，每个 label 对应一次导入，Doris 内部靠 label 做幂等——相同 label 的重复导入会被合并。它的两阶段提交需要在配置里显式打开 transaction 开关，并保证 label 的前缀在作业生命周期内唯一。这点和 Kafka 的transactionalIdPrefix思路一致。

5. 常见问题与排查技巧实录

5.1 CheckPoint 超时和事务超时是两码事

新手最容易把这两个概念混在一起。CheckPoint 超时是 Flink 层面配置的（execution.checkpointing.timeout），超时后这次 CheckPoint 被判定失败，作业会按重启策略重启；事务超时是外部系统层面的（Kafka 的transaction.timeout.ms、数据库的 XA 超时），到点后外部系统会强制 abort 当前事务，无论 Flink 是否还在进行中。

两者的关系是：事务超时必须显著大于 CheckPoint 的最坏耗时，否则会出现 Flink 还等着 commit，外部系统已经把事务 abort 了，等到 CheckpointCoordinator 通知 commit 时，Sink 会报InvalidTxnStateException。日志里常看到的报错长这样：

org.apache.kafka.common.errors.InvalidTxnStateException: The producer attempted to use a producer id which is not currently assigned to its transactional id.

或者：

TransactionCoordinator - Aborting transaction ... due to timeout

排查顺序我一般这么走：先看 CheckPoint 监控面板确认 CheckPoint 平均耗时和 P99 耗时，再核对事务超时参数是否大于 P99 耗时的 1.5 到 2 倍，不够就调大。如果 CheckPoint 耗时本身就很长（比如单次超过 5 分钟），那调事务超时只是治标，根子还在于状态太大或反压，得从反压和数据分布上查。

5.2 数据重复/丢失的定位方法论

数据出了问题，按下面的顺序排查效率最高：

第一步，确认 Source 和 Sink 的语义级别。Source 端 Kafka 消费者要打开read_committed，Sink 端要开 EXACTLY_ONCE，两端缺一不可。只开 Sink 不开 Source，Source 可能读到上游未提交的事务数据，同样会重复。

第二步，看 CheckPoint 是否稳定成功。打开 Flink UI 的 CheckPoints 页面，看最近 20 次是否全部 COMPLETED。如果有失败记录，那么作业可能在反复重启，每次重启都会重放数据，重复次数和重启次数相关。

第三步，检查事务日志。Kafka 里可以用kafka-transactions.sh列出当前悬挂的事务，看是否有长期未提交的事务堆积；Doris 里可以查SHOW LOAD看 label 的状态，是否有 PREPARE 状态长期未变成 VISIBLE。

第四步，确认下游是否幂等。即便上游做了 2PC，下游如果是从 Kafka 消费后再写数据库且没有主键去重，仍然可能因为下游自己的重复消费产生脏数据。端到端一致性是链路每一环的事，不能只在 Flink 内部保证。

5.3 几类连接器报错的速查

日常遇到的高频报错整理成一张表，看到直接对号入座：

那个flink type is datev2, but arrow type is dateday的报错很典型，是 Doris 连接器在做 Arrow 格式转换时字段类型对不上，一般出现在 Doris 表里用了DATEV2而 Flink 侧是DATE的场景，需要对齐两边类型或者升级flink-connector-doris到匹配的版本。这类问题看着像连接器 bug，其实多数是版本和类型映射没对齐。

5.4 三条我踩过的独家经验

第一，别把事务超时设得太小"省事"。有人想当然设个 5 分钟，结果业务高峰期 CheckPoint 一慢就翻车。事务超时的调整成本不高，设大一点对系统负担很小，宁可浪费点资源也别踩数据丢失的坑。

第二，transactionalIdPrefix 必须全局唯一且带上作业标识。多个作业用同一个前缀会导致 Producer 的 transactional.id 冲突，出现莫名其妙的 abort。我一般用{env}-{jobName}-{version}-这种格式，避免多环境互相干扰。

第三，开启 2PC 后作业的吞吐会下降，做好心理准备。每次 CheckPoint 都要 flush、每次提交都有网络往返，相比 at-least-once 吞吐掉 10% 到 30% 是常态。如果你的业务能接受去重成本换取性能，用 at-least-once 加下游幂等反而是更划算的选择。2PC 不是无脑开，要看业务对重复的容忍度和性能预算。

6. 性能调优：让两阶段提交别拖垮作业

6.1 CheckPoint 间隔与吞吐的平衡术

CheckPoint 间隔和事务粒度是强绑定的：间隔越短，事务越小，每次 flush 的数据量越小，但每秒的事务提交开销越大；间隔越长，事务越大，吞吐好，但失败恢复要重放的数据多，事务超时窗口也大。

我实测下来的经验区间是：中小流量作业（每秒几万条）用 1 到 3 分钟比较舒服；高吞吐作业（每秒几十万条以上）建议 3 到 5 分钟，同时把事务超时按公式放大到 15 分钟以上。极端的高吞吐场景，与其硬扛 2PC 的开销，不如评估一下下游加幂等的成本，说不定 at-least-once 更合适。

另外min-pause这个参数经常被忽略。它保证两次 CheckPoint 之间有喘息时间，避免前一次还没结束、后一次又来了。高负载下开 30 秒到 1 分钟的 min-pause，能明显减少 CheckPoint 对正常数据流的干扰。

6.2 状态后端和非对齐 CheckPoint 的影响

2PC 的第一阶段要把事务句柄写进算子状态，状态后端的写入性能直接影响 CheckPoint 耗时。HashMapStateBackend 在内存里操作快，但状态大了会撑爆内存；RocksDB 适合大状态，但写入要经过本地磁盘，慢一些。如果算子状态里全是事务句柄这种小对象，HashMapStateBackend 通常够用；如果有大量业务状态，还是得 RocksDB，配合增量 CheckPoint 降低每次快照的 IO 量。

反压严重时，Barrier 对齐会卡住整个流，CheckPoint 迟迟不到 Sink，事务就一直不能提交，超时风险陡增。这种场景下开启**非对齐 CheckPoint（Unaligned Checkpoint）**很有用，它让 Barrier 越过缓冲区直接往下游走，避免被慢算子堵住。代价是快照里要带上缓冲区数据，状态会变大一点，但换来的是 CheckPoint 稳定性和事务生命周期的可预测性，在反压常态化的作业里非常值。

6.3 个人一点收尾的体会

写到这里，其实我想强调的就一件事：两阶段提交协议是 Flink 端到端精确一次的最后一块拼图，但它不是银弹。它把所有复杂性压到了外部系统和参数配置上，你对 Kafka 事务、文件系统原子重命名、数据库 XA 的理解有多深，2PC 就能用得多稳。我见过太多作业开了 EXACTLY_ONCE 却在生产里丢数据、脏数据，最后发现全是参数没对齐、事务超时没配好、下游没做隔离级别这种细节。

如果你的作业现在还是 at-least-once，先别急着改 EXACTLY_ONCE。把下游去重做好，跑一段时间看看真实的重复率，再评估引入 2PC 的收益和成本。真要上 2PC，记住三条：事务超时算够、transactionalIdPrefix 唯一、CheckPoint 监控盯紧。这三条做到了，2PC 基本不会给你添乱，反而能在故障恢复时帮你守住数据底线。