Kafka KRaft 集群在 Kubernetes 上的部署实践

Kafka 3.6 引入了 KRaft 模式，不再依赖 ZooKeeper，让集群架构更简洁。本文记录在 K8s 上用 bitnami/kafka:3.6.2 部署两套 KRaft 集群的过程——一套用于内部服务通信，另一套带 SASL 认证对外暴露给 ETL 管线消费。

一、两套集群的设计思路

为什么要部署两套？内部集群走 PLAINTEXT 直连，简单高效；ETL 集群面向外部系统，必须走 SASL_PLAINTEXT 认证，保障数据边界安全。两者共用 csjg-kafka 命名空间，但 KAFKA_KRAFT_CLUSTER_ID 不同，互不干扰。

ETL 集群三个 Pod 跑在同一台机器上（nodeSelector: kafka-server=etl-01），通过不同端口区分（321xx / 322xx / 323xx）。

二、KRaft 核心概念

KRaft 模式下每个节点同时担任 controller 和 broker（由 KAFKA_CFG_PROCESS_ROLES 控制），集群通过 KAFKA_CFG_CONTROLLER_QUORUM_VOTERS 互相发现，不再需要独立的 ZooKeeper。

三个关键参数必须全局一致：

生成 Cluster ID 的方法：

# 在任意节点上运行
/opt/bitnami/kafka/bin/kafka-storage.sh random-uuid

三、内部集群配置

kafka.yaml 定义了 3 个 StatefulSet，每个对应一个 Kafka 节点，绑定到不同 K8s 节点。

3.1 kafka01 节点配置

apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: kafka01
  namespace: csjg-kafka
spec:
  replicas: 1
  selector:
    matchLabels:
      app: kafka01
  template:
    spec:
      hostNetwork: true
      hostAliases:
      - ip: "<KAFKA_NODE_IP_1>"
        hostnames: ["kafka01"]
      - ip: "<KAFKA_NODE_IP_2>"
        hostnames: ["kafka02"]
      - ip: "<KAFKA_NODE_IP_3>"
        hostnames: ["kafka03"]
      containers:
      - name: kafka01
        env:
        - name: KAFKA_ENABLE_KRAFT
          value: "yes"
        - name: KAFKA_CFG_NODE_ID
          value: "1"
        - name: KAFKA_KRAFT_CLUSTER_ID
          value: "<YOUR_CLUSTER_ID>"
        - name: KAFKA_CFG_PROCESS_ROLES
          value: "controller,broker"
        - name: KAFKA_CFG_CONTROLLER_QUORUM_VOTERS
          value: "1@kafka01:9094,2@kafka02:9094,3@kafka03:9094"
        - name: KAFKA_HEAP_OPTS
          value: "-Xmx4g -Xms4g"
        - name: KAFKA_CFG_LISTENERS
          value: "PLAINTEXT://:9092,CONTROLLER://kafka01:9094,EXTERNAL://:9093"
        - name: KAFKA_CFG_ADVERTISED_LISTENERS
          value: "PLAINTEXT://kafka01:9092,EXTERNAL://kafka-server01:9093"
        - name: KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP
          value: "CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,EXTERNAL:PLAINTEXT"
        - name: KAFKA_CFG_CONTROLLER_LISTENER_NAMES
          value: "CONTROLLER"
        - name: KAFKA_CFG_INTER_BROKER_LISTENER_NAME
          value: "PLAINTEXT"
        - name: KAFKA_CFG_MESSAGE_MAX_BYTES
          value: "10485760"
        - name: KAFKA_CFG_REPLICA_FETCH_MAX_BYTES
          value: "20485760"
        image: <YOUR_REGISTRY>/library/bitnami/kafka:3.6.2
        ports:
        - containerPort: 9091
        - containerPort: 9092
        - containerPort: 9093
        - containerPort: 9094
        volumeMounts:
        - name: kafka-data
          mountPath: /bitnami/kafka/
      volumes:
      - name: kafka-data
        hostPath:
          path: /home/kafka/data/
      nodeSelector:
        kafka-server: "01"

3.2 端口说明

3.3 kafka02 / kafka03 的差异

三个节点配置几乎相同，只需要改以下几处：

四、ETL 对外集群配置

ETL 集群面向外部系统，开启了 SASL_PLAINTEXT 认证。三个 Pod 部署在同一台节点上，通过不同端口隔离。

4.1 kafkaetl01 节点配置

apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: kafkaetl01
  namespace: csjg-kafka
spec:
  replicas: 1
  template:
    spec:
      hostNetwork: true
      hostAliases:
      - ip: "<ETL_NODE_IP>"
        hostnames: ["kafkaetl01"]
      - ip: "<ETL_NODE_IP>"
        hostnames: ["kafkaetl02"]
      - ip: "<ETL_NODE_IP>"
        hostnames: ["kafkaetl03"]
      containers:
      - name: kafkaetl01
        env:
        - name: KAFKA_ENABLE_KRAFT
          value: "yes"
        - name: KAFKA_CFG_NODE_ID
          value: "1"
        - name: KAFKA_KRAFT_CLUSTER_ID
          value: "<YOUR_ETL_CLUSTER_ID>"
        - name: KAFKA_CFG_PROCESS_ROLES
          value: "controller,broker"
        - name: KAFKA_CFG_CONTROLLER_QUORUM_VOTERS
          value: "1@kafkaetl01:32193,2@kafkaetl02:32293,3@kafkaetl03:32393"
        - name: KAFKA_HEAP_OPTS
          value: "-Xmx4g -Xms4g"
        - name: KAFKA_CFG_LISTENERS
          value: "PLAINTEXT://:32192,CONTROLLER://kafkaetl01:32193,EXTERNAL://:32194"
        - name: KAFKA_CFG_ADVERTISED_LISTENERS
          value: "PLAINTEXT://kafkaetl01:32192,EXTERNAL://kafkaetl-server01:32194"
        - name: KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP
          value: "CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,EXTERNAL:SASL_PLAINTEXT"
        - name: KAFKA_CFG_SASL_ENABLED_MECHANISMS
          value: "PLAIN"
        - name: KAFKA_CLIENT_USERS
          value: "<USER1>,<USER2>,..."
        - name: KAFKA_CLIENT_PASSWORDS
          value: "<PASS1>,<PASS2>,..."
        - name: KAFKA_CFG_LOG_RETENTION_HOURS
          value: "2160"
        image: <YOUR_REGISTRY>/library/bitnami/kafka:3.6.2
        ports:
        - containerPort: 32191
        - containerPort: 32192
        - containerPort: 32193
        - containerPort: 32194
        volumeMounts:
        - name: kafka-data
          mountPath: /bitnami/kafka/
      volumes:
      - name: kafka-data
        hostPath:
          path: /home/kafka/data/1/
      nodeSelector:
        kafka-server: "etl-01"

4.2 ETL 集群端口分配

为什么三个 Pod 在同一台机器？ ETL 集群消费量不大，单机部署节省资源。通过不同端口和不同数据目录（data/1/、data/2/、data/3/）确保数据隔离。

4.3 SASL 认证配置

KAFKA_CLIENT_USERS 和 KAFKA_CLIENT_PASSWORDS 用逗号分隔，一一对应：

users:    localuser, cicc01,  csjg01,  csjg02, ...
passwords: <PASS1>,  <PASS2>, <PASS3>, <PASS4>, ...

客户端连接时需配置 sasl.mechanism=PLAIN 和对应的用户名密码。

4.4 消息保留策略

KAFKA_CFG_LOG_RETENTION_HOURS=2160 即 90 天，适用于 ETL 场景中的历史数据回溯需求。

五、部署步骤

5.1 内部集群

# 为节点打标签
kubectl label nodes <NODE_1> kafka-server=01 kafka-app=true
kubectl label nodes <NODE_2> kafka-server=02 kafka-app=true
kubectl label nodes <NODE_3> kafka-server=03 kafka-app=true

# 创建数据目录
mkdir -p /home/kafka/data
chmod -R 777 /home/kafka/data

# 创建命名空间
kubectl create namespace csjg-kafka

# 部署
kubectl apply -f kafka.yaml

5.2 ETL 集群

# 为 ETL 节点打标签
kubectl label nodes <ETL_NODE> kafka-server=etl-01

# 创建三个数据目录
mkdir -p /home/kafka/data/1 /home/kafka/data/2 /home/kafka/data/3
chmod -R 777 /home/kafka/data

# 部署
kubectl apply -f kafka-etl.yaml

5.3 验证

kubectl get pod -n csjg-kafka -o wide

# 查看某个 Pod 的日志
kubectl -n csjg-kafka logs kafka01-0 --all-containers=true

# 进入 Pod
kubectl exec -it pod/kafka01-0 -n csjg-kafka -- bash

六、常用运维命令

进入 Kafka Pod 后可使用的工具：

# 列出所有 Consumer Group
/opt/bitnami/kafka/bin/kafka-consumer-groups.sh --bootstrap-server kafka01:9092 --list

# 查看 Consumer Group 详情
/opt/bitnami/kafka/bin/kafka-consumer-groups.sh --describe --group <GROUP_NAME> --bootstrap-server kafka01:9092

# 查看 Topic 详情
/opt/bitnami/kafka/bin/kafka-topics.sh --describe --topic <TOPIC_NAME> --bootstrap-server kafka01:9092

# 重置 Consumer Offset（跳到最新）
/opt/bitnami/kafka/bin/kafka-consumer-groups.sh --bootstrap-server kafka01:9092 \
  --group <GROUP_NAME> --reset-offsets --all-topics --to-latest --execute

七、注意事项

1. KAFKA_KRAFT_CLUSTER_ID 必须一致 — 同一集群内所有节点使用相同的 ID，否则无法加入集群
2. hostAliases 必须完整 — 每个节点都需要知道所有其他节点的主机名映射
3. hostNetwork 端口冲突 — 确保宿主机上 9091-9094 和 321xx-323xx 端口未被占用
4. JVM 堆内存 — 建议生产环境 Xms 和 Xmx 设为相同值，4g 是最低推荐
5. 数据目录隔离 — ETL 集群在同一台机器上，必须用不同的 hostPath 子目录
6. SASL 密码安全 — KAFKA_CLIENT_PASSWORDS 中包含明文密码，建议通过 K8s Secret 管理
7. advertised_listeners 准确性 — 客户端通过 advertised.listeners 连接，主机名必须可达

附录：资源清单