如果你在 Flink 任务的配置里把执行模式设成 exactly-once就觉得 Kafka 到 MySQL 这条链路一条不丢、一条不重那大概率要在真实环境里栽跟头。我自己为这个事排查过整整一天来源端是 Kafka落地端是 MySQL任务重启一次库里就多出一批重复数据。后来才搞清楚Flink 的 exactly-once 只管它自己的状态快照MySQL 这一端如果不参与两阶段提交2PC前面所有 checkpoint 工作都白做了。这篇文章就是一次完整的“Flink EOS MySQL 验证 2PC”实操实录从原理拆解到自定义 XA Sink 的代码再到用日志、XA RECOVER、binlog 三条证据链确认事务真正按 PREPARE/COMMIT 走了完整流程。遇到同样问题的人可以直接按这些步骤在你自己环境里复制一遍。1. 为什么要把 EOS、MySQL、2PC 这三个词放在一起验证1.1 标题里的三个关键词分别意味着什么EOS 全称是 Exactly-Once Semantics也就是精确一次语义。这个词在流处理领域被讲得最多也最容易让人误解。它描述的是一条数据从进入 Flink 开始到最终写入某个外部系统在整个处理链条里只生效一次。Flink 官方也一直在强调它自己内部的算子状态能做到精确一次但“内部精确一次”和“端到端精确一次”是两码事。MySQL 在这条链路里是外部存储承担最终数据落地的角色。它本身支持 XA 事务这是它能接住 Flink 的 2PC 协议的前提。很多团队对选型没什么概念觉得“只要 Flink 开了 exactly-onceMySQL 随便写写就行了”结果就是重复数据、对不上账、重跑后结果漂移。2PC 是 Two-Phase Commit 的缩写翻译过来就是两阶段提交。它把一个跨系统的“提交”动作拆成 prepare 和 commit 两个阶段目的是避免多个参与者之间出现“一边提交了一边回滚了”这种不一致。Flink 把整套协议用在了自己 Sink 端的生命周期管理上。把这三点放在一起验证本质上就是在回答一个问题你现在这套 Flink 作业到底有没有真正做到端到端精确一次如果只是看配置面板里选了 exactly-once 就说“做了”我建议你还是把下面的验证流程跑一遍。1.2 端到端一致性是链路上的协议不是单点行为很多人对精确一次的理解有个误区以为它是 Flink 单方面能承诺的事情。实际上一条数据从 Kafka 读出来中间经过各种算子最后落到 MySQL这条链路上的每一个环节都要配合最终才能得到一致结果。打个比方Flink 内部的状态快照相当于一家公司内部的账本备份。账本备份得再完整如果负责发货的仓库不配合说“我没收到指令给你多发了一次货”那么账面数据和实际库存还是对不上的。这里仓库就是 MySQL它必须参与提交协议才知道哪些货该发、哪些货不该发。所以端到端精确一次是整条链路的“协议”不是 Flink 单点的“特性”。Kafka 负责精确重放 offsetFlink 负责把状态和进度做成 checkpointMySQL 负责用事务接收最终结果。任何一环断了都谈不上精确一次。1.3 这个验证到底能证明什么适合谁读这个验证能证明的东西很明确你的 Flink 作业在写入 MySQL 时Sink 端确实走完了 XA 事务的 PREPARE 和 COMMIT 流程状态恢复时能够把未完成事务重新提交或者回滚最终 MySQL 里的数据总量不会因为任务重启而改变。适合读这篇文章的人大概是两类。一类是刚开始用 Flink、被领导或者同事问到“数据到底会不会重复啊”的人你可以把验证结果直接拿出来说话。另一类是在做技术方案评估打算用 Flink 接 MySQL 做实时数仓或业务同步需要确认一致性和事务开销是否可接受。这篇文章不涉及深奥的分布式系统理论按着操作就能看到结果。2. Flink 内部是怎么做到 Exactly-Once 的2.1 Barrier 和数据流的“结算点”要理解 Flink 的 2PC 怎么跟 MySQL 配合先要把 Flink 内部的精确一次机制讲清楚。Flink 在各算子之间注入了一种叫 Barrier 的特殊标记它跟着数据流一起流动把数据流切成一段一段。Barrier 到达某个算子时代表这条算子之前的数据已经处理完毕后面再到的数据属于下一个检查点周期。Barrier 对齐是一个关键动作。想象一个双车道汇入单车道的过程左边车道的车先到了但它必须等右边车道的车也到齐再一起放行。Flink 就是让各条输入流在同一个 Barrier 处对齐确保整个流处理进度是一致的不会出现某个上游已经处理到第 100 条另一个上游还在处理第 80 条的状态混乱。这种对齐机制是状态一致性的地基。只要 Barrier 对齐这件事做到了Flink 内部所有算子对数据进度的理解就是一致的。处理结果不会再因为你盲目处理乱序数据而出现错乱。2.2 Checkpoint把整个流的状态拍一张快照Barrier 对齐之后Flink 会把所有算子的状态做一次快照这个动作就叫 Checkpoint。状态包括 keyed state、operator state、source 的消费位点等等。快照完成后只要任务崩溃Flink 可以把整个作业回滚到最近一次 checkpoint 的位置重新开始读数据和处理数据。这里的核心价值在于Flink 把“进度”和“计算结果”打包在了一起。如果 source 是 Kafka那么 checkpoint 时会保存每个分区的消费 offset如果中间算子维护了累加器或者状态表那么这些状态也一起被写进 checkpoint。恢复的时候数据从保存的 offset 开始重读状态从保存的版本开始重建整个过程看起来就像时间倒流了一样。但注意这个时间倒流只发生在 Flink 自己的世界里。外部系统并不知道发生了什么如果 Sink 在崩溃前已经把一部分数据提交到了 MySQL恢复后这些数据又会被重新算一遍、重新写一遍重复就出现了。2.3 为什么内部精确一次不等于端到端精确一次内部精确一次和端到端精确一次之间隔着一个“外部系统是否愿意配合”的鸿沟。Flink 可以保证自己会重新计算所有未完成的数据但它无法阻止 MySQL 里已经存在的那些写入。要做到端到端精确一次Sink 侧必须满足两个条件之一一是把写入包装成事务而且这个事务要支持 Prepare 这种“半提交”状态二是写入本身具备幂等性也就是同一份数据哪怕被写入多次最终效果也跟写入一次相同。MySQL 这类关系型数据库天然支持前者也就是 XA 分布式事务。所以才有了 TwoPhaseCommitSinkFunction 这种抽象。它把 Flink 的 checkpoint 机制和外部系统的事务能力结合起来让外部提交和状态快照同步推进。没有这套机制Flink 内部再精确货物到了仓库那一端还是会乱。3. 2PC 在 Flink 中的落地由 TwoPhaseCommitSinkFunction 统一驱动3.1 两阶段提交的核心流程2PC 协议里有两个角色协调者和参与者。在 Flink 的场景里协调者是 Flink 的 Checkpoint Coordinator参与者就是 MySQL。整个提交流程可以拆成几个关键状态。第一阶段是 Prepare。协调者让所有参与者把数据处理完但不最终提交而是进入一个“预提交”状态。参与者要在这个阶段把自己的结果写稳保证后续可以提交或者回滚。第二阶段是 Commit。协调者确认所有参与者都准备好之后再统一发指令让它们真正提交。如果任何一个参与者 Prepapre 失败协调者就会让所有参与者回滚。这套协议的好处是不会出现“A 已经提交B 还没提交”的中间不一致状态。坏处是Prepare 阶段会长时间持有数据库资源锁的粒度需要仔细控制。Flink 的 TwoPhaseCommitSinkFunction 正是围绕这套状态机来设计的。3.2 Flink 回调时机preCommit 在 checkpoint 完成前commit 在完成后TwoPhaseCommitSinkFunction 的生命周期和 Flink 的 checkpoint 流程紧紧绑定。它的 beginTransaction 会在算子初始化时打开一个新事务每条数据通过 invoke 方法写入这个事务等到 checkpoint 触发时Flink 调用 preCommit 方法把事务推入 Prepared 状态整个 checkpoint 成功结束之后Flink 回调 commit 方法让事务真正提交如果任务失败则回调 abort 或者 recoverAndAbort 来清理。这里有一个非常关键的时间窗口preCommit 发生在 checkpoint barrier 被处理的时候而 commit 发生在 JobManager 确认 checkpoint 完成之后。也就是说在 checkpoint 还在进行中的那段时间里MySQL 里可以看到一批已经 PREPARE 但还没被 COMMIT 的事务这正好是我们做验证的最佳时机。理解这个时机之后你就知道为什么在 MySQL 里偶尔能看到残留的 Prepared 事务是正常现象了。它们不是泄漏而是恰好处于分布式事务的中间状态。如果它们长时间不消失那才说明 Flink 侧可能出问题了。3.3 MySQL 的 XA 命令如何与 Flink 回调一一对应MySQL 从很早的版本就开始支持 XA 分布式事务协议它提供了一组专门的 SQL 命令来配合 2PC 使用。这组命令和 Flink 的 TwoPhaseCommitSinkFunction 回调方法能一一对应起来。beginTransaction 对应 XA START表示开启一个 XA 事务数据写入就是在 XA START 和 XA END 之间执行普通的 INSERT 或 UPDATEpreCommit 对应 XA END 和 XA PREPARE表示事务进入可提交状态commit 对应 XA COMMIT真正把数据提交abort 对应 XA ROLLBACK清理未提交事务recoverAndCommit 和 recoverAndAbort 对应的是事务恢复场景在 Flink 从 checkpoint 恢复时执行同样的命令。这张对照关系是整个验证实验的骨架。掌握了它你看 Flink 源码时就不会一头雾水反过来看 MySQL 端的状态也能知道 Flink 正处在哪个阶段。Flink TwoPhaseCommitSinkFunction 回调MySQL XA 命令说明beginTransactionXA START开启一个 XA 事务invoke普通 INSERT/UPDATE在事务中写入业务数据preCommitXA END XA PREPARE结束事务读写并进入预提交commitXA COMMIT事务正式提交abortXA ROLLBACK事务回滚recoverAndCommitXA COMMIT恢复后提交未完成事务recoverAndAbortXA ROLLBACK恢复后回滚未完成事务4. 搭建一个最小验证环境4.1 版本选择与开启 binlog我把验证环境锁定在一套比较稳妥的组合上Flink 1.16 版本MySQL 8.0 版本MySQL Connector/J 8.0 版本。Flink 版本不同Sink API 的类名和序列化方式会有细微差异但核心逻辑不变。MySQL 8.0 的 XA 支持比较完整perf 方面也稳定。最关键的一步是打开 MySQL 的 binlog。XA 事务的 PREPARE 和 COMMIT 记录要写进 binlog任务恢复时才能正确判断事务状态。如果 binlog 没开恢复阶段可能连事务边界都无法确认严重时候直接抛错。在 my.cnf 里可以这样配[mysqld] server-id1 log-binmysql-bin binlog_formatROW配置完成之后重启 MySQL然后用 SHOW MASTER STATUS 确认 binlog 已经开启。注意 binlog_format 一定要用 ROW否则事务中的具体操作细节在恢复时可能看不全。另外用于验证的账号需要具备 XA_RECOVER_ADMIN 权限否则执行 XA RECOVER 会报权限不足。4.2 自定义 MySQL XA Sink 的代码骨架我建议不要直接用 Flink 封装好的 JDBC Sink因为那样你只看到接口看不到内部状态流转。为了验证 2PC我们用自定义 XA Sink 更直观。下面这套代码我把核心部分抽了出来足够在本地环境跑通。首先定义一个事务包装类注意连接对象不能序列化所以用 transient 标记真正要持久化的只有 xid 和连接参数import java.io.Serializable; import java.sql.Connection; public class XATransaction implements Serializable { public String xid; public String jdbcUrl; public String user; public String password; public transient Connection connection; public XATransaction(String xid, String jdbcUrl, String user, String password) { this.xid xid; this.jdbcUrl jdbcUrl; this.user user; this.password password; } }然后实现序列化器只把必要信息写进 checkpointimport org.apache.flink.api.common.serialization.SimpleVersionedSerializer; import java.io.IOException; import java.nio.charset.StandardCharsets; public class XaTransactionSerializer implements SimpleVersionedSerializerXATransaction { Override public int getVersion() { return 1; } Override public byte[] serialize(XATransaction obj) throws IOException { return (obj.xid \u0001 obj.jdbcUrl \u0001 obj.user \u0001 obj.password) .getBytes(StandardCharsets.UTF_8); } Override public XATransaction deserialize(int version, byte[] bytes) throws IOException { String[] parts new String(bytes, StandardCharsets.UTF_8).split(\u0001, -1); return new XATransaction(parts[0], parts[1], parts[2], parts[3]); } }核心 Sink 类继承 TwoPhaseCommitSinkFunction再把各个回调方法和 XA 命令接起来import com.mysql.cj.jdbc.MysqlXADataSource; import org.apache.flink.api.common.typeutils.base.VoidSerializer; import org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunction; import javax.sql.XAConnection; import java.sql.Connection; import java.sql.Statement; import java.util.UUID; public class FlinkXaSink extends TwoPhaseCommitSinkFunctionTuple2String, Integer, XATransaction, Void { private final String jdbcUrl; private final String user; private final String password; public FlinkXaSink(String jdbcUrl, String user, String password) { super(new XaTransactionSerializer(), new VoidSerializer()); this.jdbcUrl jdbcUrl; this.user user; this.password password; } Override protected XATransaction beginTransaction() throws Exception { XATransaction tx new XATransaction(UUID.randomUUID().toString(), jdbcUrl, user, password); tx.connection createConnection(); tx.connection.setAutoCommit(false); exec(tx.connection, XA START tx.xid ); return tx; } Override protected void invoke(Tuple2String, Integer value, Context context) throws Exception { XATransaction tx currentTransaction(); try (Statement st tx.connection.createStatement()) { st.execute(INSERT INTO sink_log(biz_key, val, xid) VALUES ( value.f0 , value.f1 , tx.xid )); } } Override protected void preCommit(XATransaction transaction) throws Exception { exec(transaction.connection, XA END transaction.xid ); exec(transaction.connection, XA PREPARE transaction.xid ); } Override protected void commit(XATransaction transaction) throws Exception { Connection conn transaction.connection ! null ? transaction.connection : createConnection(); exec(conn, XA COMMIT transaction.xid ); if (transaction.connection ! null) { transaction.connection.close(); } } Override protected void abort(XATransaction transaction) throws Exception { if (transaction.connection ! null) { exec(transaction.connection, XA ROLLBACK transaction.xid ); transaction.connection.close(); } } Override protected void recoverAndCommit(XATransaction transaction) { try { Connection conn createConnection(); exec(conn, XA COMMIT transaction.xid ); conn.close(); } catch (Exception e) { throw new RuntimeException(recoverAndCommit failed, e); } } Override protected void recoverAndAbort(XATransaction transaction) { try { Connection conn createConnection(); exec(conn, XA ROLLBACK transaction.xid ); conn.close(); } catch (Exception e) { throw new RuntimeException(recoverAndAbort failed, e); } } private Connection createConnection() { try { MysqlXADataSource ds new MysqlXADataSource(); ds.setUrl(jdbcUrl); ds.setUser(user); ds.setPassword(password); XAConnection xaConn ds.getXAConnection(); return xaConn.getConnection(); } catch (Exception e) { throw new RuntimeException(create XA connection failed, e); } } private void exec(Connection conn, String sql) throws Exception { try (Statement st conn.createStatement()) { st.execute(sql); } } }这里有两个地方需要强调。第一事务句柄里不要放 Connection 对象参与序列化否则 checkpoint 阶段会直接报序列化失败。第二恢复阶段执行的 recoverAndCommit 会重新建连接用反序列化回来的 xid 去提交所以即使之前进程被杀掉只要 xid 一致MySQL 就能找回那个 Prepared 事务。4.3 开启 Flink 侧事务日志和检查点参数验证过程中我们想看到 TwoPhaseCommitSinkFunction 内部的状态迁移光靠 INFO 日志是不够的需要把相关包的日志级别调到 DEBUG。在 Flink 的 log4j2 配置里增加下面这行logger.two_phase_sink.name org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunction logger.two_phase_sink.level DEBUG检查点参数方面为了保证观察窗口明显建议把间隔设置得短一些比如 5 秒。同时选择 EXACTLY_ONCE 模式。自己验证环境可以把并行度设为 1减少多个并发事务同时出现带来的干扰。主程序配置大致如下StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.enableCheckpointing(5_000L); env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); env.setParallelism(1); DataStreamTuple2String, Integer stream env.addSource(yourKafkaSourceOrGenerator); stream.addSink(new FlinkXaSink(jdbc:mysql://localhost:3306/eos_test, root, root_password)); env.execute(Flink EOS MySQL 2PC Verify);source 建议用 Kafka并且确认 offset 是通过 Flink checkpoint 管理的也就是不要开启 checkpoint 之外的独立 offset 提交。这样才能保证数据重放是精确的。5. 验证实录三条证据链交叉确认5.1 证据一Flink 日志里的事务生命周期任务启动之后把日志切到 DEBUG 级别观察。正常情况下你会看到这样的事务推进节奏每隔一个 checkpoint 周期就会有一次 beginTransaction、一次 preCommit、一次 commit。这说明 Sink 严格按照两阶段提交的流程在走。如果某个 checkpoint 失败日志里会出现 abort 调用同时下游 MySQL 中对应的 XA 事务会被回滚。这个日志链路非常直观相当于 Flink 亲口告诉你我正在按照 2PC 协议驱动整个提交过程。我建议你在日志里搜索关键字 “Pre committing” 和 “Committing transaction”看看这两条日志的时间和 checkpoint 完成时间之间的关系。按照设计Pre committing 会出现在 barrier 处理阶段Committing transaction 会出现在 checkpoint 完成之后。顺序对了协议才算真正生效。5.2 证据二MySQL XA RECOVER 与 binlog这是在数据库侧最有力的证据。在 checkpoint 还在进行中的那几秒内去 MySQL 执行 XA RECOVER会看到一批处于 Prepared 状态的事务它们的 xid 跟 Flink 日志里的 XID 能对得上。执行完 XA RECOVER 之后等到 checkpoint 完成再执行一次这批 Prepared 事务应该消失因为它们已经被 XA COMMIT 提交了。如果任务崩溃没有恢复这批事务会一直停留在 Prepared 状态等待外部协调者处理。binlog 侧的证据也同样重要。查看当前 binlog 事件你会看到与 XID 绑定的 PREPARE 和 COMMIT 记录。这说明 MySQL 已经把两阶段提交的边界写进了自己的事务日志后续即使连接断开MySQL 也能根据 binlog 恢复 XA 事务状态。没有 binlog 支撑的话这里是没办法做到自动恢复的。SHOW MASTER STATUS; SHOW BINLOG EVENTS IN mysql-bin.000001; XA RECOVER;5.3 证据三故障注入后的恢复表现验证 2PC 最核心的场景是“任务中途被干掉恢复之后数据不重不丢”。我建议做两次故障注入一次温和一次暴力。温和的做法是在 Flink 日志里看到某个事务 preCommit 完成之后、commit 还未触发之前用 flink cancel 停掉作业然后再从最近一次 checkpoint 恢复。恢复之后观察日志会出现 recoverAndCommit 调用把那个已经 Prepared 的事务提交掉。MySQL 侧不会多数据也不会少数据。暴力的做法更直接在同样的时间窗口里 kill -9 掉 TaskManager 进程。整个集群没有机会执行优雅的停止逻辑Flink 只能依靠 checkpoint 来恢复。这种情况下恢复动作是否干净完全取决于 Sink 是否有能力识别并处理那些“半提交”的事务。如果 2PC 链路是通的恢复后 MySQL 中依然不会出现数据重复。5.4 最终数据核对count 与 sum 对不上就是没走完 2PC日志、XA RECOVER、binlog 都是过程证据最终还是得看数据本身。建议在 sink_log 表上用 SQL 统计总条数和总和和 source 侧实际发送的数据量做比对。SELECT COUNT(*), SUM(val), COUNT(DISTINCT biz_key) FROM sink_log;任务每重启一次就在 Flink 侧记录一个实际处理到的 offset 位置然后在 MySQL 侧核对 count 和 sum。如果重启前后数值一致说明端到端精确一次是真的如果重启后 count 变大了说明 Sink 端重复写入如果 count 变小了说明有事务被回滚但 Flink 没有重放。我自己的习惯是连续做三次故障注入每次都记录恢复前的值和恢复后的值把结果列成一张表。三次都一致才敢对业务说“这个链路是精确一次的”。6. 常见问题与排查经验6.1 异常速查表实际跑这个验证时大概率会遇到下面几个问题。我把高频问题和排查方向整理成一张表。现象可能原因排查方向XAER_NOTA 异常XA 事务不存在检查 XID 是否一致连接串是否指向同一实例事务是否已被超时清理checkpoint 一直处于 pending 状态Sink 的 preCommit 阻塞查看数据库锁等待检查是否有大事务未提交恢复后数据重复Sink 没走事务或恢复逻辑不对确认使用的是 XA 事务检查 recoverAndCommit 是否执行MySQL 里出现大量 Prepared 残留事务超时或协调者失联用 XA RECOVER 查看必要时人工执行 XA ROLLBACKXA 事务提交时报权限不足账号缺少权限给账号授予 XA_RECOVER_ADMIN 权限恢复后找不到待提交事务binlog 未开启或 ROW 格式未设置检查 binlog 配置确认事务已经完整记录6.2 几个实操中容易忽略的细节事务超时是很多人忽略的点。MySQL 的 innodb_lock_wait_timeout 和 XA 事务本身的超时机制会共同作用如果 checkpoint 间隔太长事务长期停留在 Prepared 状态可能会被 MySQL 自动清理。到时候 Flink 再去提交就会得到 XAER_NOTA。解决办法是合理设置 checkpoint 间隔别让 Prepared 状态悬挂太久。锁问题也值得重视。XA 事务在 Prepared 阶段仍然持有已经写入的行锁如果同一张表有其他写操作它们会一直等锁直到超时。所以 2PC 并不适合在单事务里塞过多数据业务侧要控制每个 checkpoint 周期内的写入量。验证环境里数据量小没关系生产环境必须评估。最后提一个兜底思路。2PC 能解决大多数场景的精确一次问题但它不是万能的。极端情况下比如网络分区导致协调者和数据库短暂失联MySQL 侧可能已经完成提交但是协调者没有收到确认消息。这种时刻光靠 2PC 还不够建议在下游表里加业务唯一键做幂等兜底。两条机制叠加才是真正稳妥的方案。结尾一点实际感触这套验证流程我前后跑了不下十次最深的体会是2PC 不是靠配置开出来的而是靠外部系统一层一层协议堆出来的。你光在 Flink 里选一个 exactly-once 选项MySQL 不配合就完全没有意义。每次看到 XA RECOVER 里那批 Prepared 事务在 checkpoint 完成后整整齐齐消失心里才踏实。最后分享一个小技巧。验证时可以把 xid 直接写进业务表就像我上面代码里那样。以后你想排查某个 checkpoint 周期的事务边界按 xid 分组一查就知道哪些记录属于哪个事务哪批数据是恢复后补写的整个生命周期肉眼可见。这个习惯后来帮我解决了不少线上对账问题推荐你也试试。