Kafka Transaction：多 Partition 原子写入

验证：一次提交的事务，两个不同 Topic 的消息会同时对 read_committed 消费者可见

命令 · java Producer<String,String> p = new KafkaProducer<>(props); // transactional.id=txn-demo-committer-1 p.initTransactions(); p.beginTransaction(); p.send(new ProducerRecord<>("txn-a", "committed-msg-in-A")); p.send(new ProducerRecord<>("txn-b", "committed-msg-in-B")); p.commitTransaction();

topic A 用 isolation.level=read_committed 读取 -> [committed-msg-in-A]
topic B 用 isolation.level=read_committed 读取 -> [committed-msg-in-B]

一个事务型 Producer 在beginTransaction()和commitTransaction()之间，往两个完全不同的 Topic（txn-a、txn-b）各发了一条消息。提交之后，两个 Topic 上的 read_committed 消费者都能读到各自那条消息——即使它们是两个独立的 Topic，分别由不同的 Partition Leader 处理，Kafka 事务做到的是"这一组写入要么都对外可见，要么都不可见"，不是靠某种分布式锁挡住其他人，而是靠一个统一的事务 ID 和协调者角色，把分散在不同 Partition 上的写入标记成同一个事务。

验证：中止的事务，read_committed 永远看不到，但真正落到日志里的前提是消息已经发出去了

命令 · java Producer<String,String> p2 = new KafkaProducer<>(props); // transactional.id=txn-demo-aborter-1 p2.beginTransaction(); p2.send(new ProducerRecord<>("txn-a", "ABORTED-msg-in-A")); p2.send(new ProducerRecord<>("txn-b", "ABORTED-msg-in-B")); p2.abortTransaction(); // 第一次运行：中间没有调用 flush()

$ kafka-dump-log.sh --files .../txn-a-0/00000000000000000000.log --print-data-log
baseOffset: 0  payload: committed-msg-in-A
baseOffset: 1  isControl: true  endTxnMarker: COMMIT
baseOffset: 2  payload: committed-msg-2-in-A
baseOffset: 3  isControl: true  endTxnMarker: COMMIT
<- ABORTED-msg-in-A 完全没有出现，连一个 ABORT 标记都没有

第一次运行时，我以为中止的事务会像提交的事务一样，先把消息写进日志、再补一条 ABORT 控制记录——用 kafka-dump-log.sh 把日志文件整段导出来看，结果完全没有 ABORTED-msg-in-A 这条消息，连一个 ABORT 标记都不存在。原因是 send() 只是把消息放进客户端本地的发送缓冲区，真正发往 Broker 是异步的；abortTransaction() 调用得足够快时，这批消息可能还压根没有发出去，客户端会直接在本地把它们标记失败、悄悄丢弃，Broker 那边什么都不知道，自然也不需要写 ABORT 标记。

命令 · java p2.send(new ProducerRecord<>("txn-a", "ABORTED-msg-in-A")); p2.send(new ProducerRecord<>("txn-b", "ABORTED-msg-in-B")); p2.flush(); // 第二次运行：中止前先强制把消息发到 Broker p2.abortTransaction();

baseOffset: 0  payload: committed-msg-in-A
baseOffset: 1  isControl: true  endTxnMarker: COMMIT
baseOffset: 2  payload: ABORTED-msg-in-A          <- 这次写进了日志
baseOffset: 3  isControl: true  endTxnMarker: ABORT
baseOffset: 4  payload: committed-msg-2-in-A
baseOffset: 5  isControl: true  endTxnMarker: COMMIT

isolation.level=read_committed 读取 topic A -> [committed-msg-in-A, committed-msg-2-in-A]
isolation.level=read_uncommitted 读取 topic A -> [committed-msg-in-A, ABORTED-msg-in-A, committed-msg-2-in-A]

加一次flush()强制消息发到 Broker 之后，日志里才出现了 ABORTED-msg-in-A 这条消息和紧随其后的 ABORT 控制记录。这时候用 read_committed 读，看到的是 [committed-msg-in-A, committed-msg-2-in-A]，中止的那条被跳过；换成 read_uncommitted，三条消息（含被中止的那条）按日志里的物理顺序全部读得到。read_uncommitted 不看任何事务状态，就是把日志从头到尾原样吐出来，只是控制记录本身不会被当成一条消息交给应用层。

验证：一个还没提交的事务，会卡住 read_committed 消费者，连别人后写的普通消息也读不到

命令 · java pa.beginTransaction(); pa.send(new ProducerRecord<>("txn-lso-demo", "from-A-still-open")).get(); // 这里先不 commit，事务保持打开 pb.send(new ProducerRecord<>("txn-lso-demo", "from-B-plain-after")).get(); // 另一个完全独立、非事务的 Producer

A 的事务仍然打开着，read_committed 消费者轮询 3 秒 -> 读到 []
（B 写的 from-B-plain-after 明明是一条普通消息，此时却完全读不到）

... 调用 pa.commitTransaction() ...

A 提交事务之后，read_committed 消费者再次读取 -> [from-A-still-open, from-B-plain-after]
（两条消息同时放出来，顺序和写入顺序一致）

A 开启事务、发送一条消息但还没提交时，另一个完全独立、不带事务的 Producer B 紧接着往同一个 Partition 写了一条普通消息。这时候用 read_committed 去读，B 那条本该"正常"的消息也读不到——因为 read_committed 消费者能读到的位置由一个叫 Last Stable Offset（LSO）的水位线决定，只要 A 的事务还没有 commit 或 abort，LSO 就卡在 A 那条消息之前，后面所有位置（不管是谁写的）都读不到。等 A 提交之后，LSO 才会跳过 A 的事务往前推进，A 和 B 的两条消息才一起被放出来，顺序和它们被写入的顺序一致。这意味着一个开着不提交的事务，会拖慢同一个 Partition 上所有 read_committed 消费者的进度，不只是影响这个事务自己的数据。

分步演示：一条被中止的消息，是怎么被两种隔离级别区别对待的

每一步都是运行上面那段实验代码、读取日志文件得到的结果。

结论

Kafka 事务靠一个统一的事务 ID 把分散在不同 Topic/Partition 上的写入捆成一组：提交时，这些写入对 read_committed 消费者同时可见；中止时，永久不可见，即使消息本身已经落到了日志里（这次撞见的细节是：中止前如果消息还没真正发出去，客户端会直接本地丢弃，日志里连 ABORT 标记都不会有）。read_uncommitted 完全不看事务状态，日志里物理存在什么就返回什么。一个长时间不提交也不中止的事务，会卡住同一个 Partition 上所有 read_committed 消费者的进度，不管卡住的这条消息和后面的消息是不是同一个 Producer 写的。