"""
第 5 章 - 自动提交 Consumer + 漏消费演示
=================================

默认 enable.auto.commit=True 时，poll() 内部按 auto.commit.interval.ms
定期提交「上次 poll 拿到的最大 offset + 1」。

风险：
  - 漏消费：消息从 poll 出来 → 5s 到 → 自动 commit → 进程崩溃
            → 重启从 commit 之后开始 → 丢了那批没处理完的消息

本脚本演示这个陷阱：
  - 业务故意 sleep(2s) 处理慢
  - 当处理完 2 条后用 sys.exit(1) 模拟崩溃
  - 重启脚本会发现「丢了第 3、4 条」（如果 commit 已经发生）

运行：
    bash ../init.sh
    python auto_commit_consumer.py            # 第一次跑会"崩溃"
    python auto_commit_consumer.py            # 第二次跑看 lag

正确做法见 manual_commit_consumer.py。
"""

from __future__ import annotations

import os
import sys
import time

from confluent_kafka import Consumer, KafkaError

BOOTSTRAP = sys.argv[1] if len(sys.argv) > 1 else "127.0.0.1:9092"
TOPIC = "learn.05.orders"
GROUP = "ch5-auto-commit-demo"
CRASH_AFTER = int(os.environ.get("CRASH_AFTER", "2"))
SLOW_PROCESS_MS = int(os.environ.get("SLOW_PROCESS_MS", "2000"))


def main() -> None:
    consumer = Consumer(
        {
            "bootstrap.servers": BOOTSTRAP,
            "group.id": GROUP,
            "client.id": f"auto-commit-{os.getpid()}",
            "auto.offset.reset": "earliest",
            "enable.auto.commit": True,
            "auto.commit.interval.ms": 1000,  # 1 秒就 commit，方便复现
            "session.timeout.ms": 10000,
            "max.poll.interval.ms": 60000,
        }
    )
    consumer.subscribe([TOPIC])
    print(f"[auto-commit-demo] group={GROUP}, will crash after {CRASH_AFTER} msgs")

    processed = 0
    try:
        while True:
            msg = consumer.poll(1.0)
            if msg is None:
                print("  ... no message in 1s, lag may be 0")
                continue
            if msg.error():
                if msg.error().code() != KafkaError._PARTITION_EOF:
                    print(f"  err: {msg.error()}")
                continue

            print(
                f"  [{processed + 1:>3}] received P{msg.partition()}@{msg.offset()} "
                f"key={msg.key().decode() if msg.key() else None}"
            )

            time.sleep(SLOW_PROCESS_MS / 1000)
            processed += 1

            print(f"  ✓ processed (took {SLOW_PROCESS_MS}ms)")

            if processed >= CRASH_AFTER:
                print("\n💥 模拟崩溃！消息可能已被 auto-commit 但没处理完")
                print("   重启脚本看下一次能拿到多少 lag")
                # 故意不调 consumer.close()，模拟崩溃
                os._exit(1)
    finally:
        consumer.close()


if __name__ == "__main__":
    main()