""" 第 2 章 · Cluster Probe —— 把整个 Kafka 集群拓扑「一键打印」 用法： python 02_architecture/code/cluster_probe.py python 02_architecture/code/cluster_probe.py --bootstrap localhost:9092 python 02_architecture/code/cluster_probe.py --bootstrap localhost:9092 --include-internal 它会做四件事： 1. 连上 Kafka 集群，拿 Cluster ID + Active Controller ID。 2. 列出所有 Broker（id / host / port / rack）。 3. 列出所有 Topic（默认隐藏内部 Topic，可用 --include-internal 显示 __consumer_offsets / __transaction_state / __cluster_metadata 等）。 4. 对每个 Topic 打印所有 Partition：partition_id / leader / replicas / isr。 输出全部用 `tabulate` 渲染成 ASCII 表格，方便复制贴到工单 / 文档里。 连不上 Broker 时会给出友好提示（端口、防火墙、bootstrap 地址等常见排查方向）。 依赖： pip install -r requirements.txt # 至少需要 confluent-kafka>=2.5 + tabulate>=0.9 预期输出（节选，3 broker × KRaft 集群，已建若干 demo Topic）： ┌────────────────────────── 集群概览 ──────────────────────────┐ │ Cluster ID : MkU3OEVBNTcwNTJENDM2Qk │ │ Bootstrap : localhost:9092 │ │ Total Brokers : 3 │ │ Active Controller : Broker 3 (KRaft Active) │ │ Total Topics : 4 (含内部 1) │ │ Total Partitions : 56 │ └──────────────────────────────────────────────────────────────┘ ── Brokers ─────────────────────────────────── ID Host Port Rack ---- ----------- ------ ------ 1 kafka1 19092 rack-a 2 kafka2 19094 rack-b 3 kafka3 19096 rack-c ── Topic: learn.02.cluster (RF=3, Partitions=6, Internal=No) ── Partition Leader Replicas ISR OOS-Replicas ----------- ------ -------- ------ -------------- 0 1 1,2,3 1,2,3 () 1 2 2,3,1 2,3,1 () ... """ from __future__ import annotations import argparse import sys import time from typing import List, Tuple try: from confluent_kafka.admin import AdminClient, ConfigResource from confluent_kafka import KafkaException except ImportError as e: print(f"[FATAL] 缺少依赖：{e}\n请先安装：pip install -r requirements.txt") sys.exit(1) try: from tabulate import tabulate except ImportError: print("[FATAL] 缺少依赖 tabulate，请先 pip install tabulate>=0.9") sys.exit(1) INTERNAL_TOPIC_PREFIXES = ("__",) # __consumer_offsets / __transaction_state ... def parse_args() -> argparse.Namespace: p = argparse.ArgumentParser( description="探针：打印 Kafka 集群拓扑（Broker + Topic + Partition）。" ) p.add_argument( "--bootstrap", default="localhost:9092", help="Kafka bootstrap.servers (默认 localhost:9092)", ) p.add_argument( "--include-internal", action="store_true", help="同时显示内部 Topic（__consumer_offsets 等）", ) p.add_argument( "--timeout", type=float, default=10.0, help="集群元数据获取超时秒数 (默认 10s)", ) return p.parse_args() def is_internal(topic: str) -> bool: return any(topic.startswith(pfx) for pfx in INTERNAL_TOPIC_PREFIXES) def fetch_metadata(admin: AdminClient, timeout: float): """连不上时，把常见原因贴出来，避免新手陷入「卡住不动」迷局。""" try: md = admin.list_topics(timeout=timeout) except KafkaException as e: print(f"\n[ERROR] 无法连接 Kafka 集群：{e}") print("\n常见排查方向：") print(" 1. Broker 是否真的在跑？ -> docker ps | grep kafka") print(" 或在 broker 节点上： netstat -lntp | grep 9092") print(" 2. bootstrap.servers 写对了吗？默认是 localhost:9092；docker compose 时可能是 19092 / 19094 / 19096。") print(" 3. 防火墙 / 安全组放通 9092 / 9093 端口。") print(" 4. KRaft 集群刚启动需要 30~60s 才进入 RUNNING 状态，可稍等再试。") print(" 5. listeners / advertised.listeners 里写的 hostname 客户端能解析吗？") sys.exit(2) except Exception as e: print(f"\n[ERROR] 元数据请求失败：{e}") sys.exit(2) return md def render_overview(md, bootstrap: str, include_internal: bool) -> None: cluster_id = md.cluster_id or "(unknown, KRaft 早期版本可能不返回)" controller = md.controller_id # 这是当前 metadata response 的 controller id total_topics = len(md.topics) visible_topics = [t for t in md.topics if include_internal or not is_internal(t)] total_parts = sum(len(t.partitions) for t in md.topics.values()) lines = [ ("Cluster ID", cluster_id), ("Bootstrap", bootstrap), ("Total Brokers", len(md.brokers)), ("Active Controller", f"Broker {controller}" if controller >= 0 else "(unknown)"), ( "Total Topics", f"{total_topics} (显示 {len(visible_topics)} | 隐藏内部 {total_topics - len(visible_topics)})", ), ("Total Partitions", total_parts), ] print() print("┌" + "─" * 26 + " 集 群 概 览 " + "─" * 26 + "┐") for k, v in lines: line = f"│ {k:<19}: {v}" print(line + " " * max(0, 65 - len(line)) + "│") print("└" + "─" * 65 + "┘") def render_brokers(md) -> None: print("\n── Brokers ──────────────────────────────────────") rows = [] for bid, b in sorted(md.brokers.items()): rack = getattr(b, "rack", None) or "-" rows.append([bid, b.host, b.port, rack]) print(tabulate( rows, headers=["ID", "Host", "Port", "Rack"], tablefmt="simple", )) def render_topics(md, include_internal: bool) -> None: topic_names = sorted(md.topics.keys()) if not include_internal: topic_names = [t for t in topic_names if not is_internal(t)] if not topic_names: print("\n[INFO] 当前集群没有任何用户 Topic。试试：") print(" kafka-topics.sh --bootstrap-server --create --topic learn.02.demo --partitions 6 --replication-factor 3") return for tname in topic_names: t = md.topics[tname] if t.error is not None: print(f"\n── Topic: {tname} ── ⚠️ 元数据错误：{t.error}") continue partitions = sorted(t.partitions.values(), key=lambda x: x.id) rf = len(partitions[0].replicas) if partitions else 0 rows = [] for p in partitions: replicas = sorted(p.replicas) isr = sorted(p.isrs) oos = [r for r in replicas if r not in isr] # Out-of-Sync Replicas rows.append([ p.id, p.leader if p.leader >= 0 else "(none)", ",".join(map(str, replicas)), ",".join(map(str, isr)) if isr else "(empty)", ",".join(map(str, oos)) if oos else "()", ]) flag_internal = "Yes" if is_internal(tname) else "No" print(f"\n── Topic: {tname} (RF={rf}, Partitions={len(partitions)}, Internal={flag_internal}) ──") print(tabulate( rows, headers=["Partition", "Leader", "Replicas", "ISR", "OOS-Replicas"], tablefmt="simple", )) unhealthy = [r for r in rows if r[4] != "()"] if unhealthy: print(f" ⚠️ 有 {len(unhealthy)} 个分区存在 Out-of-Sync 副本，建议进一步排查（磁盘 / 网络 / GC）。") no_leader = [r for r in rows if r[1] == "(none)"] if no_leader: print(f" ❌ 有 {len(no_leader)} 个分区当前无 Leader，分区不可读写！") def main() -> int: args = parse_args() print(f"[INFO] 正在连接 Kafka：{args.bootstrap} ...") admin = AdminClient({ "bootstrap.servers": args.bootstrap, "client.id": "cluster-probe", "socket.timeout.ms": int(args.timeout * 1000), }) t0 = time.time() md = fetch_metadata(admin, args.timeout) elapsed = (time.time() - t0) * 1000 print(f"[INFO] 元数据获取成功，耗时 {elapsed:.1f} ms。\n") render_overview(md, args.bootstrap, args.include_internal) render_brokers(md) render_topics(md, args.include_internal) print() print("提示：") print(" • 加 --include-internal 可看 __consumer_offsets / __transaction_state（KRaft 模式下 __cluster_metadata 不在此处显示，需用 kafka-metadata-quorum.sh）。") print(" • 上面的输出 ≈ 多次 kafka-topics.sh --describe + kafka-broker-api-versions.sh 的合并版本。") print(" • 把这段输出贴到工单 / 故障复盘文档里，就是一份合格的『集群快照』。\n") return 0 if __name__ == "__main__": try: sys.exit(main()) except KeyboardInterrupt: print("\n[INFO] 用户取消。") sys.exit(130)