事务与 Exactly-Once:跨分区原子写入

幂等解决不了什么

第 03 篇的幂等生产者保证了:一次会话内,同一分区的重试不会产生重复消息.但把视角升到"一次完整的业务处理",还剩三个缺口:

consume-transform-produce 流水线:
  1. 从 input-orders 读一条
  2. 处理(算价、风控)
  3. 写 output-orders
  4. 提交 input 的 offset

幂等只能管"同一次会话内、同一个分区的重试",三个缺口都在它的边界之外.事务就是为补齐这三个缺口设计的:把"多个分区的写入 + offset 提交"打包成一个原子单元.

事务的三个角色

Producer ──────► Transaction Coordinator(某个 Broker)
   │                     │
   │                     │ 事务状态写进 __transaction_state(内部 topic)
   │                     ▼
   ├─ 写 output 的 P0, P1 (Broker 侧记录事务标记)
   └─ 提交 input offset (第 04 篇的提交, 现在属于事务)

内部 topic __transaction_state 又是"内部状态也是日志"的套路(第 08 篇的 __consumer_offsets、第 10 篇的 __cluster_metadata):事务的状态与结果,本身就是一条有序日志.

事务的完整流程

1. InitTransactions
   Producer ──► Coordinator: 注册 TransactionalID, 拿到 PID, epoch 加一
   (旧实例若还活着: epoch 落后 → 被 Fencing)

2. BeginTransaction
   本地动作, 无需 RPC

3. 写入阶段
   首写某个分区前, 先向 Coordinator 登记该分区(AddPartitionsToTxn)
   然后 Producer ──► Broker: 写入 output 各分区的数据

4. SendOffsetsToTransaction
   把 input 的 offset 提交也纳入本事务

5. 提交 / 回滚(两阶段)
   Coordinator 先写 PREPARE_COMMIT 到 __transaction_state
      → 向所有参与分区写 commit marker(控制消息)
      → 写 COMPLETE_COMMIT, 事务结束

两个值得记住的细节:

fencing:同 TransactionalID 的新实例启动时 epoch 加一,旧实例的任何写入都会被 Broker 以 FENCED_INSTANCE_ID 拒绝.滚动发布中新旧实例短暂并存,旧实例自动失效,保证任一时刻只有一个"有效写者".这既是安全机制,也是"实例级唯一 ID"这个约束的由来.

控制消息(commit / abort marker):marker 也是写进分区日志的条目,占用 offset 空间,但对普通消费者不可见(被过滤).回看第 02 篇:offset 只增不减,事务"反悔"用的不是删除,而是"写入标记,让读者跳过".

隔离级别:读者怎么过滤

事务写下的数据,在提交前对消费者可见吗?由消费者的隔离级别决定:

read_committed 的实现靠 LSO(Last Stable Offset,最后稳定偏移):

Partition 日志:

[普通][事务A: 已提交][事务B: 进行中 ][事务C: 已提交]
                          ▲
                    LSO 卡在这里

read_committed 消费者最多读到 LSO:
  ● 事务 A 和它之前的普通消息: 可读
  ● 事务 B 及其之后(含已提交的 C): 读不到

一个未完成的事务会把 LSO 卡住,后面的已提交数据一起被挡住.所以长事务会阻塞读,这也是 transaction.timeout.ms(默认 60s)存在的意义:超时未提交的事务被协调者强制 abort,LSO 放行.

franz-go 里声明隔离级别:

kgo.FetchIsolationLevel(kgo.ReadCommitted()),

Go 实战:consume-transform-produce

franz-go 提供了 GroupTransactSession,把"事务 + 消费组 + offset 提交进事务"的样板封装好了:

session, err := kgo.NewGroupTransactSession(
    kgo.SeedBrokers("localhost:9092"),
    kgo.TransactionalID("order-etl-0"),          // 每实例唯一, 见下节
    kgo.ConsumerGroup("order-etl"),
    kgo.ConsumeTopics("input-orders"),

    // 读上游也必须是 read_committed, 否则可能读到将被回滚的数据
    kgo.FetchIsolationLevel(kgo.ReadCommitted()),
)
if err != nil {
    log.Fatal(err)
}

for {
    fetches := session.PollFetches(ctx)
    if fetches.IsClientClosed() {
        return
    }
    fetches.EachError(func(topic string, partition int32, err error) {
        slog.Error("fetch error", "topic", topic, "partition", partition, "err", err)
    })
    if fetches.NumRecords() == 0 {
        continue
    }

    fetches.EachRecord(func(r *kgo.Record) {
        result := transform(r)

        // 写入下游: 自动属于当前事务
        session.Produce(ctx, &kgo.Record{
            Topic: "output-orders",
            Key:   r.Key,
            Value: result,
        }, nil)
    })

    // 提交事务: 下游写入 + 上游 offset 一起原子提交
    if err := session.EndTransaction(ctx, kgo.TryCommit); err != nil {
        // 提交失败(网络、协调者切换等): 事务被 abort
        // 下游无痕, offset 未推进 → 下一轮重新拉取处理, 不产生重复输出
        slog.Error("transaction failed", "err", err)
        continue
    }
}

对照第 06 篇手写 BeginTransaction / EndTransaction 的版本,GroupTransactSession 处理了三件容易写错的事:

1. offset 提交自动进入事务(不用手工凑 SendOffsetsToTransaction 的时机)
2. 提交失败自动 abort,事务不留半成品
3. 下一轮循环自动开启新事务

事务的代价与使用约束

三条工程约束:

1. TransactionalID 实例级唯一且稳定:两个实例共用一个 ID 会互相 fencing,循环重启;K8s 上用 Pod 名(与第 11 篇静态成员同一思路)
2. 事务不覆盖外部系统:数据库、Redis、HTTP 调用不在事务里,这部分靠业务幂等或 Outbox 模式(第 06 篇策略三,阶段三第 20 篇展开)
3. 只有 consume-transform-produce 模式享受完整 EOS:Kafka 进、Kafka 出的链路可以端到端不丢不重;出到外部系统的链路,事务只能保证 Kafka 侧

生产注意事项

阶段二(架构原理与调优)七篇至此完成:存储引擎(08)、副本与 ISR(09)、Controller 与元数据(10)、分区再平衡(11)、生产调优(12)、消费调优(13)、事务与 EOS(14)."Kafka 为什么快、为什么可靠、怎么调"的完整图景已经拼齐.

下一阶段(阶段三 · 流处理与高级应用)从第 15 篇开始:跳出"消息队列"的视角,看 Kafka 作为事件流平台的设计哲学,Log 即数据库.