Producer 可靠性与幂等事务

acks=0:  Producer ──发──>  (不等任何确认)          最快, 网络抖动就丢
acks=1:  Producer ──发──> Leader 写入本地日志 ──确认   Leader 挂且未同步则丢
acks=all: Producer ──发──> Leader 写入
                            └─> 等 ISR 中所有副本同步完成 ──确认   最可靠

# ISR 中至少 2 个副本确认, 否则 Producer 收到 NotEnoughReplicas 异常
min.insync.replicas=2

retries=2147483647          # 默认无限重试, 由 delivery.timeout.ms 兜底
delivery.timeout.ms=120000  # 两分钟内没成功就报失败
max.in.flight.requests.per.connection=5   # 单连接未确认请求数

乱序示意 (max.in.flight=2, 无幂等):
  发送: batch1, batch2
  batch1 失败, batch2 成功写入
  batch1 重试成功
  分区里的顺序: batch2, batch1   <- 乱了!

enable.idempotence=true   # Kafka 3.x 默认已开启

Producer(PID=88) -> partition-0
  发送 seq=0 [写入]  seq=1 [写入]  seq=1 重试 [重复, 丢弃]  seq=2 [写入]
  分区日志: seq0, seq1, seq2   -- 不重不漏不乱

组件:
  transactional.id  -- 用户指定, 重启后不变, 用来找回之前的事务状态
  事务协调器 (Transaction Coordinator) -- broker 端模块
  __transaction_state -- 内部 topic, 持久化事务状态
 
流程:
  initTransactions()  向协调器注册 transactional.id, 递增 epoch
       |                (旧 epoch 的僵尸 Producer 之后的写入会被拒绝)
  beginTransaction()
  send() x N          消息正常写入各分区, 但标记为"未提交"
  commitTransaction() 协调器两阶段提交:
       |               1. 写 PREPARE_COMMIT 到 __transaction_state
       |               2. 向各分区写入 "提交标记" (control record)
  消费端 isolation.level=read_committed 只读已提交消息

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "order-svc-tx-01"); // 实例唯一且稳定
// 事务自动隐含 enable.idempotence=true, acks=all
 
Producer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions();
 
try {
    producer.beginTransaction();
    producer.send(new ProducerRecord<>("orders", "u1", "order-created"));
    producer.send(new ProducerRecord<>("order-events", "u1", "audit-log"));
    producer.commitTransaction();          // 两条消息原子可见
} catch (ProducerFencedException e) {
    producer.close();                      // 被新实例顶替(僵尸), 直接退出
} catch (KafkaException e) {
    producer.abortTransaction();           // 回滚, 消费者永远看不到这批消息
}

from confluent_kafka import Producer, KafkaException
 
p = Producer({
    "bootstrap.servers": "localhost:9092",
    "transactional.id": "order-svc-tx-01",
})
p.init_transactions()
p.begin_transaction()
try:
    p.produce("orders", key="u1", value="order-created")
    p.produce("order-events", key="u1", value="audit-log")
    p.commit_transaction()
except KafkaException:
    p.abort_transaction()