#!/usr/bin/env python3
"""
read_committed_consumer.py
==========================
对比 read_uncommitted vs read_committed 的实际效果。

用法：
  pip install confluent-kafka

  # 终端 A：跑 transactional_producer.py commit/abort 演示
  # 终端 B：用 uncommitted 模式看，会立刻看到所有事务消息（包括 abort 掉的）
  python read_committed_consumer.py learn.13.tx_orders uncommitted

  # 终端 C：用 committed 模式看，只能看到 commit 后的，abort 的永远看不到
  python read_committed_consumer.py learn.13.tx_orders committed
"""

import os
import sys

from confluent_kafka import Consumer

BOOTSTRAP = os.environ.get("KAFKA_BOOTSTRAP", "127.0.0.1:9092")


def main():
    if len(sys.argv) < 3:
        print(__doc__); sys.exit(0)
    topic = sys.argv[1]
    level = sys.argv[2]   # uncommitted | committed

    if level not in ("uncommitted", "committed"):
        print("level must be 'uncommitted' or 'committed'"); sys.exit(1)

    iso = "read_uncommitted" if level == "uncommitted" else "read_committed"

    c = Consumer({
        "bootstrap.servers": BOOTSTRAP,
        "group.id": f"reader-{level}-{os.getpid()}",
        "auto.offset.reset": "earliest",
        "enable.auto.commit": False,
        "isolation.level": iso,
    })
    c.subscribe([topic])

    print(f"== isolation.level={iso} → topic={topic} ==")
    print("（事务正在进行中的消息：")
    print("  read_uncommitted 会立刻看到（包括 abort 后还展示）；")
    print("  read_committed 只在 commit 后才看到，abort 的永远看不到）")
    print()

    n = 0
    try:
        while True:
            msg = c.poll(1.0)
            if msg is None: continue
            if msg.error():
                print(f"err: {msg.error()}"); continue
            n += 1
            print(f"  [{n:4d}] p{msg.partition()}@{msg.offset()} key={msg.key()!r} "
                  f"value={msg.value()[:80]!r}")
    except KeyboardInterrupt:
        print(f"-- received {n} msgs --")
    finally:
        c.close()


if __name__ == "__main__":
    main()