"""Python client 2.12.0: application duplicates are distinct sends."""
import argparse
import json
from pathlib import Path
from confluent_kafka import Producer

def main():
    ap=argparse.ArgumentParser()
    ap.add_argument("--brokers",default="localhost:19092")
    ap.add_argument("--topic",default="handbook.events")
    ap.add_argument("--data",type=Path,default=Path(__file__).parent/"data/events.jsonl")
    ap.add_argument("--mode",choices=["basic","idempotent","transactional"],default="idempotent")
    ap.add_argument("--transactional-id",default="handbook-logical-writer-1")
    ap.add_argument("--poison",action="store_true")
    a=ap.parse_args()
    config={"bootstrap.servers":a.brokers,"acks":"all",
            "enable.idempotence":a.mode!="basic","delivery.timeout.ms":30000}
    if a.mode=="transactional": config["transactional.id"]=a.transactional_id
    producer=Producer(config)
    errors=[]; delivered=[]
    def done(error,message):
        if error: errors.append(str(error))
        else: delivered.append((message.partition(),message.offset()))
    if a.mode=="transactional":
        producer.init_transactions(30)
        producer.begin_transaction()
    try:
        for line in a.data.read_text(encoding="utf-8").splitlines():
            event=json.loads(line)
            producer.produce(a.topic,key=event["tenant_id"],value=line, on_delivery=done)
            producer.poll(0)
        if a.poison:
            producer.produce(a.topic,key="t7",value="not-json",on_delivery=done)
        pending=producer.flush(30)
        if pending or errors: raise RuntimeError({"pending":pending,"errors":errors})
        if a.mode=="transactional":producer.commit_transaction(30)
    except BaseException:
        if a.mode=="transactional":
            try: producer.abort_transaction(30)
            except Exception: pass
        raise
    print(json.dumps({"application_records_acked":len(delivered),"mode":a.mode}))
    # A fatal/ambiguous transactional error requires documented recovery/fencing;
    # this CLI stops, never retries the complete business file under a new ID.

if __name__=="__main__":main()