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

这是一份基于前面排坑与实践沉淀的Kafka (KRaft 模式) SASL/SCRAM-SHA-512 安全认证的完整 Helm 部署教程。架构包含了声明式的用户管理、动态注册脚本、全流程对齐的 SCRAM 加密机制以及高可用存储配置。 教程目录项目目录结构完整配置文件values.yamltemplates/configmap.yamltemplates/secret.yamltemplates/statefulset.yamltemplates/service.yaml部署与生命周期管理验证与客户端接入1. 项目目录结构在 Helm Chart 根目录下按如下结构组织文件Plaintextagent-intent-kafka/ ├── Chart.yaml ├── values.yaml └── templates/ ├── configmap.yaml ├── secret.yaml ├── service.yaml └── statefulset.yaml2. 完整配置文件values.yaml用于集中配置集群镜像、端口、存储以及 SASL 用户密码。YAMLkafka: 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: ConsumerCKD6UDVah producer: ProducerCKD6UDVahtemplates/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.rolesbroker,controller controller.listener.namesCONTROLLER early.start.listenersCONTROLLER # --- 认证与授权配置 --- {{- if .Values.kafka.auth.enabled }} listenersPLAINTEXT://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.mapPLAINTEXT:PLAINTEXT,SASL_PLAINTEXT:SASL_PLAINTEXT,CONTROLLER:PLAINTEXT sasl.enabled.mechanismsSCRAM-SHA-512 sasl.mechanism.inter.broker.protocolPLAINTEXT authorizer.class.nameorg.apache.kafka.metadata.authorizer.StandardAuthorizer super.usersUser:admin;User:ANONYMOUS allow.everyone.if.no.acl.foundfalse {{- else }} listenersPLAINTEXT://0.0.0.0:{{ .Values.kafka.ports.plain }},CONTROLLER://0.0.0.0:{{ .Values.kafka.ports.controller }} listener.security.protocol.mapPLAINTEXT: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.threads16 num.network.threads8 num.partitions3 log.retention.hours168 {{- 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 usernameadmin password{{ $adminPassword }}; }; Client { org.apache.kafka.common.security.scram.ScramLoginModule required usernameadmin password{{ $adminPassword }}; }; client.properties: | security.protocolSASL_PLAINTEXT sasl.mechanismSCRAM-SHA-512 sasl.jaas.configorg.apache.kafka.common.security.scram.ScramLoginModule required usernameadmin 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.voters0localhost:{{ $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.listenersPLAINTEXT://${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.listenersPLAINTEXT://${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 将组件安装至命名空间Bashhelm upgrade --install agent-intent-kafka ./agent-intent-kafka \ -n agent-intent-system \ --create-namespace查看运行状态Bashkubectl get pods -n agent-intent-system -l app.kubernetes.io/componentkafka4. 验证与客户端接入 命令行测试容器内1. 查询 Topic 列表验证 SASL_PLAINTEXT 握手与鉴权Bashkubectl 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 \ --list2. 创建 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:9094Security Protocol:SASL_PLAINTEXTSASL Mechanism:SCRAM-SHA-512Username:adminPassword:admin-secret-pass