Kafka 中的事务

Kafka 中的 事务（Transactions） 是为了解决 消息处理的原子性和幂等性问题，确保一组消息要么全部成功写入、要么全部失败，不出现中间状态或重复写入。事务机制尤其适合于 “精确一次（Exactly-Once）” 的处理语义（EOS, Exactly Once Semantics）。

🧠 Kafka 中为什么需要事务？

在实际业务中，可能有这样的场景：

一个消费者从 Topic A 读取一条消息，然后处理它，并将处理结果写入 Topic B —— 我们希望这个“读取 + 写入”是一个整体，要么都成功，要么都失败，否则可能造成重复消费或数据不一致。

普通情况下 Kafka 只能做到：

最多一次（At most once）：消息可能丢；
至少一次（At least once）：消息可能重复；
不能保证精确一次，除非业务层做幂等控制。

因此 Kafka 引入了事务机制来支持真正的 Exactly Once Semantics（EOS）。

✅ Kafka 事务的核心概念

概念	说明
Transactional Producer	开启事务功能的生产者，可以保证一组写入的原子性。
Transactional ID	每个事务性生产者的唯一标识，用于区分和恢复未完成的事务。
事务协调器（Transaction Coordinator）	Kafka 集群中的一个 Broker 组件，负责管理事务的状态、提交与回滚。
Producer ID（PID）	Kafka 为每个事务性生产者分配的唯一 ID，用于实现幂等性和事务追踪。

✅ Kafka 事务的使用流程（简化）

初始化事务生产者（开启事务功能）：

Properties props = new Properties();
props.put("transactional.id", "txn-001");
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions();

开始事务：
```
producer.beginTransaction();
```

执行写入操作（可以写入多个 Topic、多个 Partition）：

producer.send(new ProducerRecord<>("topicA", "key1", "msg1"));
producer.send(new ProducerRecord<>("topicB", "key2", "msg2"));

提交事务（成功）或 中止事务（失败）：

producer.commitTransaction(); // 或者 producer.abortTransaction();

✅ Kafka 事务的特点与保障

1. 原子性

一次事务中的多条消息，要么全部写入成功，要么全部失败并回滚。
对消费者来说，要么能消费到完整事务内的消息，要么一条都看不到。

2. 幂等性（Idempotence）

自动启用，配合事务使用时，可以避免消息重复写入，即使重试也不会写入重复数据。

3. 隔离性

Kafka 使用 读已提交（read_committed） 和 读未提交（read_uncommitted） 的消费模式控制事务可见性。
默认：消费者只能读取已提交的事务消息，未提交或中止的事务消息不会暴露给消费者。

✅ Kafka 事务与消费者的协作（消费 + 生产）

配合 enable.auto.commit=false 和 read_committed，可以实现精确一次语义：

producer.beginTransaction();
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

for (ConsumerRecord<String, String> record : records) {
    // 处理数据，并写入结果
    producer.send(new ProducerRecord<>("output-topic", process(record)));
}

// 手动提交 offset，作为事务的一部分
Map<TopicPartition, OffsetAndMetadata> offsets = consumerOffsets(records);
producer.sendOffsetsToTransaction(offsets, consumerGroupId);

producer.commitTransaction();

通过把 消费 Offset 的提交 和 生产消息的提交 绑定到同一个事务中，Kafka 实现了端到端的 Exactly-Once 保证。

✅ Kafka 事务机制本质

Kafka 事务的回滚机制，并不是自动触发的，开发者必须在代码逻辑中显式地判断是否出错，然后手动调用：

producer.abortTransaction();

如果开发者只调用了 producer.commitTransaction()，而没有判断出错，也没有手动调用 abortTransaction()，那么出问题时 Kafka 不会自动回滚！需要开发者自己判断、自己调用！

✅ 正确的事务控制流程应该是这样的：

try {
    producer.beginTransaction();

    // 消费 + 处理 + 发送
    producer.send(...);
    producer.send(...);

    // 提交 offset
    producer.sendOffsetsToTransaction(offsets, consumerGroupId);

    // 提交事务
    producer.commitTransaction();   // ✅ 成功，整个事务写入生效！

} catch (Exception e) {
    // 出现任何异常，都应该回滚
    producer.abortTransaction();   // ❗回滚事务，所有写入 + offset 统统丢弃
    e.printStackTrace();
}

🚨 如果直接调用 `commitTransaction()` 会怎样？

如果前面的 send() 或 sendOffsetsToTransaction() 抛了异常，但没 try-catch 捕获，程序继续调用 commitTransaction()，结果是：

Kafka 会检测到前面出了问题，会抛出 ProducerFencedException、IllegalStateException 等；
此时事务已经处于非法状态；
commitTransaction() 会失败，Kafka 不会自动回滚！
如果你不手动调用 abortTransaction()，这个事务就会卡在中间状态，不生效，也没回滚。

✅ 做法：必须写 try-catch 包住整个事务过程，出错就 `abortTransaction()`！

这就是标准的 事务控制模式（跟数据库事务的 try-catch 是一样的思路）：

⚠️ 常见会触发事务失败的场景：

场景	会发生什么
网络波动、Broker 写入超时	`send()` 抛异常
offset 提交失败	`sendOffsetsToTransaction()` 抛异常
重复使用 `transactional.id` 被踢出	抛 `ProducerFencedException`
使用了错误的调用顺序	抛 `IllegalStateException`
`commitTransaction()` 时事务非法	提交失败，不会自动回滚

✅ 补充一句

Kafka 的事务机制本质上是「声明式事务」，但实现方式是「编程式事务」，不像数据库事务那样自动提交或自动回滚 —— 所以你写代码的时候，一定要有清晰的事务控制逻辑。

✅ 总结一下你该怎么做

✅ 成功就 commitTransaction()
❌ 出错就 abortTransaction()
🧠 判断错没错，靠自己的业务代码来 try-catch 控制

什么是事务非法？

commitTransaction() 时事务非法，这句话是什么意思？“事务非法”到底是个啥意思？

在 Kafka 中，一个事务是有“状态”的，它不是你想提交就能提交的。只有当事务状态是“合法/活跃”的时候，才能 commitTransaction()，否则就会抛异常。

所以，“事务非法” = 事务已经处于异常、失效、终止状态，不能提交。

📊 Kafka 中事务的几种状态（简化理解）

事务状态	描述
`Initialized`	`initTransactions()` 调用后，初始化完成
`Started`	调用 `beginTransaction()` 后，事务已开始
`InFlight`	事务进行中，已发送消息，或 offset
✅ `ReadyToCommit`	一切正常，可以提交
❌ `Invalid`	出现异常、被踢出、操作错误 → 事务非法
`Committed`	已提交成功
`Aborted`	已主动回滚

🚨 什么情况下事务会变成“非法状态”？

以下几种情况会让事务“非法化”，从而你调用 commitTransaction() 时直接失败：

1. ❌ 你调用顺序错了

比如你根本没有调用 beginTransaction()，就直接调用 send() 或 commitTransaction()：

producer.initTransactions();
// producer.beginTransaction();  // ❌ 忘了这行！

producer.send(...);             // ❌ 错误用法
producer.commitTransaction();   // ❌ 会抛 IllegalStateException

这时候 Kafka 会认为你“乱搞”，把事务标为非法状态。

2. ❌ 事务内某个操作失败（比如 send 抛异常）

producer.beginTransaction();

try {
    producer.send(...);  // ⚠️ 如果这里失败了，比如网络问题
    producer.commitTransaction(); // ❌ 事务状态非法，提交失败
} catch (Exception e) {
    producer.abortTransaction();  // ✅ 你得主动回滚！
}

3. ❌ 你被 Kafka 判定为“被踢出事务”

Kafka 是通过 transactional.id 标识一个事务性的 producer 的，一个 transactional.id 只能在一个 producer 实例中使用。
如果你重复使用了这个 ID（比如程序重启未清理），Kafka 会抛：

org.apache.kafka.common.errors.ProducerFencedException

这时，Kafka 会把你当前的事务标记为非法，你必须关闭 producer 实例，否则不能提交也不能继续。

4. ❌ offset 提交失败了

producer.sendOffsetsToTransaction(...);  // 如果这里异常，事务就“坏了”
producer.commitTransaction();            // ❌ 再提交，事务非法

🧠 为什么 Kafka 要这么严格？

因为 Kafka 的事务要保证：

要么全写入、全提交，要么一个字节都不留下。

所以一旦你有步骤失败，它就会保护性地禁止你再提交，以免产生脏数据（比如你写了一半就崩了，还提交 offset，那就“假成功”了）。

✅ 那应该怎么做？

使用事务时，标准模板写法如下：

producer.initTransactions();

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
    try {
        producer.beginTransaction();

        for (ConsumerRecord<String, String> record : records) {
            // 处理 + 转发
            producer.send(new ProducerRecord<>("output-topic", record.key(), process(record.value())));
        }

        // 把消费 offset 提交到事务中
        Map<TopicPartition, OffsetAndMetadata> offsets = ...;
        producer.sendOffsetsToTransaction(offsets, consumerGroupId);

        producer.commitTransaction();  // ✅ 事务提交

    } catch (Exception e) {
        producer.abortTransaction();  // ❗出现问题，事务回滚
    }
}