📚 课程目录第2章集成测试环境搭建

2. 集成测试环境搭建：Docker Compose 部署 Kafka 集群、Zookeeper 配置、网络拓扑设计

集成测试环境是验证 Kafka 任务版本回滚与故障排查机制的基础设施。本章使用 Docker Compose 编排一个生产级 Kafka 集群，包含多 Broker、Zookeeper 仲裁、网络隔离与监控组件，为后续的故障注入与回滚测试提供可复现的沙箱环境。

2.1 网络拓扑设计

测试集群采用三层网络拓扑：管理层（Zookeeper 仲裁）、数据层（Kafka Broker 集群）、接入层（测试客户端与监控工具）。所有容器通过自定义 Docker 网络 kafka-test-net 互联，实现服务发现与端口映射。

  拓扑要点：
  Zookeeper 使用 3 节点仲裁（奇数节点保证 Leader 选举可用性）。
Kafka Broker 初始部署 3 节点，后续可通过 Compose scale 扩展。
单独部署 Kafka Manager（如 CMAK）用于可视化监控。
客户端容器（测试脚本运行环境）与集群网络隔离，通过 Broker 暴露端口通信。

2.2 Docker Compose 服务编排

以下 docker-compose.yml 定义了完整的测试集群。每个服务配置了资源限制、健康检查与持久化卷，确保环境可重复销毁与重建。

version: '3.8'

networks:
  kafka-test-net:
    driver: bridge
    ipam:
      config:
        - subnet: 172.20.0.0/16

services:
  # ---------- Zookeeper 仲裁 ----------
  zk1:
    image: confluentinc/cp-zookeeper:7.5.0
    hostname: zk1
    container_name: zk1
    networks:
      kafka-test-net:
        ipv4_address: 172.20.0.11
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_SERVER_ID: 1
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
      ZOOKEEPER_INIT_LIMIT: 5
      ZOOKEEPER_SYNC_LIMIT: 2
      ZOOKEEPER_SERVERS: "zk1:2888:3888;zk2:2888:3888;zk3:2888:3888"
    volumes:
      - zk1-data:/var/lib/zookeeper/data
      - zk1-log:/var/lib/zookeeper/log
    healthcheck:
      test: ["CMD", "bash", "-c", "echo ruok | nc localhost 2181"]
      interval: 10s
      timeout: 5s
      retries: 3

  zk2:
    image: confluentinc/cp-zookeeper:7.5.0
    hostname: zk2
    container_name: zk2
    networks:
      kafka-test-net:
        ipv4_address: 172.20.0.12
    ports:
      - "2182:2181"
    environment:
      ZOOKEEPER_SERVER_ID: 2
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
      ZOOKEEPER_INIT_LIMIT: 5
      ZOOKEEPER_SYNC_LIMIT: 2
      ZOOKEEPER_SERVERS: "zk1:2888:3888;zk2:2888:3888;zk3:2888:3888"
    volumes:
      - zk2-data:/var/lib/zookeeper/data
      - zk2-log:/var/lib/zookeeper/log
    healthcheck:
      test: ["CMD", "bash", "-c", "echo ruok | nc localhost 2181"]
      interval: 10s
      timeout: 5s
      retries: 3

  zk3:
    image: confluentinc/cp-zookeeper:7.5.0
    hostname: zk3
    container_name: zk3
    networks:
      kafka-test-net:
        ipv4_address: 172.20.0.13
    ports:
      - "2183:2181"
    environment:
      ZOOKEEPER_SERVER_ID: 3
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
      ZOOKEEPER_INIT_LIMIT: 5
      ZOOKEEPER_SYNC_LIMIT: 2
      ZOOKEEPER_SERVERS: "zk1:2888:3888;zk2:2888:3888;zk3:2888:3888"
    volumes:
      - zk3-data:/var/lib/zookeeper/data
      - zk3-log:/var/lib/zookeeper/log
    healthcheck:
      test: ["CMD", "bash", "-c", "echo ruok | nc localhost 2181"]
      interval: 10s
      timeout: 5s
      retries: 3

  # ---------- Kafka Broker 集群 ----------
  kafka1:
    image: confluentinc/cp-kafka:7.5.0
    hostname: kafka1
    container_name: kafka1
    networks:
      kafka-test-net:
        ipv4_address: 172.20.0.21
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: "zk1:2181,zk2:2181,zk3:2181"
      KAFKA_ADVERTISED_LISTENERS: "PLAINTEXT://172.20.0.21:9092"
      KAFKA_LISTENERS: "PLAINTEXT://0.0.0.0:9092"
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2
      KAFKA_DEFAULT_REPLICATION_FACTOR: 3
      KAFKA_NUM_PARTITIONS: 3
      KAFKA_LOG_RETENTION_HOURS: 168
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false"
    volumes:
      - kafka1-data:/var/lib/kafka/data
    depends_on:
      zk1: { condition: service_healthy }
      zk2: { condition: service_healthy }
      zk3: { condition: service_healthy }
    healthcheck:
      test: ["CMD", "kafka-broker-api-versions", "--bootstrap-server", "localhost:9092"]
      interval: 15s
      timeout: 10s
      retries: 5

  kafka2:
    image: confluentinc/cp-kafka:7.5.0
    hostname: kafka2
    container_name: kafka2
    networks:
      kafka-test-net:
        ipv4_address: 172.20.0.22
    ports:
      - "9093:9092"
    environment:
      KAFKA_BROKER_ID: 2
      KAFKA_ZOOKEEPER_CONNECT: "zk1:2181,zk2:2181,zk3:2181"
      KAFKA_ADVERTISED_LISTENERS: "PLAINTEXT://172.20.0.22:9092"
      KAFKA_LISTENERS: "PLAINTEXT://0.0.0.0:9092"
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2
      KAFKA_DEFAULT_REPLICATION_FACTOR: 3
      KAFKA_NUM_PARTITIONS: 3
      KAFKA_LOG_RETENTION_HOURS: 168
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false"
    volumes:
      - kafka2-data:/var/lib/kafka/data
    depends_on:
      zk1: { condition: service_healthy }
      zk2: { condition: service_healthy }
      zk3: { condition: service_healthy }
    healthcheck:
      test: ["CMD", "kafka-broker-api-versions", "--bootstrap-server", "localhost:9092"]
      interval: 15s
      timeout: 10s
      retries: 5

  kafka3:
    image: confluentinc/cp-kafka:7.5.0
    hostname: kafka3
    container_name: kafka3
    networks:
      kafka-test-net:
        ipv4_address: 172.20.0.23
    ports:
      - "9094:9092"
    environment:
      KAFKA_BROKER_ID: 3
      KAFKA_ZOOKEEPER_CONNECT: "zk1:2181,zk2:2181,zk3:2181"
      KAFKA_ADVERTISED_LISTENERS: "PLAINTEXT://172.20.0.23:9092"
      KAFKA_LISTENERS: "PLAINTEXT://0.0.0.0:9092"
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2
      KAFKA_DEFAULT_REPLICATION_FACTOR: 3
      KAFKA_NUM_PARTITIONS: 3
      KAFKA_LOG_RETENTION_HOURS: 168
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false"
    volumes:
      - kafka3-data:/var/lib/kafka/data
    depends_on:
      zk1: { condition: service_healthy }
      zk2: { condition: service_healthy }
      zk3: { condition: service_healthy }
    healthcheck:
      test: ["CMD", "kafka-broker-api-versions", "--bootstrap-server", "localhost:9092"]
      interval: 15s
      timeout: 10s
      retries: 5

  # ---------- 测试客户端 ----------
  test-client:
    image: python:3.11-slim
    container_name: test-client
    networks:
      kafka-test-net:
        ipv4_address: 172.20.0.100
    command: ["tail", "-f", "/dev/null"]
    volumes:
      - ./test-scripts:/test-scripts
    depends_on:
      kafka1: { condition: service_healthy }
      kafka2: { condition: service_healthy }
      kafka3: { condition: service_healthy }

  # ---------- Kafka Manager (CMAK) ----------
  cmak:
    image: ghcr.io/eshepelyuk/docker-cmak:3.0.0.6
    hostname: cmak
    container_name: cmak
    networks:
      kafka-test-net:
        ipv4_address: 172.20.0.200
    ports:
      - "9000:9000"
    environment:
      ZK_HOSTS: "zk1:2181,zk2:2181,zk3:2181"
      APPLICATION_SECRET: "test-secret"
    depends_on:
      zk1: { condition: service_healthy }
      zk2: { condition: service_healthy }
      zk3: { condition: service_healthy }

volumes:
  zk1-data:   zk2-data:   zk3-data:
  kafka1-data: kafka2-data: kafka3-data:

2.3 Zookeeper 配置详解

Zookeeper 仲裁的稳定性直接影响 Kafka 集群的元数据一致性。以下参数在测试环境中必须显式配置：

参数	值	说明
`ZOOKEEPER_SERVER_ID`	1/2/3	每个节点唯一 ID，与 `myid` 文件对应
`ZOOKEEPER_CLIENT_PORT`	2181	客户端连接端口，容器内统一使用 2181
`ZOOKEEPER_TICK_TIME`	2000	基本时间单元（ms），影响会话超时与心跳间隔
`ZOOKEEPER_INIT_LIMIT`	5	Leader 初始化连接的最大 tick 数
`ZOOKEEPER_SYNC_LIMIT`	2	Follower 与 Leader 同步的最大 tick 数
`ZOOKEEPER_SERVERS`	zk1:2888:3888;zk2:2888:3888;zk3:2888:3888	仲裁成员列表：`host:leaderPort:electionPort`

测试建议：在故障回滚测试中，可以通过停止 ZK 节点（docker stop zk1）模拟仲裁丢失，观察 Kafka 集群是否进入“仅元数据可用”状态。恢复 ZK 后，Kafka 应自动重新加载元数据。

2.4 Kafka Broker 关键配置

Broker 配置中，advertised.listeners 必须使用 Docker 网络 IP 而非 localhost，否则客户端容器无法连接。以下为测试环境特有的参数调整：

参数	值	测试用途
`KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR`	3	确保 offset 主题高可用，模拟生产环境
`KAFKA_DEFAULT_REPLICATION_FACTOR`	3	自动创建主题的副本数，测试分区迁移时使用
`KAFKA_NUM_PARTITIONS`	3	默认分区数，便于测试并行消费与再均衡
`KAFKA_AUTO_CREATE_TOPICS_ENABLE`	false	禁止自动创建，测试脚本必须显式创建主题
`KAFKA_LOG_RETENTION_HOURS`	168	日志保留 7 天，避免测试数据堆积撑爆磁盘

2.5 启动与验证步骤

启动集群：docker compose -f docker-compose.yml up -d
检查所有服务健康状态：docker compose ps，确保所有容器显示 healthy
验证 Zookeeper 仲裁：进入任意 ZK 容器执行 echo stat | nc localhost 2181 | grep Mode，应看到 Mode: leader 或 Mode: follower
验证 Kafka 集群：进入 kafka1 容器执行 kafka-broker-api-versions --bootstrap-server localhost:9092，返回 Broker 版本信息即成功
创建测试主题：kafka-topics --bootstrap-server kafka1:9092 --create --topic test-rollback --partitions 3 --replication-factor 3
验证主题分布：kafka-topics --bootstrap-server kafka1:9092 --describe --topic test-rollback，确认分区均匀分布在 3 个 Broker 上

常见问题：如果客户端容器无法连接 Broker，请检查 KAFKA_ADVERTISED_LISTENERS 是否设置为 Docker 网络 IP（如 172.20.0.21），而非 localhost 或 127.0.0.1。生产环境使用主机名或负载均衡器地址。

2.6 网络隔离与故障注入准备

集成测试需要模拟网络分区、节点宕机等故障场景。Docker Compose 网络设计支持以下故障注入方式：

节点停止：docker stop kafka2 模拟 Broker 宕机，观察 Leader 选举与 ISR 收缩。
网络隔离：docker network disconnect kafka-test-net kafka1 模拟网络分区，测试客户端重连与请求超时处理。
延迟注入：使用 tc 命令在容器内增加网络延迟（需容器支持 NET_ADMIN 权限）。
磁盘故障：通过 docker volume 操作模拟数据目录丢失，测试日志恢复与数据一致性。

2.7 环境清理与重置

每次测试结束后，建议执行完整清理，避免残留数据影响下一轮测试结果：

# 停止并移除所有容器
docker compose -f docker-compose.yml down

# 删除所有数据卷（谨慎操作，会丢失所有测试数据）
docker volume rm kafka-test_zk1-data kafka-test_zk2-data kafka-test_zk3-data \
  kafka-test_kafka1-data kafka-test_kafka2-data kafka-test_kafka3-data

# 可选：清理未使用的网络
docker network prune -f

  最佳实践：在 CI/CD 流水线中，每次集成测试前执行 docker compose down -v 确保环境完全干净。对于需要保留测试日志的场景，可将日志卷挂载到宿主机目录，避免数据丢失。