Appearance
订单消息处理完了，Kafka 为什么还会再投一次？
订单服务收到一条 order-created 消息，把订单处理记录写进数据库，然后准备告诉 Kafka：“这条我处理好了。”偏偏就在这时，进程退出。服务重启，同一条消息又来了。
如果只看业务数据库，会觉得 Kafka 在“重复投递”；如果只看 Kafka，又会觉得消费者从来没确认过它。两边说的都没错。问题出在“处理完”这三个字里藏着两个互不相干的动作：数据库提交了业务结果，Kafka 提交了消费位置。它们不是同一笔事务。
这期不从“消息队列有哪些交付语义”开始背概念。我在本地启动了 Kafka 4.3.1，把消费者停在几个不同的时间点，分别看数据库行数、Kafka 已提交的 offset，以及重启后能不能再拿到那条消息。实验代码和完整运行方法放在后面。先把容易混淆的两根进度条说清楚，不然实验输出也不好读。
读到了 offset 0，不等于已经确认 offset 0
Kafka 的一个分区里，消息有顺序编号。这里用最小的情况：一个 topic、一个分区、第一条消息的 offset 是 0。消费者调用 poll() 拿到它后，客户端的当前位置会前进到 1，意思是“在这个进程里，下次从 1 往后读”。但消费者组的已提交位置可能还不存在，或者仍停在 0。当前位置是进程内读到哪里，已提交位置是进程失败后从哪里恢复。
我用的 Kafka 4.3.1 客户端源码里，poll 返回记录之前，消费位置已经更新；无参 commitSync() 又会取客户端记录的“已消费位置”去提交。两个动作被放在不同的方法里，正说明 poll 本身不是业务确认。实验第一次收到消息时打印出：
text
received offset=0 position=1 committed=none这里的 position=1 很容易误导人。它只描述当前客户端的游标，Kafka 还不知道数据库有没有写完。若进程这时退出，后继消费者会按已提交位置恢复，而不会继承前一个进程内存里的 position=1。
再强调一个容易写错的数字：处理了 offset 0，手动提交时填的是 1，也就是下一条希望读取的 offset。填 0 的意思是下次还从 0 开始。这个约定在批处理时尤其重要，因为“本批最后一条”与“下一条”只差 1，日志里看起来却都像合理的数字。
把 poll、业务提交和 offset 提交排成一行
正常路径看起来很短：
text
poll 得到 offset 0
→ 数据库事务提交 event-1
→ commitSync 提交 offset 1偏偏这个顺序里有两个裂缝。第一条裂缝在 poll 和数据库提交之间：消息取到了，业务没做成。第二条裂缝在数据库提交和 offset 提交之间：业务做成了，Kafka 不知道。
有人会问，那把最后两步换个顺序呢？先提交 offset 1，再写数据库。这样确实不会因为数据库写完后进程退出而重投，但会打开另一条更危险的裂缝：Kafka 已经认为消息处理完了，数据库却还没有记录。进程在这时退出，重启后消息不再回来，业务结果也没有。
这不是哪一个 API 不够聪明。commitSync() 操作的是 Kafka 保存的消费者组位置，业务事务操作的是另一套数据库。只给其中一边提交成功，另一边随后失败，两个系统就会短暂或永久地看到不同的进度。单靠调整两个方法的调用顺序，消不掉这个边界。
我把进程停在了这三个位置
实验使用 Kafka 4.3.1 单节点 broker、Java 客户端和本地 H2 文件数据库。每个场景用新的 topic 和消费者组，分区数为 1，只生产一条 event-1。关闭自动提交；业务表把 event_id 设为主键。消费者进程拿到消息后，可以按参数在指定位置调用 Runtime.halt()，直接结束 JVM，不走正常的 close() 路径。
这不是生产部署模板。单节点、较短的心跳和会话超时，以及同步刷盘的 H2，都是为了让故障点能在本地看见。Kafka 的 broker 与客户端版本固定为 4.3.1；不同版本和消费组协议的细节要以各自的源码及测试为准。
实验的核心代码可以缩成下面一段。完整的 OffsetFailureLab.java 包括建 topic、生产消息、H2 建表、查询行数和完整参数，文章末尾有运行命令。
java
ConsumerRecord<String, String> record = consumer.poll(Duration.ofMillis(500))
.iterator().next();
TopicPartition partition = new TopicPartition(record.topic(), record.partition());
if (mode.equals("crash-before-db")) Runtime.getRuntime().halt(11);
if (mode.equals("commit-before-db")) {
consumer.commitSync(Map.of(partition,
new OffsetAndMetadata(record.offset() + 1)));
Runtime.getRuntime().halt(12);
}
// 同一事务写业务表；event_id 有唯一约束。
writeBusinessRow(connection, record);
if (mode.equals("crash-after-db")) Runtime.getRuntime().halt(13);
consumer.commitSync(Map.of(partition,
new OffsetAndMetadata(record.offset() + 1)));为了让“数据库已经提交”在强制结束进程之后仍然可验证，完整程序显式提交 H2 事务，再执行同步检查点。这个小细节是实际跑实验时补上的：最初版本只在 executeUpdate() 返回后打印“写入成功”，进程强退后数据库却查不到那行。那次输出只能证明 SQL 执行到了那一步，不能证明故障后可恢复。修改实验装置、重新用一个新 topic 测，才得到下面的结果。真实业务也一样：别把“方法返回了”当成“两个系统都持久化了”。
| 停在哪一步 | 强退后数据库 | Kafka 已提交 offset | 再启动消费者 |
|---|---|---|---|
| 数据库写入前 | 0 行 | 无 | 再读到 offset 0，正常写入 |
| 数据库提交后、offset 提交前 | 1 行 | 无 | 再读到 offset 0，触发唯一键冲突，随后提交 offset 1 |
| offset 提交后、数据库写入前 | 0 行 | 1 | 20 秒内未再读到该消息 |
表里的“20 秒内未再读到”是本次实验观察，不是对 Kafka 永远不会重投的数学证明；同一消费者组从已提交 offset 1 恢复时，这条 offset 0 消息确实已经被越过去了。可以手动换一个新的消费者组重读历史消息，但那已经是另一种恢复操作，不能指望原消费者组自动完成。
数据库写完后重投，问题出在哪儿
第二行最接近开头的订单故障。第一次进程输出是：
text
received offset=0 position=1 committed=none mode=crash-after-db
database=committed event-1进程退出后，H2 里确实有 1 行，消费者组仍没有已提交 offset。重启同一消费者组，得到：
text
received offset=0 position=1 committed=none mode=normal
database=duplicate event-1
offset=committed after database write查数据库还是 1 行；再查消费者组，CURRENT-OFFSET=1、LOG-END-OFFSET=1、LAG=0。这组输出很清楚地把两件事分开了：消息确实又送到了应用代码，业务效果没有重复发生。挡住第二次插入的是数据库的唯一约束，而不是 Kafka 发现“这张订单已处理过”。
把第二次插入的主键换成一个新随机数，或者完全不设唯一约束，重复业务结果就会出现。因此幂等键必须来自这次业务操作本身，比如稳定的事件 ID 或订单操作 ID；重试时重新生成一个 UUID，等于把同一件事伪装成两件事。若同一 ID 的消息内容变了，还要比对参数摘要或业务版本，不能只说“主键已存在”就把不一致的内容吞掉。
这也是为什么我更愿意说“至少投递一次，加业务去重”，而不在这里写“消费恰好一次”。数据库一行不等于消费只发生一次：日志里明明有两次 received offset=0。唯一约束只保护这张表。若程序在插入这张表之前已经调用支付接口、发了短信，或者在冲突处理前做了别的外部动作，那些动作仍可能重复。需要把去重记录与业务写入放进同一事务；跨到支付、通知等系统时，还要看各自的幂等接口、状态查询或 Outbox 机制。
先提交 offset，会把哪一步弄丢
第三行是顺序倒过来的结果。消费者拿到 offset 0，先调用 commitSync(offset=1)，然后在数据库写入前退出。此时消费者组显示 CURRENT-OFFSET=1，数据库是 0 行。重启程序，等了 20 秒没有拿到消息，程序报 No record received。
从 Kafka 的角度看，这个消费者组已经完成了 offset 0；它不会因为 H2 里没有订单处理记录，就回头查业务数据库。若后台页面只根据 Kafka lag 判断健康，看到的还是 LAG=0。这就是“进度条看起来追平，业务却漏了”的典型陷阱。监控消费积压能告诉你处理到了哪里，不能替你证明业务处理成功了什么。
同样的危险不一定非要手写“先 commit 后写库”才出现。自动提交开启后，提交发生在客户端的节奏里，而业务可能在另一个线程处理，或者一批记录只完成了一半。只要已提交位置跑到尚未完成的业务前面，故障后就有漏处理窗口。关闭自动提交是必要的控制手段，但手动提交也不会自动替你算“哪些记录已经真正完成”。
处理慢了，旧消费者还能把 offset 交上去吗
崩溃不一定是进程死掉。还有一种更绕的情况：消费者拿到消息，数据库已经提交，业务代码却长时间没有再次调用 poll()。Kafka 用 max.poll.interval.ms 限制两次 poll 之间的最长间隔，避免一个进程一直占着分区但不继续推进。测试里把它设为 5 秒；第一个消费者写库后暂停 12 秒，暂停期间启动同组第二个消费者。
第二个消费者又读到 offset 0，插入业务表时遇到 event-1 的唯一键冲突，然后提交 offset 1。第一个消费者睡醒后再 commitSync()，得到 CommitFailedException，错误说明它已经不是活跃组成员。最后数据库仍是 1 行，消费者组 offset 为 1。
这里有个值得留意的先后关系：失去分区所有权，不会自动撤销已经提交到外部数据库的事务。Kafka 可以拒绝旧成员提交 offset，却无法回滚那张 H2 表。若旧成员在失去资格后还继续调用外部服务，第二个消费者也开始处理同一事件，就会出现并发的重复业务。业务幂等不是“偶发重试时锦上添花”，而是消费者组再均衡后守住外部副作用的基本条件。
生产系统里还要看静态成员、所用消费组协议和会话超时。max.poll.interval.ms 到期后分区具体何时转交，不应只凭一个固定秒数推断。本实验明确设了 classic 协议、6 秒 session timeout 和 1 秒 heartbeat；换协议和配置要重新测。面对耗时不稳定的处理，简单把 max.poll.interval.ms 调得很大也有代价：真正卡死的消费者会更久才交出分区。可以限制单次 poll 的记录数，或把处理放到工作线程并暂停相关分区，但后者必须追踪每个分区连续完成的位置，绝不能因为线程池里某条较新的记录先完成，就把前面尚未完成的记录一并确认。
批量处理时，提交哪个数字
单条消息把 offset 加 1 很好算。若一次 poll 返回 offset 10、11、12，程序先完成了 10 和 12，11 仍在处理，此时能提交 13 吗？不能。提交 13 表示重启后从 13 读，11 会被跳过去。最远只能提交 11，表示 10 已完成，11 还要再来。等 11、12 都完成，再提交 13。
多分区时这本账还要分开算。分区 A 完成到 offset 20，不表示分区 B 的 offset 20 也完成；两个分区的编号没有横向比较意义。批量提交应按 TopicPartition 分别记录“已连续完成的最后一条”，然后提交它的下一位。若业务程序把多分区消息扔给线程池处理，记录完成顺序可能与 Kafka 分区顺序不同；这时只维护一个“最大的已完成 offset”反而是错的。
Kafka 4.3.1 的 commitSync() 有无参和显式 map 两种写法。无参版本会从客户端的已消费位置取值，而 poll 返回不代表你的数据库事务已完成，所以这里显式传入 Map<TopicPartition, OffsetAndMetadata>，让代码审查时能看见“究竟确认了哪个分区的哪个位置”。这仍要求我们自己算对值，也不能让数据库提交与 offset 提交变成原子操作。
如果用异步提交，还得处理回调里的失败。它不会像同步提交那样在当前调用处直接挡住后续流程。程序退出时常会补一次同步提交，但这只是把先前的异步请求收尾，并没有改变业务数据库与 Kafka 之间的事务边界。把“提交 API 返回了”当成“本批处理全成功”，和把 poll 当成确认，是同一种偷换。
这里还有一种比较烦人的中间状态：数据库已经提交，commitSync() 却抛了超时异常。业务代码拿到异常时，不能立刻断言 Kafka 一定没收到 offset。提交请求可能没有送到协调器，也可能协调器已经写入成功，只是响应没有回来。下一次启动时，先看消费者组实际从哪里恢复；若再次收到这条消息，按事件 ID 查询数据库，确认之前的业务结果，再推进 offset。这个流程和接口超时后怎么重试是同一个道理：通信失败，只能说明客户端不知道结果。
消费者组提交的位置保存在 Kafka 侧，重启后的进程会从那里恢复。数据库的处理标记则保存在业务库。把它们分别记录到日志里，比只打一行“消费成功”有用得多：故障复盘时可以沿着事件 ID 看到“收到了哪一条、业务事务是否提交、尝试提交哪个 offset、提交结果是否可确认”。这样才能区分重复投递、业务重复执行和进度提交失败。三者听起来很像，处置却完全不同。
Kafka 自己的事务，能不能把数据库也包进去
Kafka 的事务能把写往 Kafka topic 的消息与消费 offset 放在同一个 Kafka 事务里。例如消费输入 topic、生成输出 topic，再用事务提交输出与输入位置，这是一条有明确边界的 Kafka 内部处理链。若输出改成 MySQL、H2 或支付接口，Kafka 事务不会顺手把那个外部系统的提交也纳入进来。
要达到“业务结果只生效一次”，得先定义结果是什么。对本文这张表，可以在一个数据库事务里同时写入业务行与事件处理标记，靠稳定的事件 ID 和唯一约束使重投变成可识别的重复。数据库提交后再提交 Kafka offset；即使第二步失败，重投也只会重新检查和确认，不再重复业务效果。反过来先提交 offset 再写库，会留下漏处理窗口，所以不能为了少看见重复日志就这么做。
如果还要往别的系统发消息，可以把待发布事件作为 Outbox 行与业务更新一起提交，然后由独立发布器可靠地送往 Kafka。发布器也可能重试，因此下游仍要识别重复。对支付这类外部调用，要使用其稳定的业务请求号和结果查询机制；单靠消费端数据库唯一键，不能保证第三方只收了一次款。讨论“恰好一次”时，务必说明哪一个系统里的哪一种效果只发生一次，别把整个跨系统流程含糊地包进四个字。
真要上线，先把这几种失败跑出来
这次实验只处理一条、一个分区、一个数据库表，目的是把时序看清。真实消费者通常还要覆盖这些情况：
- 同一个事件 ID 的两条消息同时到达两个进程，数据库唯一约束是否仍只留下一个业务结果；另一方如何识别“已处理”与“同 ID 但参数不一致”。
- 业务事务已经提交，进程在 offset 提交前退出；重启后能否查到原结果，并最终推进 offset。
- offset 提交请求超时，客户端不知道提交是否成功；恢复时按 Kafka 实际已提交位置与业务去重记录处理，不能凭异常文本断言“肯定没提交”。
- 一批消息中间有一条一直失败；是否暂停该分区、如何重试或转入待处理队列，避免跳过它时还把后面的 offset 提前提交。
- 消费者在再均衡后仍运行旧的异步任务；任务落库或调用外部接口时，是否有业务幂等或更严格的所有权保护。
监控也应分两层。Kafka 的消费延迟、组成员变化和提交失败告诉你读取进度；业务库中的处理成功数、去重命中数、待确认事件数告诉你效果。只盯 lag=0，就会漏掉前面那个“offset 已到 1、数据库仍是 0 行”的场景。去重命中突然上升也值得追查：可能是一次正常的故障恢复，也可能是消费处理过慢导致频繁再均衡。
归根结底，Kafka 的 offset 是读取进度，业务数据库记录的是处理结果。它们之间的时间差无法靠一句“处理成功后再提交”彻底消失，但可以被设计得可恢复：先让业务操作可识别、可去重，再推进 offset，并把每个故障窗口实际跑一遍。这样重投来了，你知道该查什么；没重投时，也不会因为进度追平就误以为订单一定处理成功。
实验怎么跑
下载并校验 Apache Kafka 4.3.1 二进制发布包，用本机 Java 17 或更高版本按 Kafka 4.3 Quick Start 启动单节点 broker。本实验把客户端端口设为 127.0.0.1:19092。再准备 H2 2.4.240 的 jar，编译上面的完整程序。下面的命令假设 KAFKA_HOME、H2_JAR 已指向各自文件，且当前目录放着下载的 OffsetFailureLab.java：
bash
mkdir -p classes
javac -cp "$KAFKA_HOME/libs/*:$H2_JAR" -d classes OffsetFailureLab.java
export LAB_CP="classes:$KAFKA_HOME/libs/*:$H2_JAR"
# 每组实验使用一个没用过的 topic 名称；seed 会建一分区 topic 并生产一条消息。
java -cp "$LAB_CP" OffsetFailureLab seed demo-after-db-001
java -cp "$LAB_CP" OffsetFailureLab consume demo-after-db-001 crash-after-db
# 上一步故意以 13 退出。等待旧成员退出消费组，再启动同组消费者。
java -cp "$LAB_CP" OffsetFailureLab status demo-after-db-001
java -cp "$LAB_CP" OffsetFailureLab consume demo-after-db-001 normal
java -cp "$LAB_CP" OffsetFailureLab status demo-after-db-001
# 把 crash-after-db 换成 commit-before-db，可观察另一侧的失败窗口。程序把 H2 数据库放在 /tmp/kafka-seo-db-<topic>。不要在同一个 topic 上重复执行 seed：它用于新建实验场景，topic 已存在时会报错。程序里的 Runtime.halt() 也确实会跳过 JVM 的正常清理，只在本地实验环境运行。想观察处理过慢的场景，用 slow-after-db 启动第一个消费者，看到它打印 pausing 12 seconds 时，再启动第二个同组消费者执行 normal。