#!/usr/bin/env python3 """ kill_leader_demo.py — 自动化「停 Broker → 观察 Leader 切换 → 起回来」演示 它会做这些事： 1) 用 AdminClient 创建一个 3 副本 Topic（如果不存在）； 2) 起一个后台 Producer，持续以 acks=all 写入； 3) 拉取分区 P0 的 Leader Broker； 4) 用 docker compose stop kafkaX 把 Leader 那台 Broker 干掉； 5) 实时打印 ISR / Leader 变化（直到 Leader 切换完成）； 6) sleep 几秒再 docker compose start 起回来； 7) 等 ISR 重新扩张回去，结束。 前置： * 已经用 `docker compose up -d` 起好了 3 broker 集群； * 容器名为 kafka1 / kafka2 / kafka3，对外端口 9092 / 9094 / 9096； * 当前用户能执行 `docker compose`（在 docker-compose.yml 同目录）。 用法： python kill_leader_demo.py python kill_leader_demo.py --topic learn.09.killdemo --partition 0 \ --compose-dir /data/workspace/learnNote/kafka ⚠️ 仅用于学习环境，不要在生产集群运行！ """ from __future__ import annotations import argparse import os import subprocess import threading import time from datetime import datetime from confluent_kafka import Producer from confluent_kafka.admin import AdminClient, NewTopic # 容器名 → 对外端口（与 09 章 docker-compose.yml 保持一致） BROKER_ID_TO_CONTAINER = {1: "kafka1", 2: "kafka2", 3: "kafka3"} def parse_args(): p = argparse.ArgumentParser() p.add_argument("--bootstrap", default="localhost:9092,localhost:9094,localhost:9096") p.add_argument("--topic", default="learn.09.killdemo") p.add_argument("--partition", type=int, default=0) p.add_argument("--compose-dir", default="/data/workspace/learnNote/kafka", help="docker-compose.yml 所在目录") p.add_argument("--down-secs", type=int, default=15, help="Leader 容器停掉后等待多少秒再起回来") return p.parse_args() def ensure_topic(admin: AdminClient, topic: str, partitions: int = 3, rf: int = 3): md = admin.list_topics(timeout=10) if topic in md.topics and md.topics[topic].error is None: print(f"[init] topic {topic} 已存在") return print(f"[init] 创建 topic {topic} (partitions={partitions}, rf={rf})") nt = NewTopic(topic, num_partitions=partitions, replication_factor=rf, config={"min.insync.replicas": "2"}) fs = admin.create_topics([nt]) for t, f in fs.items(): try: f.result() print(f"[init] topic {t} 创建成功") except Exception as e: print(f"[init] topic {t} 创建失败: {e}") raise def get_partition_meta(admin: AdminClient, topic: str, pid: int): md = admin.list_topics(topic=topic, timeout=10) t = md.topics[topic] p = t.partitions[pid] return {"leader": p.leader, "replicas": list(p.replicas), "isr": list(p.isrs)} def docker_compose(cmd: list[str], compose_dir: str): full = ["docker", "compose"] + cmd print(f"[shell] cd {compose_dir} && {' '.join(full)}") return subprocess.run(full, cwd=compose_dir, check=False, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) class BackgroundProducer(threading.Thread): """后台持续 produce，统计成功 / 失败次数""" def __init__(self, bootstrap, topic, partition): super().__init__(daemon=True) self.topic = topic self.partition = partition self.stop_flag = threading.Event() self.ok = 0 self.fail = 0 self.last_err = None self.p = Producer({ "bootstrap.servers": bootstrap, "acks": "all", "enable.idempotence": True, "delivery.timeout.ms": 30000, "retries": 1000000, "linger.ms": 5, }) def _on_delivery(self, err, msg): if err is not None: self.fail += 1 self.last_err = str(err) else: self.ok += 1 def run(self): i = 0 while not self.stop_flag.is_set(): payload = f"msg-{i}-{datetime.now().isoformat()}".encode() try: self.p.produce(self.topic, value=payload, partition=self.partition, on_delivery=self._on_delivery) self.p.poll(0) except BufferError: self.p.poll(0.1) i += 1 time.sleep(0.05) self.p.flush(10) def watch_until(admin, topic, pid, predicate, label, timeout=60): """周期查询元数据，直到 predicate(meta) 为 True 或超时。""" print(f"[watch] 等待：{label} （超时 {timeout}s）") t0 = time.time() last = None while time.time() - t0 < timeout: meta = get_partition_meta(admin, topic, pid) snap = (meta["leader"], tuple(meta["isr"])) if snap != last: print(f"[watch] {datetime.now().strftime('%H:%M:%S')} " f"Leader={meta['leader']} ISR={meta['isr']} Replicas={meta['replicas']}") last = snap if predicate(meta): print(f"[watch] ✓ 满足条件：{label}") return meta time.sleep(1) raise TimeoutError(f"{label} 在 {timeout}s 内没有发生") def main(): args = parse_args() admin = AdminClient({"bootstrap.servers": args.bootstrap}) ensure_topic(admin, args.topic, partitions=3, rf=3) meta = get_partition_meta(admin, args.topic, args.partition) print(f"[init] {args.topic}-P{args.partition} " f"Leader={meta['leader']} ISR={meta['isr']} Replicas={meta['replicas']}") if len(meta["isr"]) < 3: print("[warn] ISR 不足 3，建议先等到 ISR=Replicas 再演示") leader_id = meta["leader"] target = BROKER_ID_TO_CONTAINER.get(leader_id) if target is None: raise SystemExit(f"未知 Leader broker_id={leader_id}，请把它加入 BROKER_ID_TO_CONTAINER") bg = BackgroundProducer(args.bootstrap, args.topic, args.partition) bg.start() print(f"[producer] 后台开始 produce（acks=all, idempotent=true）") time.sleep(2) print(f"[producer] 已成功 {bg.ok} / 失败 {bg.fail}") print(f"\n=== STEP 1: 停掉 Leader 容器 {target} ===") docker_compose(["stop", target], args.compose_dir) new_meta = watch_until(admin, args.topic, args.partition, lambda m: m["leader"] != leader_id and m["leader"] >= 0, label=f"Leader 从 {leader_id} 切换到其它 Broker", timeout=60) print(f"[result] 新 Leader = {new_meta['leader']}, ISR={new_meta['isr']}") print(f"[producer] 切换期间累计成功 {bg.ok} / 失败 {bg.fail}, 最后错误: {bg.last_err}") print(f"\n=== STEP 2: 等 {args.down_secs}s 再启动回来 ===") time.sleep(args.down_secs) docker_compose(["start", target], args.compose_dir) watch_until(admin, args.topic, args.partition, lambda m: leader_id in m["isr"], label=f"Broker {leader_id} 重新加入 ISR", timeout=120) print(f"\n=== STEP 3: 触发 Preferred Leader Election，把 Leader 还回去 ===") # 简单做法：通过 kafka-leader-election.sh in container 触发 docker_compose( ["exec", "-T", "kafka2", "/opt/kafka/bin/kafka-leader-election.sh", "--bootstrap-server", "kafka1:19092", "--election-type", "preferred", "--topic", args.topic, "--partition", str(args.partition)], args.compose_dir, ) watch_until(admin, args.topic, args.partition, lambda m: m["leader"] == leader_id, label=f"Leader 切回 {leader_id}", timeout=60) print(f"\n=== DONE ===") print(f"[summary] 全程 produce 成功 {bg.ok} 条，失败 {bg.fail} 条") print("（idempotent + acks=all + min.isr=2 + 3 副本，理论上不会丢；" "失败的那部分是 Leader 切换瞬间的瞬时报错，会被 Producer 重试覆盖）") bg.stop_flag.set() bg.join(timeout=15) if __name__ == "__main__": main()