K8s 部署 Kafka (KRaft) + SASL/SCRAM-SHA-512 踩坑与终极实战指南

这是一份基于前面排坑与实践沉淀的 Kafka (KRaft 模式) + SASL/SCRAM-SHA-512 安全认证 的完整 Helm 部署教程。

架构包含了声明式的用户管理、动态注册脚本、全流程对齐的 SCRAM 加密机制以及高可用存储配置。

📖 教程目录

1. 项目目录结构

在 Helm Chart 根目录下按如下结构组织文件：

Plaintext

agent-intent-kafka/
├── Chart.yaml
├── values.yaml
└── templates/
├── configmap.yaml
├── secret.yaml
├── service.yaml
└── statefulset.yaml

2. 完整配置文件

📄 values.yaml

用于集中配置集群镜像、端口、存储以及 SASL 用户密码。

YAML

kafka:
enabled: true
replicaCount: 1
clusterId: "agent-intent-kafka-cluster-id"

image:
repository: apache/kafka
tag: 3.7.0
pullPolicy: IfNotPresent

ports:
plain: 9092
sasl: 9094
controller: 9093

storage: 10Gi
storageClass: "" # 根据实际集群填写，为空则使用默认 StorageClass

resources:
limits:
cpu: "2"
memory: 4Gi
requests:
cpu: "500m"
memory: 2Gi

auth:
enabled: true
users:
admin: "admin-secret-pass"
consumer: "Consumer@CKD6UDVah"
producer: "Producer@CKD6UDVah"

📄 templates/configmap.yaml

定义 Kafka KRaft 核心服务配置（包含角色分配、监听器与 StandardAuthorizer 鉴权类）。

YAML

{{- if .Values.kafka.enabled }}
apiVersion: v1
kind: ConfigMap
metadata:
name: {{ include "agent-intent.fullname" . }}-kafka-config
labels:
{{- include "agent-intent.labels" . | nindent 4 }}
app.kubernetes.io/component: kafka
data:
server.properties: |
# — KRaft 核心集群角色定义 —
process.roles=broker,controller
controller.listener.names=CONTROLLER
early.start.listeners=CONTROLLER

# — 认证与授权配置 —
{{- if .Values.kafka.auth.enabled }}
listeners=PLAINTEXT://0.0.0.0:{{ .Values.kafka.ports.plain }},SASL_PLAINTEXT://0.0.0.0:{{ .Values.kafka.ports.sasl }},CONTROLLER://0.0.0.0:{{ .Values.kafka.ports.controller }}
listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SASL_PLAINTEXT:SASL_PLAINTEXT,CONTROLLER:PLAINTEXT
sasl.enabled.mechanisms=SCRAM-SHA-512
sasl.mechanism.inter.broker.protocol=PLAINTEXT
authorizer.class.name=org.apache.kafka.metadata.authorizer.StandardAuthorizer
super.users=User:admin;User:ANONYMOUS
allow.everyone.if.no.acl.found=false
{{- else }}
listeners=PLAINTEXT://0.0.0.0:{{ .Values.kafka.ports.plain }},CONTROLLER://0.0.0.0:{{ .Values.kafka.ports.controller }}
listener.security.protocol.map=PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT
{{- end }}

# — 存储与数据目录 —
log.dirs=/var/lib/kafka/data

# — 副本因子配置 —
{{- $replicaCount := .Values.kafka.replicaCount | int }}
{{- $rf := $replicaCount }}
{{- $minIsr := $replicaCount | int }}
{{- if gt $minIsr 1 }}{{ $minIsr = sub $minIsr 1 }}{{ end }}
{{- if lt $minIsr 1 }}{{ $minIsr = 1 }}{{ end }}
default.replication.factor={{ $rf }}
offsets.topic.replication.factor={{ $rf }}
transaction.state.log.replication.factor={{ $rf }}
transaction.state.log.min.isr={{ $minIsr }}
min.insync.replicas={{ $minIsr }}

# — 性能调优 —
num.io.threads=16
num.network.threads=8
num.partitions=3
log.retention.hours=168
{{- end }}

📄 templates/secret.yaml

集中定义服务端/客户端 JAAS 配置文件，严格排除注释与特殊字符，对齐 ScramLoginModule。

YAML

{{- if .Values.kafka.enabled }}
apiVersion: v1
kind: Secret
metadata:
name: {{ include "agent-intent.fullname" . }}-kafka-auth
labels:
{{- include "agent-intent.labels" . | nindent 4 }}
app.kubernetes.io/component: kafka
type: Opaque
stringData:
cluster-id: "{{ .Values.kafka.clusterId | default (randAlphaNum 16) }}"

{{- if .Values.kafka.auth.enabled }}
{{- $adminPassword := index .Values.kafka.auth.users "admin" | default "admin-secret-pass" }}

kafka_server_jaas.conf: |
KafkaServer {
org.apache.kafka.common.security.scram.ScramLoginModule required
username="admin"
password="{{ $adminPassword }}";
};

Client {
org.apache.kafka.common.security.scram.ScramLoginModule required
username="admin"
password="{{ $adminPassword }}";
};

client.properties: |
security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="{{ $adminPassword }}";
{{- end }}
{{- end }}

📄 templates/statefulset.yaml

核心控制器，内置 (…) & 后台异步注册任务，在服务连通瞬间自动将 SCRAM 凭证写入 KRaft 元数据。

YAML

{{- if .Values.kafka.enabled }}
apiVersion: apps/v1
kind: StatefulSet
metadata:
name: {{ include "agent-intent.fullname" . }}-kafka
labels:
{{- include "agent-intent.labels" . | nindent 4 }}
app.kubernetes.io/component: kafka
spec:
serviceName: {{ include "agent-intent.fullname" . }}-kafka-headless
replicas: {{ .Values.kafka.replicaCount }}
podManagementPolicy: Parallel
selector:
matchLabels:
{{- include "agent-intent.selectorLabels" . | nindent 6 }}
app.kubernetes.io/component: kafka
template:
metadata:
labels:
{{- include "agent-intent.selectorLabels" . | nindent 8 }}
app.kubernetes.io/component: kafka
spec:
containers:
– name: kafka
image: "{{ .Values.kafka.image.repository }}:{{ .Values.kafka.image.tag }}"
imagePullPolicy: {{ .Values.kafka.image.pullPolicy }}
command:
– /bin/bash
– -ec
– |
ID=${HOSTNAME##*-}
CONFIG=/tmp/server.properties

# 复制只读模板到 /tmp 生成动态配置文件
cp /etc/kafka/server.properties.template ${CONFIG}
echo "node.id=${ID}" >> ${CONFIG}

{{- $replicaCount := .Values.kafka.replicaCount | int }}
{{- $fullname := include "agent-intent.fullname" . }}
{{- $namespace := .Release.Namespace }}
{{- $controllerPort := .Values.kafka.ports.controller }}

{{- if eq $replicaCount 1 }}
echo "controller.quorum.voters=0@localhost:{{ $controllerPort }}" >> ${CONFIG}
{{- else }}
echo "controller.quorum.voters={{- range $i := until $replicaCount }}{{ if gt $i 0 }},{{ end }}{{ $i }}@{{ $fullname }}-kafka-{{ $i }}.{{ $fullname }}-kafka-headless.{{ $namespace }}.svc.cluster.local:{{ $controllerPort }}{{- end }}" >> ${CONFIG}
{{- end }}

{{- if .Values.kafka.auth.enabled }}
export ADVERTISE_HOST="${HOSTNAME}.{{ include "agent-intent.fullname" . }}-kafka-headless.{{ .Release.Namespace }}.svc.cluster.local"
echo "advertised.listeners=PLAINTEXT://${ADVERTISE_HOST}:{{ .Values.kafka.ports.plain | default 9092 }},SASL_PLAINTEXT://${ADVERTISE_HOST}:{{ .Values.kafka.ports.sasl | default 9094 }}" >> ${CONFIG}
{{- else }}
export ADVERTISE_HOST="${HOSTNAME}.{{ include "agent-intent.fullname" . }}-kafka-headless.{{ .Release.Namespace }}.svc.cluster.local"
echo "advertised.listeners=PLAINTEXT://${ADVERTISE_HOST}:{{ .Values.kafka.ports.plain | default 9092 }}" >> ${CONFIG}
{{- end }}

if [ ! -f /var/lib/kafka/data/meta.properties ]; then
/opt/kafka/bin/kafka-storage.sh format \\
–ignore-formatted \\
-t "${CLUSTER_ID}" \\
-c ${CONFIG}
fi

{{- if .Values.kafka.auth.enabled }}
# 后台安全异步注入 SCRAM 凭证（轮询本地 9092 端口，连通即注入，不阻塞主流程）
(
while ! </dev/tcp/127.0.0.1/{{ .Values.kafka.ports.plain | default 9092 }}; do
sleep 0.5
done

{{- $adminPassword := index .Values.kafka.auth.users "admin" | default "admin-secret-pass" }}
/opt/kafka/bin/kafka-configs.sh –bootstrap-server 127.0.0.1:9092 \\
–entity-type users –entity-name admin –alter \\
–add-config 'SCRAM-SHA-512=[password={{ $adminPassword }}]' || true

{{- range $user, $pass := .Values.kafka.auth.users }}
{{- if ne $user "admin" }}
/opt/kafka/bin/kafka-configs.sh –bootstrap-server 127.0.0.1:9092 \\
–entity-type users –entity-name {{ $user }} –alter \\
–add-config 'SCRAM-SHA-512=[password={{ $pass }}]' || true
{{- end }}
{{- end }}
) &
{{- end }}

exec /opt/kafka/bin/kafka-server-start.sh ${CONFIG}

ports:
– containerPort: {{ .Values.kafka.ports.plain | default 9092 }}
name: plain-port
{{- if .Values.kafka.auth.enabled }}
– containerPort: {{ .Values.kafka.ports.sasl | default 9094 }}
name: sasl-port
{{- end }}

startupProbe:
exec:
command:
– /bin/bash
– -c
– "</dev/tcp/127.0.0.1/{{ .Values.kafka.ports.plain | default 9092 }}"
initialDelaySeconds: 1
periodSeconds: 2
timeoutSeconds: 2
failureThreshold: 30

livenessProbe:
exec:
command:
– /bin/bash
– -c
– "</dev/tcp/127.0.0.1/{{ .Values.kafka.ports.plain | default 9092 }}"
periodSeconds: 15
timeoutSeconds: 5
failureThreshold: 3

readinessProbe:
exec:
command:
– /bin/bash
– -c
– "</dev/tcp/127.0.0.1/{{ .Values.kafka.ports.plain | default 9092 }}"
initialDelaySeconds: 2
periodSeconds: 5
timeoutSeconds: 3
failureThreshold: 3

env:
– name: POD_NAMESPACE
valueFrom:
fieldRef:
fieldPath: metadata.namespace
– name: CLUSTER_ID
valueFrom:
secretKeyRef:
name: {{ include "agent-intent.fullname" . }}-kafka-auth
key: cluster-id
{{- if .Values.kafka.auth.enabled }}
– name: KAFKA_OPTS
value: "-Djava.security.auth.login.config=/etc/kafka/jaas/kafka_server_jaas.conf"
{{- end }}
– name: KAFKA_HEAP_OPTS
value: "-Xmx2G -Xms2G"

volumeMounts:
– name: kafka-data
mountPath: /var/lib/kafka/data
– name: kafka-config
mountPath: /etc/kafka/server.properties.template
subPath: server.properties
readOnly: true
{{- if .Values.kafka.auth.enabled }}
– name: kafka-jaas-config
mountPath: /etc/kafka/jaas/kafka_server_jaas.conf
subPath: kafka_server_jaas.conf
readOnly: true
– name: kafka-jaas-config
mountPath: /etc/kafka/client.properties
subPath: client.properties
readOnly: true
{{- end }}

resources:
{{- toYaml .Values.kafka.resources | nindent 10 }}

volumes:
– name: kafka-config
configMap:
name: {{ include "agent-intent.fullname" . }}-kafka-config
{{- if .Values.kafka.auth.enabled }}
– name: kafka-jaas-config
secret:
secretName: {{ include "agent-intent.fullname" . }}-kafka-auth
{{- end }}

volumeClaimTemplates:
– metadata:
name: kafka-data
spec:
accessModes: [ "ReadWriteOnce" ]
resources:
requests:
storage: {{ .Values.kafka.storage }}
{{- if .Values.kafka.storageClass }}
storageClassName: {{ .Values.kafka.storageClass }}
{{- end }}
{{- end }}

📄 templates/service.yaml

为外部暴露集群 Headless 服务，内部与外部认证端口隔离。

YAML

{{- if .Values.kafka.enabled }}
apiVersion: v1
kind: Service
metadata:
name: {{ include "agent-intent.fullname" . }}-kafka-headless
namespace: {{ .Release.Namespace }}
labels:
{{- include "agent-intent.labels" . | nindent 4 }}
app.kubernetes.io/component: kafka
app.kubernetes.io/part-of: kafka
spec:
clusterIP: None
publishNotReadyAddresses: true
ports:
– name: plaintext
port: {{ .Values.kafka.ports.plain | default 9092 }}
targetPort: {{ .Values.kafka.ports.plain | default 9092 }}
protocol: TCP
{{- if .Values.kafka.auth.enabled }}
– name: sasl-plaintext
port: {{ .Values.kafka.ports.sasl | default 9094 }}
targetPort: {{ .Values.kafka.ports.sasl | default 9094 }}
protocol: TCP
{{- end }}
– name: controller
port: {{ .Values.kafka.ports.controller | default 9093 }}
targetPort: {{ .Values.kafka.ports.controller | default 9093 }}
protocol: TCP
selector:
{{- include "agent-intent.selectorLabels" . | nindent 4 }}
app.kubernetes.io/component: kafka
{{- end }}

3. 部署与生命周期管理

部署 Chart

使用 Helm 将组件安装至命名空间：

Bash

helm upgrade –install agent-intent-kafka ./agent-intent-kafka \\
-n agent-intent-system \\
–create-namespace

查看运行状态

Bash

kubectl get pods -n agent-intent-system -l app.kubernetes.io/component=kafka

4. 验证与客户端接入

🔐 命令行测试（容器内）

1. 查询 Topic 列表（验证 SASL_PLAINTEXT 握手与鉴权）：

Bash

kubectl exec -it agent-intent-kafka-0 -n agent-intent-system — \\
/opt/kafka/bin/kafka-topics.sh \\
–bootstrap-server 127.0.0.1:9094 \\
–command-config /etc/kafka/client.properties \\
–list

2. 创建 Topic 与收发消息验证：

Bash

# 创建 Topic
kubectl exec -it agent-intent-kafka-0 -n agent-intent-system — \\
/opt/kafka/bin/kafka-topics.sh \\
–bootstrap-server 127.0.0.1:9094 \\
–command-config /etc/kafka/client.properties \\
–create –topic demo-topic –partitions 1 –replication-factor 1

# 写入测试消息
kubectl exec -it agent-intent-kafka-0 -n agent-intent-system — \\
bash -c 'echo "hello kafka scram sha 512" | /opt/kafka/bin/kafka-console-producer.sh \\
–bootstrap-server 127.0.0.1:9094 \\
–producer.config /etc/kafka/client.properties \\
–topic demo-topic'

# 消费测试消息
kubectl exec -it agent-intent-kafka-0 -n agent-intent-system — \\
/opt/kafka/bin/kafka-console-consumer.sh \\
–bootstrap-server 127.0.0.1:9094 \\
–consumer.config /etc/kafka/client.properties \\
–topic demo-topic \\
–from-beginning –max-messages 1

💻 业务微服务连接配置参考

跨集群或同 Namespace 微服务接入时，请在应用服务配置中增加如下参数（以 Python / Java 为例）：

• Bootstrap Server: agent-intent-kafka-0.agent-intent-kafka-headless.agent-intent-system.svc.cluster.local:9094

• Security Protocol: SASL_PLAINTEXT

• SASL Mechanism: SCRAM-SHA-512

• Username: admin

• Password: admin-secret-pass