4. 事务边界定义：beginTransaction、commitTransaction、abortTransaction 的 API 与生命周期

Kafka 事务的核心在于精确地定义事务的开始、提交与终止三个边界。这三个 API 调用构成了事务的完整生命周期，任何跨分区的原子写入操作都必须严格遵循这一边界协议。理解这些 API 的语义、阻塞行为以及异常处理路径，是构建可靠 Exactly-Once 语义（EOS）生产者的前提。

4.1 事务生命周期总览

一个 Kafka 事务从初始化到最终完成，经历以下四个阶段：

1. 初始化阶段：通过 initTransactions() 注册 transactional.id 并获取 Producer ID（PID）与 Epoch。
2. 事务开始：调用 beginTransaction() 标记事务起点。
3. 数据写入与消费：在事务内执行 send() 或 sendOffsetsToTransaction()。
4. 事务结束：调用 commitTransaction() 提交或 abortTransaction() 终止。

4.2 核心 API 详解

4.2.1 beginTransaction()

beginTransaction() 是一个轻量级本地操作，它不会向 Broker 发送任何网络请求。其作用是在生产者客户端内部将状态从 READY 切换为 IN_TRANSACTION，并递增本地事务序列号。调用此方法后，所有后续的 send() 请求都会携带当前事务的 PID 与 Epoch。

4.2.2 commitTransaction()

commitTransaction() 是一个阻塞式协调操作，它执行以下步骤：

1. Flush 所有待发送消息：确保事务内的所有 send() 请求已被 Broker 确认。
2. 提交事务协调器：向事务协调器（Transaction Coordinator）发送 EndTxn 请求，标记为 COMMIT。
3. 写入 COMMIT 标记：协调器在事务日志中写入 COMMIT 标记，并通知所有参与的分区 Leader 写入控制消息（Control Batch）。
4. 状态切换：生产者状态从 IN_TRANSACTION 回到 READY。

4.2.3 abortTransaction()

abortTransaction() 的流程与提交类似，但最终写入的是 ABORT 标记。消费者端在配置 isolation.level=read_committed 时，会自动过滤掉被终止事务写入的消息。

4.3 API 参数与异常对照

API	阻塞行为	超时配置	常见异常
beginTransaction()	非阻塞（本地操作）	无	IllegalStateException（状态错误）
commitTransaction()	阻塞，等待协调器确认	transaction.timeout.ms（默认 60000ms）	TimeoutException、ProducerFencedException
abortTransaction()	阻塞，等待协调器确认	transaction.timeout.ms	TimeoutException、KafkaException

4.4 事务生命周期中的状态机

Kafka 生产者客户端内部维护了一个严格的状态机，确保事务 API 的调用顺序合法。以下状态转换是允许的：

• READY → IN_TRANSACTION：通过 beginTransaction()。
• IN_TRANSACTION → READY：通过 commitTransaction() 或 abortTransaction()。
• IN_TRANSACTION → FENCED：当检测到 Epoch 冲突时（另一个生产者实例使用了相同的 transactional.id）。

4.5 代码示例：完整的事务边界使用

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("transactional.id", "my-transactional-id");
props.put("enable.idempotence", "true");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

// 1. 初始化事务（仅需一次）
producer.initTransactions();

try {
    // 2. 开始事务
    producer.beginTransaction();

    // 3. 事务内发送消息
    producer.send(new ProducerRecord<>("topic-a", "key1", "value1"));
    producer.send(new ProducerRecord<>("topic-b", "key2", "value2"));

    // 4. 提交事务
    producer.commitTransaction();
    System.out.println("事务提交成功");
} catch (ProducerFencedException e) {
    // 生产者被 fencing，必须关闭重建
    producer.close();
    throw new RuntimeException("生产者被 fencing", e);
} catch (KafkaException e) {
    // 其他异常，终止事务
    producer.abortTransaction();
    System.err.println("事务终止: " + e.getMessage());
} finally {
    producer.close();
}

4.6 事务边界对性能的影响

事务边界的定义直接影响吞吐量与延迟。以下对比展示了不同事务粒度下的性能特征：

事务粒度	每秒事务数（TPS）	端到端延迟（P99）	适用场景
每条消息一个事务	低（约 500-2000）	高（>100ms）	金融交易、严格有序
批量消息一个事务	中（约 5000-20000）	中（20-50ms）	日志聚合、CDC 同步
无事务（幂等生产者）	高（>50000）	低（<5ms）	高吞吐日志、监控数据