消费—处理—再生产如何实现事务一致性

验证：读、写、提交offset，用一个事务真实包起来

命令 · python (confluent-kafka) p.begin_transaction() msgs = consumer.poll(...) # 真实读到2条 for m in msgs: p.produce("rpw-output", value=transform(m.value())) p.flush() p.send_offsets_to_transaction( [TopicPartition(m.topic(), m.partition(), m.offset()+1) for m in msgs], consumer.consumer_group_metadata(), ) p.abort_transaction() # 模拟：处理完了，但提交前进程真实崩溃

committed offset after abort: -1001 (should still be None/unmoved)
rpw-output content after abort: [] (should be empty, the produce was rolled back too)

-1001是confluent-kafka客户端里"这个分区从来没有提交过offset"的哨兵值。哪怕处理逻辑真实跑完了，写到输出topic的那两条消息也真实发出去了，一旦事务真实没有提交，send_offsets_to_transaction 记的offset和刚才produce的输出消息真实会一起被回滚——输出topic里查不到任何东西，就像这次处理从未发生过。

验证：重启后重新处理同样的输入，这次真实提交

命令 · python consumer2 = Consumer({...}) # 同一个group.id p2.begin_transaction() msgs2 = consumer2.poll(...)

re-read messages (should be the SAME order-0/order-1, since offset never advanced):
  ['order-0', 'order-1']

committed offset after commit: 2
rpw-output content after commit:
  ['ORDER-0-PROCESSED', 'ORDER-1-PROCESSED']
  (exactly 2 entries, no duplicates from the aborted round 1 attempt)

分步演示：读—处理—写循环真实怎么做到只生效一次

每一步都是本文上面两轮真实执行得到的原始结果。

结论

"消费—处理—再生产"这条链路真实的风险点在于：处理输入、写输出、确认已经消费到哪，是三件真实独立的事，任何一步之后崩溃都可能导致"数据丢了"或者"数据被处理了两次却看起来只处理一次"。把输出的produce和consumer的offset提交真实塞进同一个Kafka事务，这三件事就变成了真实同生共死的一个整体：要么全部生效（下游看到处理结果、offset真实往前走），要么全部不生效（下游看不到任何东西、下次重新从原来的offset开始）——没有中间状态。