Kafka 事务与 Exactly-Once 语义：从幂等生产者到端到端精确一次

① 用户发起转账 → ② 扣款事件写入 Kafka → ③ 消费者把扣款写入账户库
                     （发送时网络抖动，重试成功）
                     ④ 同一事件又发一次 → ⑤ 账户库重复扣款

<Producer ID（PID）, 分区, 序列号>
  → broker 按（PID, 分区）记录最大连续序列号
  → 收到重复序列号 → 直接返回成功（已存过）
  → 收到乱序/缺失 → 报 OutOfOrderSequenceException

Properties props = new Properties();
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);   // 开启幂等
props.put(ProducerConfig.ACKS_CONFIG, "all");                 // 幂等要求 acks=all
props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);
props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);

幂等生产者：单个生产者实例，重试不重复
    ↓ 升级
事务生产者：同一 producer，多批次原子提交 + 故障恢复（fencing）

生产者 → FindCoordinator 定位协调器 → 协调器在 __transaction_state 记录事务状态
状态机：Empty → Ongoing → PrepareCommit → CompleteCommit
                └→ PrepareAbort → CompleteAbort

① 生产者发起事务（initTransactions）
② 发送消息到各分区（此时消息「未提交」，其他事务消费者不可见）
③ 发送 offset 提交到 __consumer_offsets（可选项，绑定消费位移）
④ 生产者调用 commitTransaction → 协调器协调所有参与分区写入事务标记
⑤ 协调器向 __transaction_state 写 PrepareCommit → 完成后写 CompleteCommit

// 事务生产者：转账扣款 + 通知 两条消息原子提交
Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "payment-producer-1"); // 必须全局唯一
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions();           // 与协调器注册事务 ID

try {
    producer.beginTransaction();
    // 业务事件 + 通知事件原子写入
    producer.send(new ProducerRecord<>("payments", "txn-001", "{\"debit\":100}"));
    producer.send(new ProducerRecord<>("notifications", "txn-001", "{\"msg\":\"扣款成功\"}"));
    producer.commitTransaction();
} catch (ProducerFencedException e) {
    // 被新实例接管（fencing），必须放弃
    producer.close();
} catch (Exception e) {
    producer.abortTransaction();       // 回滚
}

每个事务 ID 有递增的 epoch
协调器收到更老 epoch 的生产者请求 → 抛 ProducerFencedException 拒绝
旧实例被「隔离」，无法再写 → 保证同一事务 ID 只有一个活跃生产者

消费 Topic A（读取输入）
    ↓ 处理（去重、聚合、转换）
写入 Topic B（产出输出）
    ↓
消费 Topic C（下一个处理阶段）

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "order-enricher");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG,
          StreamsConfig.EXACTLY_ONCE_V2);   // 精确一次（V2 起效率更高）

// 消费者：绑定到事务（提交位移参与事务）
props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
consumer.subscribe(List.of("orders"));

// 生产者：开启事务
producer.initTransactions();

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
    if (records.isEmpty()) continue;

    producer.beginTransaction();
    Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();

    for (ConsumerRecord<String, String> r : records) {
        // 业务处理
        String enriched = enrich(r.value());
        // 产出到下游 Topic
        producer.send(new ProducerRecord<>("enriched-orders", r.key(), enriched));
        offsets.put(new TopicPartition(r.topic(), r.partition()),
                    new OffsetAndMetadata(r.offset() + 1));
    }
    // 关键：把消费位移一起提交到事务里
    producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());
    producer.commitTransaction();
}

props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");

订单服务（本地事务）：
  ① INSERT INTO orders (...)              -- 业务数据
  ② INSERT INTO outbox (id, topic, key, payload)   -- 待发事件
  两行写同一事务 → 要么都成功，要么都失败
然后：
  ③ Debezium/自研 worker 读 outbox → 发 Kafka → 标记已发送

CREATE TABLE outbox (
  id          BIGINT PRIMARY KEY AUTO_INCREMENT,
  aggregate_id VARCHAR(64)  NOT NULL,      -- 业务聚合根 ID
  topic       VARCHAR(255) NOT NULL,       -- 目标 Topic
  key         VARCHAR(255),                -- 分区键
  payload     JSON         NOT NULL,       -- 事件体
  status      TINYINT      DEFAULT 0,      -- 0=待发 1=已发
  created_at  TIMESTAMP    DEFAULT CURRENT_TIMESTAMP
) ENGINE=InnoDB;