Kafka KRaft 三节点 ARM64 Docker 部署

适用环境：Linux ARM64、Docker、KRaft 无 Zookeeper 架构、SASL_SSL 双向加密集群

1 前置安全规范

2 阶段 0：全局变量 & 集群环境文件

三台节点最终 cluster.env 文件内容必须完全一致，推荐节点 1 修改 IP 后同步至 2、3 节点。

2.1 创建部署根目录

mkdir -p /opt/kafka-deploy
cd /opt/kafka-deploy

2.2 全局变量定义（替换标记「自行替换」的 IP/CIDR）

# 部署工作目录
DEPLOY_DIR="/opt/kafka-deploy"
# Kafka数据、证书总根
KAFKA_ROOT="/data/kafka"
# Docker容器固定运行用户，不可修改
KAFKA_UID="1000"
KAFKA_GID="1000"

# ============ 三台服务器IP（必须替换真实地址）============
NODE_1_IP="节点1内网IP，自行替换"
NODE_1_PUBLIC_IP="节点1公网IP，自行替换"
NODE_2_IP="节点2内网IP，自行替换"
NODE_2_PUBLIC_IP="节点2公网IP，自行替换"
NODE_3_IP="节点3内网IP，自行替换"
NODE_3_PUBLIC_IP="节点3公网IP，自行替换"
# 业务服务访问内网网段
CLIENT_CIDR="业务私网CIDR，自行替换"

# 端口规划
CLIENT_PORT="19092" # 业务SASL_SSL客户端端口
CONTROLLER_PORT="19093" # KRaft控制器内部mTLS端口
BROKER_PORT="19094" # Broker副本同步mTLS端口

# 固定镜像版本+防篡改哈希
KAFKA_IMAGE="apache/kafka:4.3.1"
KAFKA_UI_IMAGE_TAG="ghcr.io/kafbat/kafka-ui:v1.5.0"
KAFKA_UI_IMAGE="${KAFKA_UI_IMAGE_TAG}@sha256:7cda86a33344160309fdb65146332e4da65db81a945614f2fe32e210803f6fd1"
KAFKA_UI_ARM64_MANIFEST_DIGEST="sha256:3928503e8b05193c9bfed691e08adc0ef5ad0063ea5fb2ca70941e1e2832d936"
KAFKA_UI_IMAGE_ID="sha256:d6b6560726ee79f7b3d985048675b23d55ab3defb70be82cf7b7792fa54fad71"

# 容器资源限制
KAFKA_MEMORY="10g"
KAFKA_CPU_LIMIT="6"
KAFKA_HEAP="6g"
KAFKA_NOFILE="100000"

# 消息留存策略：7天自动清理
KAFKA_RETENTION_HOURS="168"
KAFKA_RETENTION_MS="604800000"
KAFKA_SEGMENT_BYTES="1073741824"
KAFKA_DISK_BUDGET_BYTES="1099511627776"

# 集群超级管理员
ADMIN_USER="platform-admin"
# 加密算法
SCRAM_MECHANISM="SCRAM-SHA-512"

# Kafka UI配置（仅节点1部署）
KAFKA_UI_ROOT="/data/kafka-ui"
KAFKA_UI_CONTAINER_NAME="kafka-ui"
KAFKA_UI_BIND_ADDRESS="0.0.0.0"
KAFKA_UI_HOST_PORT="10002"
KAFKA_UI_ALLOWED_CIDR="0.0.0.0/0"
KAFKA_UI_CONTAINER_PORT="8080"
KAFKA_UI_KAFKA_USER="kafka-ui-readonly"
KAFKA_UI_WEB_USER="kafka-ui-operator"
KAFKA_UI_MEMORY="1g"
KAFKA_UI_HEAP="768m"
KAFKA_UI_CPU_LIMIT="1"
KAFKA_UI_PIDS_LIMIT="256"

# 集群连接自动拼接（无需手动修改）
CONTROLLER_QUORUM_VOTERS="1@${NODE_1_IP}:${CONTROLLER_PORT},2@${NODE_2_IP}:${CONTROLLER_PORT},3@${NODE_3_IP}:${CONTROLLER_PORT}"
KAFKA_BOOTSTRAP_SERVERS="${NODE_1_IP}:${CLIENT_PORT},${NODE_2_IP}:${CLIENT_PORT},${NODE_3_IP}:${CLIENT_PORT}"

# 派生目录路径
KAFKA_BOOTSTRAP_DIR="${KAFKA_ROOT}/bootstrap"
KAFKA_CA_DIR="${KAFKA_BOOTSTRAP_DIR}/ca"
KAFKA_BUNDLE_DIR="${KAFKA_BOOTSTRAP_DIR}/bundles"
KAFKA_TLS_DIR="${KAFKA_ROOT}/tls"
KAFKA_SECRETS_DIR="${KAFKA_ROOT}/secrets"
KAFKA_DATA_DIR="${KAFKA_ROOT}/data"
KAFKA_METADATA_DIR="${KAFKA_ROOT}/metadata"
KAFKA_LOGS_DIR="${KAFKA_ROOT}/logs"
KAFKA_CONFIG_DIR="${KAFKA_ROOT}/config"
KAFKA_SERVER_PROPERTIES="${KAFKA_CONFIG_DIR}/server.properties"
KAFKA_UI_SECRETS_DIR="${KAFKA_UI_ROOT}/secrets"
KAFKA_UI_ENV_FILE="${KAFKA_UI_SECRETS_DIR}/kafka-ui.env"
CLUSTER_ENV_FILE="${DEPLOY_DIR}/cluster.env"

2.3 写入集群环境文件

# 校验是否存在未替换占位符
for v in NODE_1_IP NODE_2_IP NODE_3_IP CLIENT_CIDR; do
eval "val=\\${$v}"
[[ "${val}" != *自行替换* ]] || { echo "错误：请修改IP变量"; exit 1; }
done

# 生成env配置文件
cat > "${CLUSTER_ENV_FILE}" <<EOF
#!/usr/bin/env bash
# 集群公共配置，不含明文密码
DEPLOY_DIR="${DEPLOY_DIR}"
KAFKA_ROOT="${KAFKA_ROOT}"
KAFKA_UID="${KAFKA_UID}"
KAFKA_GID="${KAFKA_GID}"
NODE_1_IP="${NODE_1_IP}"
NODE_1_PUBLIC_IP="${NODE_1_PUBLIC_IP}"
NODE_2_IP="${NODE_2_IP}"
NODE_2_PUBLIC_IP="${NODE_2_PUBLIC_IP}"
NODE_3_IP="${NODE_3_IP}"
NODE_3_PUBLIC_IP="${NODE_3_PUBLIC_IP}"
CLIENT_CIDR="${CLIENT_CIDR}"
CLIENT_PORT="${CLIENT_PORT}"
CONTROLLER_PORT="${CONTROLLER_PORT}"
BROKER_PORT="${BROKER_PORT}"
KAFKA_IMAGE="${KAFKA_IMAGE}"
KAFKA_UI_IMAGE_TAG="${KAFKA_UI_IMAGE_TAG}"
KAFKA_UI_IMAGE="${KAFKA_UI_IMAGE}"
KAFKA_UI_ARM64_MANIFEST_DIGEST="${KAFKA_UI_ARM64_MANIFEST_DIGEST}"
KAFKA_UI_IMAGE_ID="${KAFKA_UI_IMAGE_ID}"
KAFKA_MEMORY="${KAFKA_MEMORY}"
KAFKA_CPU_LIMIT="${KAFKA_CPU_LIMIT}"
KAFKA_HEAP="${KAFKA_HEAP}"
KAFKA_NOFILE="${KAFKA_NOFILE}"
KAFKA_RETENTION_HOURS="${KAFKA_RETENTION_HOURS}"
KAFKA_RETENTION_MS="${KAFKA_RETENTION_MS}"
KAFKA_SEGMENT_BYTES="${KAFKA_SEGMENT_BYTES}"
KAFKA_DISK_BUDGET_BYTES="${KAFKA_DISK_BUDGET_BYTES}"
ADMIN_USER="${ADMIN_USER}"
SCRAM_MECHANISM="${SCRAM_MECHANISM}"
KAFKA_UI_ROOT="${KAFKA_UI_ROOT}"
KAFKA_UI_CONTAINER_NAME="${KAFKA_UI_CONTAINER_NAME}"
KAFKA_UI_BIND_ADDRESS="${KAFKA_UI_BIND_ADDRESS}"
KAFKA_UI_HOST_PORT="${KAFKA_UI_HOST_PORT}"
KAFKA_UI_ALLOWED_CIDR="${KAFKA_UI_ALLOWED_CIDR}"
KAFKA_UI_CONTAINER_PORT="${KAFKA_UI_CONTAINER_PORT}"
KAFKA_UI_KAFKA_USER="${KAFKA_UI_KAFKA_USER}"
KAFKA_UI_WEB_USER="${KAFKA_UI_WEB_USER}"
KAFKA_UI_MEMORY="${KAFKA_UI_MEMORY}"
KAFKA_UI_HEAP="${KAFKA_UI_HEAP}"
KAFKA_UI_CPU_LIMIT="${KAFKA_CPU_LIMIT}"
KAFKA_UI_PIDS_LIMIT="${KAFKA_UI_PIDS_LIMIT}"
CONTROLLER_QUORUM_VOTERS="${CONTROLLER_QUORUM_VOTERS}"
KAFKA_BOOTSTRAP_SERVERS="${KAFKA_BOOTSTRAP_SERVERS}"
KAFKA_BOOTSTRAP_DIR="${KAFKA_BOOTSTRAP_DIR}"
KAFKA_CA_DIR="${KAFKA_CA_DIR}"
KAFKA_BUNDLE_DIR="${KAFKA_BUNDLE_DIR}"
KAFKA_TLS_DIR="${KAFKA_TLS_DIR}"
KAFKA_SECRETS_DIR="${KAFKA_SECRETS_DIR}"
KAFKA_DATA_DIR="${KAFKA_DATA_DIR}"
KAFKA_METADATA_DIR="${KAFKA_METADATA_DIR}"
KAFKA_LOGS_DIR="${KAFKA_LOGS_DIR}"
KAFKA_CONFIG_DIR="${KAFKA_CONFIG_DIR}"
KAFKA_SERVER_PROPERTIES="${KAFKA_SERVER_PROPERTIES}"
KAFKA_UI_SECRETS_DIR="${KAFKA_UI_SECRETS_DIR}"
KAFKA_UI_ENV_FILE="${KAFKA_UI_ENV_FILE}"
CLUSTER_ENV_FILE="${CLUSTER_ENV_FILE}"
EOF

# 权限收紧
chown root:root "${CLUSTER_ENV_FILE}"
chmod 640 "${CLUSTER_ENV_FILE}"

# 打印核对信息
echo "三台节点内网IP：${NODE_1_IP} ${NODE_2_IP} ${NODE_3_IP}"
echo "KRaft集群投票地址：${CONTROLLER_QUORUM_VOTERS}"
echo "业务连接地址：${KAFKA_BOOTSTRAP_SERVERS}"
ls -l "${CLUSTER_ENV_FILE}"

2.4 同步文件到另外两台服务器

scp "${CLUSTER_ENV_FILE}" root@${NODE_2_IP}:${CLUSTER_ENV_FILE}
scp "${CLUSTER_ENV_FILE}" root@${NODE_3_IP}:${CLUSTER_ENV_FILE}

2.5 所有操作前置加载环境变量（新开终端必执行）

cd /opt/kafka-deploy
set -a
source ./cluster.env
set +a

2.6 封装 kafka 管理员快捷命令

# 客户端docker封装
kafka_admin_cli() {
docker run –rm –network host –user "${KAFKA_UID}:${KAFKA_GID}" \\
–mount "type=bind,src=${CLIENT_CONFIG},dst=/tmp/admin-client.properties,readonly" \\
–mount "type=bind,src=${KAFKA_TLS_DIR},dst=/tmp/kafka-tls,readonly" \\
–mount "type=bind,src=${KAFKA_SECRETS_DIR},dst=/tmp/kafka-secrets,readonly" \\
"${KAFKA_IMAGE}" "$@"
}

# 根据密码生成临时客户端配置
write_admin_client_config() {
local admin_password="$1"
CLIENT_CONFIG="$(mktemp)"
cat >"${CLIENT_CONFIG}" <<EOF
security.protocol=SASL_SSL
sasl.mechanism=${SCRAM_MECHANISM}
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="${ADMIN_USER}" password="${admin_password}";
ssl.truststore.type=PKCS12
ssl.truststore.location=/tmp/kafka-tls/truststore.p12
ssl.truststore.password=\\${file:/tmp/kafka-secrets/tls-passwords.properties:store_password}
ssl.endpoint.identification.algorithm=https
config.providers=file
config.providers.file.class=org.apache.kafka.common.config.provider.FileConfigProvider
EOF
chown "${KAFKA_UID}:${KAFKA_GID}" "${CLIENT_CONFIG}"
chmod 400 "${CLIENT_CONFIG}"
}

3 阶段 1：三台服务器前置环境校验

加载环境后逐条执行，全部无报错再部署

# 1. 查看服务器架构
uname -m
# 2. Docker运行状态
systemctl is-active docker
# 3. Docker版本
docker –version
# 4. 依赖工具校验
command -v openssl tar sha256sum ip ss mountpoint firewall-cmd
# 5. 系统文件句柄上限
sysctl -n fs.file-max
# 6. /data是否独立挂载数据盘
mountpoint -q /data && echo "数据盘挂载正常"
# 7. 磁盘剩余空间
df -h /data
# 8. 本机内网IP核对
ip -o -4 addr show | awk '{print $4}' | cut -d/ -f1
# 9. 端口占用检查
ss -ltn | grep -E ":${CLIENT_PORT}|:${CONTROLLER_PORT}|:${BROKER_PORT}" || echo "端口空闲"

校验异常处理

4 阶段 2：拉取 Docker 镜像

4.1 Kafka 镜像（三台都执行）

docker pull "${KAFKA_IMAGE}"
# 校验CPU架构
docker image inspect "${KAFKA_IMAGE}" –format '{{.Architecture}} {{.Os}}'
# 记录镜像哈希用于存档
docker image inspect "${KAFKA_IMAGE}" –format '{{index .RepoDigests 0}}'

4.2 Kafka UI 镜像（仅节点 1）

docker pull "${KAFKA_UI_IMAGE}"
# 校验架构与镜像ID
docker image inspect "${KAFKA_UI_IMAGE}" –format '{{.Architecture}} {{.Id}}'

5 阶段 3：防火墙永久放行端口（三台统一）

# 获取当前防火墙区域
ZONE=$(firewall-cmd –get-active-zones)

# 放行业务端口，限制业务网段来源
firewall-cmd –permanent –zone="${ZONE}" \\
–add-rich-rule="rule family=\\"ipv4\\" source address=\\"${CLIENT_CIDR}\\" port port=\\"${CLIENT_PORT}\\" protocol=\\"tcp\\" accept"

# 三节点互相放行集群通信端口
for src in "${NODE_1_IP}" "${NODE_2_IP}" "${NODE_3_IP}"; do
firewall-cmd –permanent –zone="${ZONE}" \\
–add-rich-rule="rule family=\\"ipv4\\" source address=\\"${src}\\" port port=\\"${CONTROLLER_PORT}\\" protocol=\\"tcp\\" accept"
firewall-cmd –permanent –zone="${ZONE}" \\
–add-rich-rule="rule family=\\"ipv4\\" source address=\\"${src}\\" port port=\\"${BROKER_PORT}\\" protocol=\\"tcp\\" accept"
done

# 重载生效
firewall-cmd –reload
# 查看放行规则
firewall-cmd –zone="${ZONE}" –list-rich-rules

6 阶段 4：CA 证书 & 集群安全包（仅节点 1 执行）

说明：仅节点 1 生成根密钥（shell ca 脚手架），禁止分发；每台独立证书包，全局 ClusterID 三台完全一致。

6.1 创建证书目录收紧权限

umask 077
mkdir -p "${KAFKA_CA_DIR}" "${KAFKA_BUNDLE_DIR}"
chmod 700 "${KAFKA_BOOTSTRAP_DIR}" "${KAFKA_CA_DIR}" "${KAFKA_BUNDLE_DIR}"

# 输入证书统一密码
read -r -s -p '输入证书仓库密码：' TLS_PASSWORD; echo
read -r -s -p '确认密码：' TLS_PASSWORD2; echo
if [[ "${TLS_PASSWORD}" != "${TLS_PASSWORD2}" ]]; then
echo "两次密码不一致，退出"
exit 1
fi
unset TLS_PASSWORD2
# 临时密码文件
TLS_PASSWORD_FILE="$(mktemp)"
printf '%s' "${TLS_PASSWORD}" >"${TLS_PASSWORD_FILE}"
chmod 600 "${TLS_PASSWORD_FILE}"

6.2 生成全局集群 ID

CLUSTER_ID="$(docker run –rm –entrypoint /opt/kafka/bin/kafka-storage.sh "${KAFKA_IMAGE}" random-uuid)"
echo "集群唯一ID：${CLUSTER_ID}"
printf '%s\\n' "${CLUSTER_ID}" > "${KAFKA_BOOTSTRAP_DIR}/cluster.id"
chmod 400 "${KAFKA_BOOTSTRAP_DIR}/cluster.id"

6.3 生成根 CA 证书（10 年有效期）

# 4096位RSA私钥
openssl genrsa -out "${KAFKA_CA_DIR}/ca.key" 4096
# 签发根证书
openssl req -x509 -new -sha256 \\
-key "${KAFKA_CA_DIR}/ca.key" \\
-days 3650 \\
-subj '/CN=kafka-private-ca' \\
-out "${KAFKA_CA_DIR}/ca.crt"
chmod 400 "${KAFKA_CA_DIR}/ca.key"
chmod 444 "${KAFKA_CA_DIR}/ca.crt"
# 打印证书指纹存档
openssl x509 -in "${KAFKA_CA_DIR}/ca.crt" -noout -subject -issuer -dates -fingerprint -sha256
# 操作提醒：离线备份ca.key，服务器不要长期留存

6.4 生成节点证书包（示例节点 1）

NODE_ID=1
NODE_IP="${NODE_1_IP}"
NODE_CN="kafka-node-1"
STAGE="$(mktemp -d)"
mkdir -p "${STAGE}/tls" "${STAGE}/secrets"

# 节点公私钥
openssl genrsa -out "${STAGE}/node.key" 3072
openssl req -new -key "${STAGE}/node.key" -subj "/CN=${NODE_CN}" -out "${STAGE}/node.csr"

# SAN扩展配置（支持IP/域名双向认证）
cat >"${STAGE}/extensions.cnf" <<EOF
subjectAltName=IP:${NODE_IP},DNS:${NODE_CN}
extendedKeyUsage=serverAuth,clientAuth
keyUsage=digitalSignature,keyEncipherment
EOF

# CA签发节点证书，825天有效期
openssl x509 -req -in "${STAGE}/node.csr" \\
-CA "${KAFKA_CA_DIR}/ca.crt" \\
-CAkey "${KAFKA_CA_DIR}/ca.key" \\
-CAcreateserial -days 825 -sha256 \\
-extfile "${STAGE}/extensions.cnf" \\
-out "${STAGE}/node.crt"

# 打包PKCS12密钥库
openssl pkcs12 -export -name "${NODE_CN}" \\
-inkey "${STAGE}/node.key" -in "${STAGE}/node.crt" \\
-certfile "${KAFKA_CA_DIR}/ca.crt" \\
-out "${STAGE}/tls/node.p12" \\
-passout "file:${TLS_PASSWORD_FILE}"

# 复制根证书、集群ID、密码文件
cp "${KAFKA_CA_DIR}/ca.crt" "${STAGE}/tls/ca.crt"
printf '%s\\n' "${CLUSTER_ID}" >"${STAGE}/secrets/cluster.id"
printf 'store_password=%s\\n' "${TLS_PASSWORD}" >"${STAGE}/secrets/tls-passwords.properties"
printf '%s\\n' "${NODE_ID}" >"${STAGE}/secrets/node.id"

# 容器生成信任库
docker run –rm –network none –user 0:0 –entrypoint /bin/sh \\
–mount "type=bind,src=${STAGE},dst=/work" \\
"${KAFKA_IMAGE}" -ec \\
'keytool -importcert -noprompt -alias kafka-private-ca \\
-file /work/tls/ca.crt \\
-keystore /work/tls/truststore.p12 \\
-storetype PKCS12 \\
-storepass "$(sed -n "s/^store_password=//p" /work/secrets/tls-passwords.properties)"'

# 打包校验压缩包
(
cd "${STAGE}"
find tls secrets -type f -print0 | sort -z | xargs -0 sha256sum > MANIFEST.sha256
tar -czf "${KAFKA_BUNDLE_DIR}/kafka-node-${NODE_ID}-security.tar.gz" \\
tls secrets MANIFEST.sha256
)
chmod 600 "${KAFKA_BUNDLE_DIR}/kafka-node-${NODE_ID}-security.tar.gz"
rm -rf "${STAGE}"

6.5 清理临时密码文件

ls -l "${KAFKA_BUNDLE_DIR}"/kafka-node-*-security.tar.gz
rm -f "${TLS_PASSWORD_FILE}"
unset TLS_PASSWORD

7 阶段 5：证书压缩包分发

# 节点1本地留存
cp -a "${KAFKA_BUNDLE_DIR}/kafka-node-1-security.tar.gz" /root/
# 远程推送节点2、3
scp "${KAFKA_BUNDLE_DIR}/kafka-node-2-security.tar.gz" root@${NODE_2_IP}:/root/
scp "${KAFKA_BUNDLE_DIR}/kafka-node-3-security.tar.gz" root@${NODE_3_IP}:/root/

8 阶段 6：各节点解压部署证书

以节点 1 举例，2/3 替换压缩包名称

# 节点1参数
BUNDLE="/root/kafka-node-1-security.tar.gz"
# 节点2：BUNDLE=/root/kafka-node-2-security.tar.gz
# 节点3：BUNDLE=/root/kafka-node-3-security.tar.gz

STAGE="$(mktemp -d)"
# 解压并校验哈希完整性
tar -xzf "${BUNDLE}" -C "${STAGE}"
( cd "${STAGE}" && sha256sum -c MANIFEST.sha256 )
# 核对本机节点编号
cat "${STAGE}/secrets/node.id"

# 创建证书目录
mkdir -p "${KAFKA_TLS_DIR}" "${KAFKA_SECRETS_DIR}"
cp -a "${STAGE}/tls/." "${KAFKA_TLS_DIR}/"
cp -a "${STAGE}/secrets/." "${KAFKA_SECRETS_DIR}"

# 修改归属容器用户
chown -R "${KAFKA_UID}:${KAFKA_GID}" "${KAFKA_TLS_DIR}" "${KAFKA_SECRETS_DIR}"
# 严格权限
chmod 500 "${KAFKA_TLS_DIR}" "${KAFKA_SECRETS_DIR}"
chmod 400 "${KAFKA_TLS_DIR}/node.p12" "${KAFKA_TLS_DIR}/truststore.p12" \\
"${KAFKA_SECRETS_DIR}/cluster.id" \\
"${KAFKA_SECRETS_DIR}/tls-passwords.properties" \\
"${KAFKA_SECRETS_DIR}/node.id"
chmod 444 "${KAFKA_TLS_DIR}/ca.crt"

# 清理临时文件
rm -rf "${STAGE}"
rm -f "${BUNDLE}"
# 打印CA指纹留存
openssl x509 -in "${KAFKA_TLS_DIR}/ca.crt" -noout -fingerprint -sha256

9 阶段 7：生成 server.properties + 集群格式化

9.1 定义本机节点标识（节点 1 示例）

NODE_ID=1
NODE_IP="${NODE_1_IP}"
CONTAINER_NAME="kafka-${NODE_ID}"
# 节点2：NODE_ID=2; NODE_IP="${NODE_2_IP}"; CONTAINER_NAME="kafka-2"
# 节点3：NODE_ID=3; NODE_IP="${NODE_3_IP}"; CONTAINER_NAME="kafka-3"

9.2 生成配置文件

# 创建持久化目录
mkdir -p "${KAFKA_DATA_DIR}" "${KAFKA_METADATA_DIR}" "${KAFKA_LOGS_DIR}" "${KAFKA_CONFIG_DIR}"
chown -R "${KAFKA_UID}:${KAFKA_GID}" \\
"${KAFKA_DATA_DIR}" "${KAFKA_METADATA_DIR}" "${KAFKA_LOGS_DIR}" "${KAFKA_CONFIG_DIR}"
chmod 750 "${KAFKA_DATA_DIR}" "${KAFKA_METADATA_DIR}" "${KAFKA_LOGS_DIR}" "${KAFKA_CONFIG_DIR}"

# 记录镜像哈希存档
docker image inspect –format '{{index .RepoDigests 0}}' "${KAFKA_IMAGE}" > "${KAFKA_ROOT}/image-digest.txt"

# 完整KRaft加密配置写入
cat > "${KAFKA_SERVER_PROPERTIES}" <<EOF
# KRaft双角色：broker+controller
process.roles=broker,controller
node.id=${NODE_ID}
controller.quorum.voters=${CONTROLLER_QUORUM_VOTERS}
controller.listener.names=CONTROLLER
inter.broker.listener.name=BROKER
listeners=CLIENT://0.0.0.0:${CLIENT_PORT},CONTROLLER://0.0.0.0:${CONTROLLER_PORT},BROKER://0.0.0.0:${BROKER_PORT}
advertised.listeners=CLIENT://${NODE_IP}:${CLIENT_PORT},BROKER://${NODE_IP}:${BROKER_PORT}
listener.security.protocol.map=CLIENT:SASL_SSL,CONTROLLER:SSL,BROKER:SSL
# 数据目录（容器内路径）
log.dirs=/var/lib/kafka/data
metadata.log.dir=/var/lib/kafka/metadata
# 三副本高可用默认配置
num.partitions=3
default.replication.factor=3
min.insync.replicas=2
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=2
unclean.leader.election.enable=false
auto.create.topics.enable=false
delete.topic.enable=true
# 消息7天过期
log.retention.hours=${KAFKA_RETENTION_HOURS}
log.retention.check.interval.ms=300000
log.segment.bytes=${KAFKA_SEGMENT_BYTES}
# 线程池
num.io.threads=8
num.network.threads=6
queued.max.requests=1000
# ACL权限控制
authorizer.class.name=org.apache.kafka.metadata.authorizer.StandardAuthorizer
allow.everyone.if.no.acl.found=false
super.users=User:${ADMIN_USER};User:kafka-node-1;User:kafka-node-2;User:kafka-node-3
sasl.enabled.mechanisms=${SCRAM_MECHANISM}
listener.name.client.sasl.enabled.mechanisms=${SCRAM_MECHANISM}
listener.name.client.scram-sha-512.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required;
# TLS安全协议限制
ssl.protocol=TLSv1.3
ssl.enabled.protocols=TLSv1.2,TLSv1.3
ssl.principal.mapping.rules=RULE:^CN=([^,]+).*/\\$1/,DEFAULT
config.providers=file
config.providers.file.class=org.apache.kafka.common.config.provider.FileConfigProvider

# 客户端单向TLS
listener.name.client.ssl.keystore.type=PKCS12
listener.name.client.ssl.keystore.location=/tmp/kafka-tls/node.p12
listener.name.client.ssl.keystore.password=\\${file:/tmp/kafka-secrets/tls-passwords.properties:store_password}
listener.name.client.ssl.key.password=\\${file:/tmp/kafka-secrets/tls-passwords.properties:store_password}
listener.name.client.ssl.truststore.type=PKCS12
listener.name.client.ssl.truststore.location=/tmp/kafka-tls/truststore.p12
listener.name.client.ssl.truststore.password=\\${file:/tmp/kafka-secrets/tls-passwords.properties:store_password}
listener.name.client.ssl.client.auth=none

# 控制器双向mTLS
listener.name.controller.ssl.keystore.type=PKCS12
listener.name.controller.ssl.keystore.location=/tmp/kafka-tls/node.p12
listener.name.controller.ssl.keystore.password=\\${file:/tmp/kafka-secrets/tls-passwords.properties:store_password}
listener.name.controller.ssl.key.password=\\${file:/tmp/kafka-secrets/tls-passwords.properties:store_password}
listener.name.controller.ssl.truststore.type=PKCS12
listener.name.controller.ssl.truststore.location=/tmp/kafka-tls/truststore.p12
listener.name.controller.ssl.truststore.password=\\${file:/tmp/kafka-secrets/tls-passwords.properties:store_password}
listener.name.controller.ssl.client.auth=required

# Broker副本双向mTLS
listener.name.broker.ssl.keystore.type=PKCS12
listener.name.broker.ssl.keystore.location=/tmp/kafka-tls/node.p12
listener.name.broker.ssl.keystore.password=\\${file:/tmp/kafka-secrets/tls-passwords.properties:store_password}
listener.name.broker.ssl.key.password=\\${file:/tmp/kafka-secrets/tls-passwords.properties:store_password}
listener.name.broker.ssl.truststore.type=PKCS12
listener.name.broker.ssl.truststore.location=/tmp/kafka-tls/truststore.p12
listener.name.broker.ssl.truststore.password=\\${file:/tmp/kafka-secrets/tls-passwords.properties:store_password}
listener.name.broker.ssl.client.auth=required
EOF
chown "${KAFKA_UID}:${KAFKA_GID}" "${KAFKA_SERVER_PROPERTIES}"
chmod 440 "${KAFKA_SERVER_PROPERTIES}"

9.3 集群格式化（仅全新空数据执行）

三台输入管理员密码必须完全一致

# 检测是否已格式化
ls "${KAFKA_DATA_DIR}/meta.properties" "${KAFKA_METADATA_DIR}/meta.properties" 2>/dev/null || echo "目录空白可格式化"

read -r -s -p '输入管理员密码：' ADMIN_PASSWORD; echo
read -r -s -p '确认密码：' ADMIN_PASSWORD2; echo
if [[ "${ADMIN_PASSWORD}" != "${ADMIN_PASSWORD2}" ]]; then
echo "密码不一致退出"; exit 1
fi
unset ADMIN_PASSWORD2

CLUSTER_ID="$(tr -d '[:space:]' <"${KAFKA_SECRETS_DIR}/cluster.id")"

# 执行格式化
docker run –rm –network host –user "${KAFKA_UID}:${KAFKA_GID}" \\
–mount "type=bind,src=${KAFKA_DATA_DIR},dst=/var/lib/kafka/data" \\
–mount "type=bind,src=${KAFKA_METADATA_DIR},dst=/var/lib/kafka/metadata" \\
–mount "type=bind,src=${KAFKA_LOGS_DIR},dst=/opt/kafka/logs" \\
–mount "type=bind,src=${KAFKA_SERVER_PROPERTIES},dst=/tmp/kafka-server.properties,readonly" \\
–mount "type=bind,src=${KAFKA_TLS_DIR},dst=/tmp/kafka-tls,readonly" \\
–mount "type=bind,src=${KAFKA_SECRETS_DIR},dst=/tmp/kafka-secrets,readonly" \\
"${KAFKA_IMAGE}" \\
/opt/kafka/bin/kafka-storage.sh format \\
–config /tmp/kafka-server.properties \\
–cluster-id "${CLUSTER_ID}" \\
–add-scram "${SCRAM_MECHANISM}=[name=${ADMIN_USER},password=${ADMIN_PASSWORD}]"
unset ADMIN_PASSWORD

10 阶段 8：启动 Kafka 容器

10.1 端口空闲校验

ss -ltn | grep -E ":${CLIENT_PORT}|:${CONTROLLER_PORT}|:${BROKER_PORT}" && { echo "端口占用，无法启动"; exit 1; }

10.2 后台启动容器

docker run -d \\
–name "${CONTAINER_NAME}" \\
–restart unless-stopped \\
–network host \\
–user "${KAFKA_UID}:${KAFKA_GID}" \\
–memory "${KAFKA_MEMORY}" \\
–memory-swap "${KAFKA_MEMORY}" \\
–cpus "${KAFKA_CPU_LIMIT}" \\
–ulimit "nofile=${KAFKA_NOFILE}:${KAFKA_NOFILE}" \\
–security-opt no-new-privileges:true \\
–log-driver local \\
–log-opt max-size=100m \\
–log-opt max-file=10 \\
-e "KAFKA_HEAP_OPTS=-Xms${KAFKA_HEAP} -Xmx${KAFKA_HEAP}" \\
–mount "type=bind,src=${KAFKA_DATA_DIR},dst=/var/lib/kafka/data" \\
–mount "type=bind,src=${KAFKA_METADATA_DIR},dst=/var/lib/kafka/metadata" \\
–mount "type=bind,src=${KAFKA_LOGS_DIR},dst=/opt/kafka/logs" \\
–mount "type=bind,src=${KAFKA_SERVER_PROPERTIES},dst=/tmp/kafka-server.properties,readonly" \\
–mount "type=bind,src=${KAFKA_TLS_DIR},dst=/tmp/kafka-tls,readonly" \\
–mount "type=bind,src=${KAFKA_SECRETS_DIR},dst=/tmp/kafka-secrets,readonly" \\
"${KAFKA_IMAGE}" \\
/opt/kafka/bin/kafka-server-start.sh /tmp/kafka-server.properties

10.3 启动校验

docker ps –filter "name=${CONTAINER_NAME}"
ss -ltn | grep -E ":${CLIENT_PORT}|:${CONTROLLER_PORT}"
docker logs –tail 100 "${CONTAINER_NAME}"

11 阶段 9：集群连通验收（任意一台执行）

# 生成管理员客户端配置
read -r -s -p '输入管理员密码：' ADMIN_PASSWORD; echo
write_admin_client_config "${ADMIN_PASSWORD}"
unset ADMIN_PASSWORD

# 1 集群连通校验
kafka_admin_cli /opt/kafka/bin/kafka-broker-api-versions.sh \\
–bootstrap-server "${KAFKA_BOOTSTRAP_SERVERS}" \\
–command-config /tmp/admin-client.properties | head

# 2 查看KRaft集群状态
kafka_admin_cli /opt/kafka/bin/kafka-metadata-quorum.sh \\
–bootstrap-server "${KAFKA_BOOTSTRAP_SERVERS}" \\
–command-config /tmp/admin-client.properties describe –status

# 3 创建测试Topic
TEST_TOPIC="kafka-test-$(date +%Y%m%d%H%M%S)"
kafka_admin_cli /opt/kafka/bin/kafka-topics.sh \\
–bootstrap-server "${KAFKA_BOOTSTRAP_SERVERS}" \\
–command-config /tmp/admin-client.properties \\
–create –topic "${TEST_TOPIC}" –partitions 3 –replication-factor 3

# 查看分区副本
kafka_admin_cli /opt/kafka/bin/kafka-topics.sh \\
–bootstrap-server "${KAFKA_BOOTSTRAP_SERVERS}" \\
–command-config /tmp/admin-client.properties \\
–describe –topic "${TEST_TOPIC}"

# 删除测试Topic
kafka_admin_cli /opt/kafka/bin/kafka-topics.sh \\
–bootstrap-server "${KAFKA_BOOTSTRAP_SERVERS}" \\
–command-config /tmp/admin-client.properties \\
–delete –topic "${TEST_TOPIC}"

# 清理临时文件
rm -f "${CLIENT_CONFIG}"
df -h /data

12 阶段 10：业务 Topic 创建模板（GBT32960 车辆报文）

TOPIC="GBT3296-v2025-NEV"
kafka_admin_cli /opt/kafka/bin/kafka-topics.sh \\
–bootstrap-server "${KAFKA_BOOTSTRAP_SERVERS}" \\
–command-config /tmp/admin-client.properties \\
–create –topic "${TOPIC}" \\
–partitions 3 –replication-factor 3 \\
–config "retention.ms=${KAFKA_RETENTION_MS}" \\
–config retention.bytes=21474836480 \\
–config "segment.bytes=${KAFKA_SEGMENT_BYTES}"

# 查看Topic配置
kafka_admin_cli /opt/kafka/bin/kafka-topics.sh \\
–bootstrap-server "${KAFKA_BOOTSTRAP_SERVERS}" \\
–command-config /tmp/admin-client.properties \\
–describe –topic "${TOPIC}"

13 阶段 11：SCRAM 账号 + ACL 权限

13.1 生产者账号（写入报文）

USERNAME="entty-producer"
read -r -s -p '输入账号密码：' USER_PASSWORD; echo
read -r -s -p '确认密码：' USER_PASSWORD2; echo
if [[ "${USER_PASSWORD}" != "${USER_PASSWORD2}" ]]; then exit 1; fi
unset USER_PASSWORD2

# 创建SCRAM用户
kafka_admin_cli /opt/kafka/bin/kafka-configs.sh \\
–bootstrap-server "${KAFKA_BOOTSTRAP_SERVERS}" \\
–command-config /tmp/admin-client.properties \\
–alter –entity-type users –entity-name "${USERNAME}" \\
–add-config "${SCRAM_MECHANISM}=[iterations=4096,password=${USER_PASSWORD}]"

# 授予前缀Topic读写权限
for PREFIX in GBT3296- iot-; do
for OP in Describe Write; do
kafka_admin_cli /opt/kafka/bin/kafka-acls.sh \\
–bootstrap-server "${KAFKA_BOOTSTRAP_SERVERS}" \\
–command-config /tmp/admin-client.properties –add \\
–allow-principal "User:${USERNAME}" \\
–operation "${OP}" \\
–topic "${PREFIX}" –resource-pattern-type prefixed
done
done
# 事务写入权限
kafka_admin_cli /opt/kafka/bin/kafka-acls.sh \\
–bootstrap-server "${KAFKA_BOOTSTRAP_SERVERS}" \\
–command-config /tmp/admin-client.properties –add \\
–allow-principal "User:${USERNAME}" \\
–operation IdempotentWrite –cluster
unset USER_PASSWORD

# 查看权限
kafka_admin_cli /opt/kafka/bin/kafka-acls.sh \\
–bootstrap-server "${KAFKA_BOOTSTRAP_SERVERS}" \\
–command-config /tmp/admin-client.properties \\
–list –principal "User:${USERNAME}"

13.2 Flink 消费者账号

USERNAME="flink-consumer"
# Topic读权限
for PREFIX in GBT3296- iot-; do
for OP in Describe Read; do
kafka_admin_cli /opt/kafka/bin/kafka-acls.sh \\
–bootstrap-server "${KAFKA_BOOTSTRAP_SERVERS}" \\
–command-config /tmp/admin-client.properties –add \\
–allow-principal "User:${USERNAME}" \\
–operation "${OP}" \\
–topic "${PREFIX}" –resource-pattern-type prefixed
done
done
# 消费者组权限
kafka_admin_cli /opt/kafka/bin/kafka-acls.sh \\
–bootstrap-server "${KAFKA_BOOTSTRAP_SERVERS}" \\
–command-config /tmp/admin-client.properties –add \\
–allow-principal "User:${USERNAME}" \\
–operation Read –group flink-consumer

14 阶段 12：部署 Kafka UI（仅节点 1）

14.1 创建 UI 只读账号

UI_USER="${KAFKA_UI_KAFKA_USER}"
read -r -s -p 'Kafka账号密码：' KAFKA_UI_KAFKA_PASSWORD; echo
read -r -s -p '网页登录密码：' KAFKA_UI_WEB_PASSWORD; echo

# 创建SCRAM账号
kafka_admin_cli /opt/kafka/bin/kafka-configs.sh \\
–bootstrap-server "${KAFKA_BOOTSTRAP_SERVERS}" \\
–command-config /tmp/admin-client.properties \\
–alter –entity-type users –entity-name "${UI_USER}" \\
–add-config "${SCRAM_MECHANISM}=[iterations=4096,password=${KAFKA_UI_KAFKA_PASSWORD}]"

# 集群只读权限
for OP in Describe DescribeConfigs; do
kafka_admin_cli /opt/kafka/bin/kafka-acls.sh \\
–bootstrap-server "${KAFKA_BOOTSTRAP_SERVERS}" \\
–command-config /tmp/admin-client.properties –add \\
–allow-principal "User:${UI_USER}" –operation "${OP}" –cluster
done
# 所有Topic只读
for OP in Read Describe DescribeConfigs; do
kafka_admin_cli /opt/kafka/bin/kafka-acls.sh \\
–bootstrap-server "${KAFKA_BOOTSTRAP_SERVERS}" \\
–command-config /tmp/admin-client.properties –add \\
–allow-principal "User:${UI_USER}" –operation "${OP}" –topic '*'
done
# 消费者组查看
kafka_admin_cli /opt/kafka/bin/kafka-acls.sh \\
–bootstrap-server "${KAFKA_BOOTSTRAP_SERVERS}" \\
–command-config /tmp/admin-client.properties –add \\
–allow-principal "User:${UI_USER}" –operation Describe –group '*'

# 写入UI密码环境文件
install -d -m 700 "${KAFKA_UI_SECRETS_DIR}"
umask 077
{
printf 'KAFKA_UI_KAFKA_PASSWORD=%q\\n' "${KAFKA_UI_KAFKA_PASSWORD}"
printf 'KAFKA_UI_WEB_PASSWORD=%q\\n' "${KAFKA_UI_WEB_PASSWORD}"
} > "${KAFKA_UI_ENV_FILE}"
chown root:root "${KAFKA_UI_ENV_FILE}"
chmod 600 "${KAFKA_UI_ENV_FILE}"
unset KAFKA_UI_KAFKA_PASSWORD KAFKA_UI_WEB_PASSWORD

14.2 防火墙开放 UI 端口

ZONE="$(firewall-cmd –get-default-zone)"
firewall-cmd –permanent –zone="${ZONE}" –add-port="${KAFKA_UI_HOST_PORT}/tcp"
firewall-cmd –reload

14.3 启动 UI 容器

set -a; source "${CLUSTER_ENV_FILE}"; source "${KAFKA_UI_ENV_FILE}"; set +a
JAAS="org.apache.kafka.common.security.scram.ScramLoginModule required username=\\"${KAFKA_UI_KAFKA_USER}\\" password=\\"${KAFKA_UI_KAFKA_PASSWORD}\\";"
docker run -d \\
–name "${KAFKA_UI_CONTAINER_NAME}" \\
–restart unless-stopped \\
–publish "${KAFKA_UI_BIND_ADDRESS}:${KAFKA_UI_HOST_PORT}:${KAFKA_UI_CONTAINER_PORT}" \\
–read-only \\
–tmpfs /tmp:rw,nosuid,nodev,size=128m \\
–cap-drop ALL \\
–security-opt no-new-privileges:true \\
–memory "${KAFKA_UI_MEMORY}" \\
–memory-swap "${KAFKA_UI_MEMORY}" \\
–cpus "${KAFKA_UI_CPU_LIMIT}" \\
–pids-limit "${KAFKA_UI_PIDS_LIMIT}" \\
–log-driver local \\
–log-opt max-size=50m \\
–log-opt max-file=5 \\
–health-cmd 'wget -q -O – http://127.0.0.1:8080/actuator/health | grep -q UP' \\
–health-interval 30s \\
–health-timeout 5s \\
–health-retries 5 \\
–health-start-period 60s \\
–mount "type=bind,src=${KAFKA_TLS_DIR}/ca.crt,dst=/etc/kafkaui/ca.crt,readonly" \\
-e "JAVA_OPTS=-Xms256m -Xmx${KAFKA_UI_HEAP} -Djava.io.tmpdir=/tmp" \\
-e AUTH_TYPE=LOGIN_FORM \\
-e "SPRING_SECURITY_USER_NAME=${KAFKA_UI_WEB_USER}" \\
-e "SPRING_SECURITY_USER_PASSWORD=${KAFKA_UI_WEB_PASSWORD}" \\
-e KAFKA_CLUSTERS_0_NAME=production \\
-e "KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS=${KAFKA_BOOTSTRAP_SERVERS}" \\
-e KAFKA_CLUSTERS_0_READONLY=true \\
-e KAFKA_CLUSTERS_0_PROPERTIES_SECURITY_PROTOCOL=SASL_SSL \\
-e "KAFKA_CLUSTERS_0_PROPERTIES_SASL_MECHANISM=${SCRAM_MECHANISM}" \\
-e "KAFKA_CLUSTERS_0_PROPERTIES_SASL_JAAS_CONFIG=${JAAS}" \\
-e KAFKA_CLUSTERS_0_PROPERTIES_SSL_TRUSTSTORE_TYPE=PEM \\
-e KAFKA_CLUSTERS_0_PROPERTIES_SSL_TRUSTSTORE_LOCATION=/etc/kafkaui/ca.crt \\
-e KAFKA_CLUSTERS_0_PROPERTIES_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM=https \\
-e DYNAMIC_CONFIG_ENABLED=false \\
-e SWAGGER_UI_ENABLED=false \\
"${KAFKA_UI_IMAGE}"
unset KAFKA_UI_KAFKA_PASSWORD KAFKA_UI_WEB_PASSWORD JA
# 校验
docker ps –filter "name=${KAFKA_UI_CONTAINER_NAME}"
curl –fail "http://127.0.0.1:${KAFKA_UI_HOST_PORT}"
# 访问地址 http://节点1公IP:10002

15 阶段 13：Flink/Java 业务客户端证书配置

15.1 生成客户端信任库

APP_CERT_DIR="/opt/pig-iot/certs"
SECURE_TRANSFER="/secure-transfer"
install -d -o root -g root -m 700 "${SECURE_TRANSFER}"
# 拉取根证书
scp "root@${NODE_1_IP}:${KAFKA_TLS_DIR}/ca.crt" "${SECURE_TRANSFER}/ca.crt"
chmod 600 "${SECURE_TRANSFER}/ca.crt"
install -d -o root -g root -m 750 "${APP_CERT_DIR}"
install -o root -g root -m 644 "${SECURE_TRANSFER}/ca.crt" "${APP_CERT_DIR}/ca.crt"
# 生成p12信任库
read -r -s -p '客户端信任库密码：' KAFKA_TRUSTSTORE_PASSWORD; echo
export KAFKA_TRUSTSTORE_PASSWORD
keytool -importcert -noprompt -trustcacerts \\
-alias kafka-private-ca \\
-file "${APP_CERT_DIR}/ca.crt" \\
-keystore "${APP_CERT_DIR}/kafka-client-truststore.p12" \\
-storetype PKCS12 \\
-storepass:env KAFKA_TRUSTSTORE_PASSWORD
chmod 640 "${APP_CERT_DIR}/kafka-client-truststore.p12"
unset KAFKA_TRUSTSTORE_PASSWORD
rm -rf "${SECURE_TRANSFER}"

15.2 业务配置参数示例 (properties)

kafka.bootstrap.servers=192.168.xxx:19092,192.168.xxx:19092,192.168.xxx:19092
sasl.mechanism=SCRAM-SHA-512
security.protocol=SASL_SSL
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="entty-producer" password="业务密码";
ssl.truststore.type=PKCS12
ssl.truststore.location=/opt/pig-iot/certs/kafka-client-truststore.p12
ssl.truststore.password=你的信任库密码
ssl.endpoint.identification.algorithm=https

16 阶段 14：日常运维命令

# 启停Broker
docker stop kafka-1
docker start kafka-1
# 实时日志
docker logs –tail 200 -f kafka-1
# 重启UI
docker restart kafka-ui
# 删除UI容器（配置保留）
docker rm -f kafka-ui
# 磁盘、端口
df -h /data
ss -ltn | grep 1909
# 重建Broker（数据不删，重新执行启动脚本）
docker rm -f kafka-1