在 Compose Kafka 中停机备份数据目录，恢复 topic 元数据、消息和消费者位点并核对 lag

演练目标

• 在单节点 KRaft Kafka 中创建 topic，生产可核对的消息，并把消费者组提交到指定 offset，制造可观察的 lag。
• 导出 topic 描述、配置、分区 end offset 和消费者组位点，作为恢复基线。
• 停止 broker，复制完整数据目录作为离线备份；重启后继续写入，制造备份之后的 RPO 缺口。
• 删除数据卷模拟数据丢失，从离线备份恢复，验证 topic 元数据、消息内容、end offset、提交位点和 lag。
• 演练位点错误后的 dry-run 重置，理解“消息恢复”和“消费者位点恢复”是两件事。

安全边界：本 Lab 只使用本地 Compose 项目、固定名卷 lab-kafka-data 和 /tmp/lab-kafka-restore 目录，不连接任何生产 bootstrap server。不要在线上复制运行中的 log.dirs，不要格式化 KRaft 元数据目录，不要为了让副本上线而启用 unclean.leader.election.enable。生产恢复必须先停写入、确认快照一致性，并保留旧集群和旧数据目录作为回滚材料。

环境准备

准备 Docker 与 Compose v2，预留约 1 GB 内存。实验使用单节点 KRaft，同时承担 broker 和 controller；这样可以完整演练元数据与数据目录的成对恢复，但无法验证多副本 ISR。镜像标签不写死，请使用你已验证、且 CLI 与 broker 版本一致的 Kafka 镜像。

timeout 10 mkdir -p /tmp/lab-kafka-restore/backup /tmp/lab-kafka-restore/evidence
export KAFKA_IMAGE="${KAFKA_IMAGE:?请设置已验证的 Kafka 镜像}"
timeout 30 docker version --format '{{.Server.Version}}'
timeout 30 docker compose version
timeout 30 tee /tmp/lab-kafka-restore/docker-compose.yml >/dev/null <<'EOF'
name: lab-kafka-restore
services:
  kafka:
    image: ${KAFKA_IMAGE:?set KAFKA_IMAGE}
    container_name: lab-kafka-restore
    ports:
    - "127.0.0.1:19092:19092"
    environment:
      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093,EXTERNAL://:19092
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092,EXTERNAL://127.0.0.1:19092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,EXTERNAL:PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_LOG_DIRS: /var/lib/kafka/data
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false"
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_HEAP_OPTS: -Xmx512m -Xms512m
    volumes:
    - kafkadata:/var/lib/kafka/data
volumes:
  kafkadata:
    name: lab-kafka-data
EOF
COMPOSE="docker compose -f /tmp/lab-kafka-restore/docker-compose.yml"
timeout 180 $COMPOSE up -d
timeout 240 bash -c 'until docker compose -f /tmp/lab-kafka-restore/docker-compose.yml exec -T kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list >/dev/null 2>&1; do sleep 3; done'
timeout 30 $COMPOSE exec -T kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list

命令块按 bash 语法书写；zsh 不会对 $COMPOSE 分词，请在 bash 中执行或把 $COMPOSE 展开为完整命令。CLUSTER_ID 是实验专用的固定值，只用于让同一次演练内的数据和元数据匹配；生产集群必须保留自己的 cluster ID，不能在恢复时重新生成。PLAINTEXT://localhost:9092 供容器内 CLI 使用，EXTERNAL://127.0.0.1:19092 供宿主机客户端使用，两个监听地址不能混用。

创建单分区 topic，并生产 12 条可核对的消息。单分区让 offset 与 lag 的计算保持确定：

timeout 30 $COMPOSE exec -T kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 \
  --create --topic orders --partitions 1 --replication-factor 1 \
  --config min.insync.replicas=1 --config cleanup.policy=delete --config retention.ms=86400000
timeout 30 tee /tmp/lab-kafka-restore/messages.ndjson >/dev/null <<'EOF'
key-01:{"order":1,"amount":100}
key-02:{"order":2,"amount":200}
key-03:{"order":3,"amount":300}
key-04:{"order":4,"amount":400}
key-05:{"order":5,"amount":500}
key-06:{"order":6,"amount":600}
key-07:{"order":7,"amount":700}
key-08:{"order":8,"amount":800}
key-09:{"order":9,"amount":900}
key-10:{"order":10,"amount":1000}
key-11:{"order":11,"amount":1100}
key-12:{"order":12,"amount":1200}
EOF
timeout 60 $COMPOSE exec -T kafka /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server localhost:9092 --topic orders \
  --property parse.key=true --property key.separator=: < /tmp/lab-kafka-restore/messages.ndjson
timeout 30 $COMPOSE exec -T kafka /opt/kafka/bin/kafka-get-offsets.sh \
  --bootstrap-server localhost:9092 --topic orders | timeout 30 tee /tmp/lab-kafka-restore/evidence/offsets-before.txt

把消费者组 order-svc 的提交位点设为 8，形成 4 条 lag。这里不启动真实消费者，只写入位点，避免自动提交和 rebalance 造成不确定结果：

timeout 30 $COMPOSE exec -T kafka /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group order-svc --topic orders --reset-offsets --to-offset 8 --dry-run \
  | timeout 30 tee /tmp/lab-kafka-restore/evidence/offset-dryrun.txt
timeout 30 $COMPOSE exec -T kafka /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group order-svc --topic orders --reset-offsets --to-offset 8 --execute
timeout 30 $COMPOSE exec -T kafka /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group order-svc | timeout 30 tee /tmp/lab-kafka-restore/evidence/group-before.txt
timeout 30 $COMPOSE exec -T kafka /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group order-svc | timeout 30 awk '$1=="order-svc" && $2=="orders" {print "lag=" $6}'

保存 topic 元数据、配置和完整消息基线。消息内容哈希用于恢复后逐字对比，不能只看 topic 是否存在：

timeout 30 $COMPOSE exec -T kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 \
  --describe --topic orders | timeout 30 tee /tmp/lab-kafka-restore/evidence/describe-before.txt
timeout 30 $COMPOSE exec -T kafka /opt/kafka/bin/kafka-configs.sh --bootstrap-server localhost:9092 \
  --describe --entity-type topics --entity-name orders | timeout 30 tee /tmp/lab-kafka-restore/evidence/config-before.txt
timeout 60 $COMPOSE exec -T kafka /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic orders --from-beginning --max-messages 12 --property print.key=true --property print.offset=true \
  | timeout 30 tee /tmp/lab-kafka-restore/evidence/messages-before.txt
timeout 30 shasum -a 256 /tmp/lab-kafka-restore/evidence/messages-before.txt \
  | timeout 30 tee /tmp/lab-kafka-restore/evidence/messages-before.sha256

演练步骤

1. 停机备份数据目录

先停止 broker，确认容器不再写入，再复制整个 log.dirs。这份副本同时包含 topic 元数据、消息日志和 __consumer_offsets，必须成对恢复，不能只挑数据日志：

timeout 60 $COMPOSE stop kafka
timeout 30 docker ps -a --filter name=lab-kafka-restore --format '{{.Names}} {{.Status}}'
timeout 60 docker cp lab-kafka-restore:/var/lib/kafka/data /tmp/lab-kafka-restore/backup/data
timeout 30 du -sh /tmp/lab-kafka-restore/backup/data
timeout 30 find /tmp/lab-kafka-restore/backup/data -maxdepth 2 -type f | head -20

2. 备份后继续写入并记录 RPO

重新启动 broker，追加 2 条消息。此时 end offset 应为 14，而备份只包含 12 条；恢复后丢失的 2 条就是本次演练的 RPO。记录后再停止 broker，准备模拟数据丢失：

timeout 120 $COMPOSE start kafka
timeout 240 bash -c 'until docker compose -f /tmp/lab-kafka-restore/docker-compose.yml exec -T kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list >/dev/null 2>&1; do sleep 3; done'
timeout 30 tee /tmp/lab-kafka-restore/post-backup.ndjson >/dev/null <<'EOF'
key-13:{"order":13,"amount":1300}
key-14:{"order":14,"amount":1400}
EOF
timeout 60 $COMPOSE exec -T kafka /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server localhost:9092 --topic orders \
  --property parse.key=true --property key.separator=: < /tmp/lab-kafka-restore/post-backup.ndjson
timeout 30 $COMPOSE exec -T kafka /opt/kafka/bin/kafka-get-offsets.sh \
  --bootstrap-server localhost:9092 --topic orders | timeout 30 tee /tmp/lab-kafka-restore/evidence/offsets-post-backup.txt
timeout 60 $COMPOSE stop kafka

3. 删除数据卷并从备份恢复

删除容器和固定名卷，重建空环境，再把备份内容复制回卷。复制后必须修正数据目录属主，确保 broker 运行用户可以读写；不要用 root 直接运行 broker 来绕过权限问题：

timeout 120 $COMPOSE down
timeout 30 docker volume rm lab-kafka-data 2>/dev/null || true
timeout 120 $COMPOSE create
timeout 60 docker cp -a /tmp/lab-kafka-restore/backup/data/. lab-kafka-restore:/var/lib/kafka/data
timeout 60 docker run --rm -u 0 -v lab-kafka-data:/data --entrypoint chown "$KAFKA_IMAGE" -R appuser /data
timeout 120 $COMPOSE start
timeout 240 bash -c 'until docker compose -f /tmp/lab-kafka-restore/docker-compose.yml exec -T kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list >/dev/null 2>&1; do sleep 3; done'
timeout 30 docker logs lab-kafka-restore 2>&1 | tail -40 | timeout 30 tee /tmp/lab-kafka-restore/evidence/broker-restore.log

4. 核对元数据、消息与位点

先看 topic 描述和配置，再看 end offset、消费者位点和 lag，最后对比完整消息哈希。恢复后的 end offset 应是 12、提交位点 8、lag 4；如果 topic 存在但消息为空，说明只恢复了元数据或复制了错误目录：

timeout 30 $COMPOSE exec -T kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 \
  --describe --topic orders | timeout 30 tee /tmp/lab-kafka-restore/evidence/describe-after.txt
timeout 30 $COMPOSE exec -T kafka /opt/kafka/bin/kafka-configs.sh --bootstrap-server localhost:9092 \
  --describe --entity-type topics --entity-name orders | timeout 30 tee /tmp/lab-kafka-restore/evidence/config-after.txt
timeout 30 $COMPOSE exec -T kafka /opt/kafka/bin/kafka-get-offsets.sh \
  --bootstrap-server localhost:9092 --topic orders | timeout 30 tee /tmp/lab-kafka-restore/evidence/offsets-after.txt
timeout 30 diff -u /tmp/lab-kafka-restore/evidence/offsets-before.txt /tmp/lab-kafka-restore/evidence/offsets-after.txt
timeout 30 $COMPOSE exec -T kafka /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group order-svc | timeout 30 tee /tmp/lab-kafka-restore/evidence/group-after.txt
timeout 30 $COMPOSE exec -T kafka /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group order-svc | timeout 30 awk '$1=="order-svc" && $2=="orders" {print "lag=" $6}'
timeout 60 $COMPOSE exec -T kafka /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic orders --from-beginning --max-messages 12 --property print.key=true --property print.offset=true \
  | timeout 30 tee /tmp/lab-kafka-restore/evidence/messages-after.txt
timeout 30 shasum -a 256 /tmp/lab-kafka-restore/evidence/messages-after.txt
timeout 30 diff -u /tmp/lab-kafka-restore/evidence/config-before.txt /tmp/lab-kafka-restore/evidence/config-after.txt
timeout 30 diff -u /tmp/lab-kafka-restore/evidence/messages-before.txt /tmp/lab-kafka-restore/evidence/messages-after.txt

5. 演练位点丢失后的重置

把提交位点错误地重置到 2，观察 lag 变为 10；再用 dry-run 和 execute 恢复到 8，确认 lag 回到 4。整个过程先停消费者，避免活跃成员让重置失败或产生新的提交：

timeout 30 $COMPOSE exec -T kafka /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group order-svc --topic orders --reset-offsets --to-offset 2 --execute
timeout 30 $COMPOSE exec -T kafka /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group order-svc | timeout 30 tee /tmp/lab-kafka-restore/evidence/group-offset-2.txt
timeout 30 $COMPOSE exec -T kafka /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group order-svc --topic orders --reset-offsets --to-offset 8 --dry-run \
  | timeout 30 tee /tmp/lab-kafka-restore/evidence/reset-to-8-dryrun.txt
timeout 30 $COMPOSE exec -T kafka /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group order-svc --topic orders --reset-offsets --to-offset 8 --execute
timeout 30 $COMPOSE exec -T kafka /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group order-svc | timeout 30 tee /tmp/lab-kafka-restore/evidence/group-offset-8.txt

若当前 Kafka 版本拒绝为不存在的消费者组重置位点，先用同一 group 正常消费并提交 8 条，或改为对已有组做重置；不要在生产反复尝试。

观察现象

备份前 topic 有 1 个分区、1 个副本、end offset 12，消费者组提交位点 8，lag 4。备份后追加 2 条消息，end offset 变为 14，lag 6；恢复后回到 end offset 12、位点 8、lag 4，备份后的 2 条消息丢失，这就是 RPO。恢复出的 describe、配置和完整消息文件应与基线一致，broker 日志没有 cluster ID、node ID 或权限错误。把位点重置到 2 后 lag 为 10，恢复到 8 后 lag 回到 4。

成功的判据是：broker 正常启动且 controller 活跃；topic 名称、分区数、副本因子和配置与基线一致；end offset 与备份时点一致；消费者组提交位点和 lag 可解释；12 条消息内容哈希一致；位点重置先 dry-run 后 execute，lag 单调回到预期；RPO 和不可用窗口已记录。失败判据是：broker 反复退出；topic 被自动创建成错误分区数；消息存在但 offset 回退或跳变；消费者组位点指向已过期数据；为了启动删除元数据目录或格式化存储；未停消费者就批量重置全部组。

判断与定位

1. 先看 broker 是否真的形成 quorum。kafka-metadata-quorum.sh describe --status 应显示 active controller；容器 Running、端口可连，不代表元数据日志可用。
2. topic 存在但消息为空，通常是只恢复了数据日志、没有恢复对应元数据，或复制了错误的 log.dirs。数据与元数据必须来自同一时点、同一 cluster ID 和 node ID。
3. end offset 比预期少，先确认备份时点，再看是否有备份后写入。不要把正常 RPO 当成恢复失败，也不要用重放上游消息来掩盖索引或位点错误。
4. 消费者 lag 异常时，分别看 CURRENT-OFFSET、LOG-END-OFFSET 和组状态。位点大于日志终点会得到负 lag，说明恢复到了更早的数据或位点来自另一个集群。
5. 位点重置可能造成重复消费或漏数。重置到 earliest 会重复，重置到 latest 会跳过中间消息；必须先 dry-run、记录当前位点，并确认业务幂等和可接受的数据窗口。
6. 单节点实验不能证明生产多副本恢复。生产还要检查 ISR、min.insync.replicas、controller quorum、机架分布和跨节点快照一致性，不能用本 Lab 的单节点结果直接承诺 RTO。

恢复与回滚

如果 broker 无法启动，先保留容器日志、数据目录和备份，检查 cluster ID、node ID、meta.properties、目录权限和卷挂载。不要用格式化命令“修复”元数据目录，也不要把不同节点的数据目录拼在一起；格式化会生成新的 cluster ID，旧数据将无法再加入 quorum。

如果快照或备份数据不一致，立即停止所有 broker，重新从上一份通过校验的备份恢复全部节点。单节点实验只需删除卷并重新复制备份；生产环境必须所有节点使用同一时点快照，先让 controller quorum 形成，再启动 broker。若位点错误，停止消费者并记录当前 offset，用 dry-run 确认目标后再执行，无法确定时宁可回到 earliest 配合幂等消费，也不要直接跳到 latest 静默漏数。

回滚的前提是旧数据目录和旧集群仍然保留。若恢复集群尚未接客，停止它并切回旧集群；若旧集群在故障后继续接收写入，先核对 offset 和业务差异，不能只改 DNS。实验结束先归档证据，再清理本项目的容器和固定名卷：

timeout 120 $COMPOSE down -v --remove-orphans
timeout 30 docker volume rm lab-kafka-data 2>/dev/null || true
timeout 30 rm -rf /tmp/lab-kafka-restore

验证

• broker 启动成功，controller 活跃，日志没有 cluster ID、node ID、权限或元数据日志错误。
• topic orders 存在，分区数 1、副本因子 1，min.insync.replicas、清理策略和保留配置与基线一致。
• 分区 end offset 为 12，与备份时点一致；备份后的 13、14 号消息确实丢失，RPO 为 2 条。
• 12 条消息的 key、offset、内容和 SHA-256 与基线完全一致，没有跨分区错位或重复。
• 消费者组 order-svc 的提交位点为 8，lag 为 4；位点重置到 2 时 lag 为 10，恢复到 8 后回到 4。
• __consumer_offsets 等内部 topic 存在且可服务；未使用事务时 __transaction_state 可能不存在。没有 active member 时才能安全重置。
• kafka-topics.sh --describe --under-replicated-partitions 没有输出；单节点实验仍需说明它不能替代生产 ISR 验证。
• RPO 消息数、不可用窗口、恢复命令和验证结果已归档，临时卷和目录已清理。

复盘

• 为什么复制 Kafka 数据目录前必须停止 broker？在线复制可能得到什么样的不一致文件？
• 为什么恢复 topic 数据时必须同时恢复 KRaft 元数据？cluster ID 和 node ID 不一致会怎样？
• 恢复后 end offset 比源集群少 2 条，如何判断这是正常 RPO 还是恢复失败？
• 消费者位点重置到 earliest 和 latest 分别有什么业务风险？为什么必须先 dry-run？
• 单节点实验能证明生产多副本集群的恢复能力吗？还缺哪些验证？

相关命令

• runbook-kafka-restore — topic、元数据、副本与消费者位点恢复
• kafka — topic、分区、副本和消费者组概念
• kafka-broker — broker 启停、日志目录与副本重分配
• kafka-topics — 创建、描述和修改 topic
• kafka-cheatsheet — 生产消费与消费组命令速查
• kafka-under-replicated — ISR 不足的定位与修复
• kafka-consumer-lag — lag 与 rebalance 排查
• kafka-disk-full — 磁盘写满与日志目录故障
• build-kafka-cluster — 新集群搭建与副本规划
• lab-kafka-broker-replace — 副本迁移与 broker 替换演练

参考链接

• Apache Kafka: Basic Operations
• Apache Kafka: Replication

读完自测

1. 为什么恢复 Kafka 时必须把数据日志和 KRaft 元数据成对恢复，不能只复制分区目录？
2. 恢复后消费者提交位点超过日志 end offset，应该怎么处理？
3. 为什么本 Lab 要在备份后追加 2 条消息，再执行恢复？