#!/usr/bin/env bash
# =============================================================================
# dump_metadata_log.sh
#   一键解析 KRaft 集群的 __cluster_metadata 日志，把里头的元数据事件
#   （TopicRecord、PartitionRecord、PartitionChangeRecord、RegisterBrokerRecord …）
#   人类可读地打印出来。
#
# 这是排查「Topic 创建后 Broker 看不到」「ACL 不生效」「Quorum 元数据撕裂」
# 类问题的「核武器」工具。
#
# 用法（在 docker-compose 起来的 KRaft 集群里）：
#   ./dump_metadata_log.sh                        # 默认 dump kafka1 的元数据
#   ./dump_metadata_log.sh kafka2                 # dump kafka2
#   ./dump_metadata_log.sh kafka1 shell           # 进入交互式 metadata-shell
#   ./dump_metadata_log.sh kafka1 latest          # 只看最新 segment 的可读输出
#
# 依赖：
#   * Kafka 3.3+（kafka-metadata-shell.sh 和 kafka-dump-log.sh 都自带）
#   * docker compose（容器名 kafka1/kafka2/kafka3）
#
# 关键命令解释：
#   1) kafka-metadata-shell.sh    交互式查询元数据快照（支持 ls / cat / find）
#   2) kafka-dump-log.sh          按 record 一条条打印日志原文
#   3) kafka-metadata-quorum.sh   看 Quorum 状态 / 各 Voter 复制位
# =============================================================================

set -euo pipefail

CONTAINER=${1:-kafka1}
MODE=${2:-dump}
META_DIR="/var/lib/kafka/data/__cluster_metadata-0"

usage() {
  cat <<EOF
用法：$0 [container] [mode]
  container : kafka1 / kafka2 / kafka3 （默认 kafka1）
  mode      : dump      → 全量打印所有 segment（默认）
              latest    → 只打印当前最新 segment
              shell     → 进入交互式 kafka-metadata-shell.sh
              quorum    → 显示 Quorum 状态（不解析日志）
              snapshot  → 列出已生成的快照文件
EOF
  exit 1
}

case "$MODE" in
  dump|latest|shell|quorum|snapshot) ;;
  -h|--help) usage ;;
  *) echo "未知 mode: $MODE"; usage ;;
esac

echo ">>> 容器：$CONTAINER   模式：$MODE"
echo

# 检查容器是否在跑
if ! docker compose ps --format json 2>/dev/null | grep -q "$CONTAINER"; then
  if ! docker ps --format '{{.Names}}' | grep -q "^${CONTAINER}$"; then
    echo "❌ 容器 $CONTAINER 不在运行，先 docker compose up -d"
    exit 2
  fi
fi

run_in() {
  docker exec -i "$CONTAINER" bash -c "$1"
}

case "$MODE" in
  quorum)
    echo "=== Quorum Status ==="
    run_in "/opt/kafka/bin/kafka-metadata-quorum.sh \
              --bootstrap-server localhost:19092 describe --status"
    echo
    echo "=== Quorum Replication ==="
    run_in "/opt/kafka/bin/kafka-metadata-quorum.sh \
              --bootstrap-server localhost:19092 describe --replication"
    ;;

  snapshot)
    echo "=== 元数据目录与 snapshot 文件 ==="
    run_in "ls -lh $META_DIR/ | head -30"
    ;;

  shell)
    echo "=== 启动交互式 kafka-metadata-shell.sh ==="
    echo "（进去后可以执行 ls /  /  cat /topics/<name>/<partition>/data 等）"
    echo
    # 用最新一个 .log 段；如果有 .checkpoint（snapshot）也可以用
    LATEST_LOG=$(run_in "ls $META_DIR/*.log 2>/dev/null | sort | tail -1")
    if [[ -z "$LATEST_LOG" ]]; then
      echo "❌ 没找到 $META_DIR/*.log"; exit 3
    fi
    echo "使用文件: $LATEST_LOG"
    docker exec -it "$CONTAINER" /opt/kafka/bin/kafka-metadata-shell.sh \
      --snapshot "$LATEST_LOG"
    ;;

  latest|dump)
    if [[ "$MODE" == "latest" ]]; then
      FILES=$(run_in "ls $META_DIR/*.log 2>/dev/null | sort | tail -1")
    else
      FILES=$(run_in "ls $META_DIR/*.log 2>/dev/null | sort")
    fi
    if [[ -z "$FILES" ]]; then
      echo "❌ 在容器 $CONTAINER 的 $META_DIR 下没找到 .log 文件"
      echo "   说明该节点不是 Controller Quorum 成员（普通 Broker 不存元数据日志副本）"
      echo "   Controller Quorum 一般是独立节点；如果你用了 combined 模式，"
      echo "   则 broker,controller 节点会有这个目录。"
      exit 4
    fi

    for f in $FILES; do
      echo "================================================================"
      echo "  解析: $f"
      echo "================================================================"
      run_in "/opt/kafka/bin/kafka-dump-log.sh \
                --cluster-metadata-decoder \
                --files $f \
                --print-data-log" \
        | head -300
      echo
    done

    cat <<EOF

读法说明：
  * baseOffset / lastOffset：本批次（RecordBatch）的 offset 区间；
  * payload 反序列化为 KRaft 定义的元数据事件类型；常见的有：
      RegisterBrokerRecord     —— Broker 加入集群
      UnregisterBrokerRecord   —— Broker 注销
      TopicRecord              —— 创建 Topic
      PartitionRecord          —— 创建分区时的初始状态
      PartitionChangeRecord    —— Leader/ISR/Replicas 变更
      ConfigRecord             —— 动态配置修改
      AccessControlEntryRecord —— ACL 变更
  * --cluster-metadata-decoder 是关键选项；不加它解出来都是十六进制乱码。

下一步：
  * 想交互查询：./dump_metadata_log.sh $CONTAINER shell
  * 想看 Quorum 健康：./dump_metadata_log.sh $CONTAINER quorum
EOF
    ;;
esac