k8s helm部署kafka集群(KRaft模式)——筑梦之路

添加helm仓库

helm repo add bitnami "https://helm-charts.itboon.top/bitnami" --force-update
helm repo add grafana "https://helm-charts.itboon.top/grafana" --force-update
helm repo add prometheus-community "https://helm-charts.itboon.top/prometheus-community" --force-update
helm repo add ingress-nginx "https://helm-charts.itboon.top/ingress-nginx" --force-update
helm repo update

搜索kafka版本

# helm  search repo bitnami/kafka -l
NAME          	CHART VERSION	APP VERSION	DESCRIPTION
bitnami2/kafka	31.1.1       	3.9.0      	           
bitnami2/kafka	31.0.0       	3.9.0      	           
bitnami2/kafka	30.1.8       	3.8.1      	           
bitnami2/kafka	30.1.4       	3.8.0      	           
bitnami2/kafka	30.0.5       	3.8.0      	           
bitnami2/kafka	29.3.13      	3.7.1      	           
bitnami2/kafka	29.3.4       	3.7.0      	           
bitnami2/kafka	29.2.0       	3.7.0      	           
bitnami2/kafka	28.1.1       	3.7.0      	           
bitnami2/kafka	28.0.0       	3.7.0      	           
bitnami2/kafka	26.11.4      	3.6.1      	           
bitnami2/kafka	26.8.3       	3.6.1

编辑kafka.yaml

image:registry: docker.iorepository: bitnami/kafkatag: 3.9.0-debian-12-r4
listeners:client:protocol: PLAINTEXT #关闭访问认证controller:protocol: PLAINTEXT #关闭访问认证interbroker:protocol: PLAINTEXT #关闭访问认证external:protocol: PLAINTEXT #关闭访问认证
controller:replicaCount: 3 #副本数controllerOnly: false #controller+broker共用模式heapOpts: -Xmx4096m -Xms2048m #KAFKA JVMresources:limits:cpu: 4 memory: 8Girequests:cpu: 500mmemory: 512Miaffinity: #仅部署在master节点,不限制可删除nodeAffinity:requiredDuringSchedulingIgnoredDuringExecution:nodeSelectorTerms:- matchExpressions:- key: node-role.kubernetes.io/masteroperator: Exists- matchExpressions:- key: node-role.kubernetes.io/control-planeoperator: Existstolerations: #仅部署在master节点,不限制可删除- operator: Existseffect: NoSchedule- operator: Existseffect: NoExecutepersistence:storageClass: "local-path" #存储卷类型size: 10Gi #每个pod的存储大小
externalAccess:enabled: true #开启外部访问controller:service:type: NodePort #使用NodePort方式nodePorts:- 30091 #对外端口- 30092 #对外端口- 30093 #对外端口useHostIPs: true #使用宿主机IP

安装部署 

helm install kafka bitnami/kafka -f kafka.yaml --dry-runhelm install kafka bitnami-china/kafka -f kafka.yaml

内部访问

kafka-controller-headless.default:9092kafka-controller-0.kafka-controller-headless.default:9092
kafka-controller-1.kafka-controller-headless.default:9092
kafka-controller-2.kafka-controller-headless.default:9092

外部访问

# node ip +设置的nodeport端口,注意端口对应的节点的ip
192.168.100.110:30091  
192.168.100.111:30092  
192.168.100.112:30093# 从pod的配置中查找外部访问信息
kubectl exec -it kafka-controller-0 -- cat /opt/bitnami/kafka/config/server.properties | grep advertised.listeners

测试验证

# 创建一个podkubectl run kafka-client --restart='Never' --image bitnami/kafka:3.9.0-debian-12-r4 --namespace default --command -- sleep infinity# 进入pod生产消息
kubectl exec --tty -i kafka-client --namespace default -- bash
kafka-console-producer.sh \--broker-list kafka-controller-0.kafka-controller-headless.default.svc.cluster.local:9092,kafka-controller-1.kafka-controller-headless.default.svc.cluster.local:9092,kafka-controller-2.kafka-controller-headless.default.svc.cluster.local:9092 \--topic test# 进入pod消费消息
kubectl exec --tty -i kafka-client --namespace default -- bash
kafka-console-consumer.sh \--bootstrap-server kafka.default.svc.cluster.local:9092 \--topic test \--from-beginning

仅供参考

所有yaml文件

---
# Source: kafka/templates/networkpolicy.yaml
kind: NetworkPolicy
apiVersion: networking.k8s.io/v1
metadata:name: kafkanamespace: "default"labels:app.kubernetes.io/instance: kafkaapp.kubernetes.io/managed-by: Helmapp.kubernetes.io/name: kafkaapp.kubernetes.io/version: 3.9.0helm.sh/chart: kafka-31.1.1
spec:podSelector:matchLabels:app.kubernetes.io/instance: kafkaapp.kubernetes.io/name: kafkapolicyTypes:- Ingress- Egressegress:- {}ingress:# Allow client connections- ports:- port: 9092- port: 9094- port: 9093- port: 9095
---
# Source: kafka/templates/broker/pdb.yaml
apiVersion: policy/v1
kind: PodDisruptionBudget
metadata:name: kafka-brokernamespace: "default"labels:app.kubernetes.io/instance: kafkaapp.kubernetes.io/managed-by: Helmapp.kubernetes.io/name: kafkaapp.kubernetes.io/version: 3.9.0helm.sh/chart: kafka-31.1.1app.kubernetes.io/component: brokerapp.kubernetes.io/part-of: kafka
spec:maxUnavailable: 1selector:matchLabels:app.kubernetes.io/instance: kafkaapp.kubernetes.io/name: kafkaapp.kubernetes.io/component: brokerapp.kubernetes.io/part-of: kafka
---
# Source: kafka/templates/controller-eligible/pdb.yaml
apiVersion: policy/v1
kind: PodDisruptionBudget
metadata:name: kafka-controllernamespace: "default"labels:app.kubernetes.io/instance: kafkaapp.kubernetes.io/managed-by: Helmapp.kubernetes.io/name: kafkaapp.kubernetes.io/version: 3.9.0helm.sh/chart: kafka-31.1.1app.kubernetes.io/component: controller-eligibleapp.kubernetes.io/part-of: kafka
spec:maxUnavailable: 1selector:matchLabels:app.kubernetes.io/instance: kafkaapp.kubernetes.io/name: kafkaapp.kubernetes.io/component: controller-eligibleapp.kubernetes.io/part-of: kafka
---
# Source: kafka/templates/provisioning/serviceaccount.yaml
apiVersion: v1
kind: ServiceAccount
metadata:name: kafka-provisioningnamespace: "default"labels:app.kubernetes.io/instance: kafkaapp.kubernetes.io/managed-by: Helmapp.kubernetes.io/name: kafkaapp.kubernetes.io/version: 3.9.0helm.sh/chart: kafka-31.1.1
automountServiceAccountToken: false
---
# Source: kafka/templates/rbac/serviceaccount.yaml
apiVersion: v1
kind: ServiceAccount
metadata:name: kafkanamespace: "default"labels:app.kubernetes.io/instance: kafkaapp.kubernetes.io/managed-by: Helmapp.kubernetes.io/name: kafkaapp.kubernetes.io/version: 3.9.0helm.sh/chart: kafka-31.1.1app.kubernetes.io/component: kafka
automountServiceAccountToken: false
---
# Source: kafka/templates/secrets.yaml
apiVersion: v1
kind: Secret
metadata:name: kafka-kraft-cluster-idnamespace: "default"labels:app.kubernetes.io/instance: kafkaapp.kubernetes.io/managed-by: Helmapp.kubernetes.io/name: kafkaapp.kubernetes.io/version: 3.9.0helm.sh/chart: kafka-31.1.1
type: Opaque
data:kraft-cluster-id: "eDJrTHBicnVhQ1ZIUExEVU5BZVMxUA=="
---
# Source: kafka/templates/controller-eligible/configmap.yaml
apiVersion: v1
kind: ConfigMap
metadata:name: kafka-controller-configurationnamespace: "default"labels:app.kubernetes.io/instance: kafkaapp.kubernetes.io/managed-by: Helmapp.kubernetes.io/name: kafkaapp.kubernetes.io/version: 3.9.0helm.sh/chart: kafka-31.1.1app.kubernetes.io/component: controller-eligibleapp.kubernetes.io/part-of: kafka
data:server.properties: |-# Listeners configurationlisteners=CLIENT://:9092,INTERNAL://:9094,EXTERNAL://:9095,CONTROLLER://:9093advertised.listeners=CLIENT://advertised-address-placeholder:9092,INTERNAL://advertised-address-placeholder:9094listener.security.protocol.map=CLIENT:PLAINTEXT,INTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT# KRaft process rolesprocess.roles=controller,broker#node.id=controller.listener.names=CONTROLLERcontroller.quorum.voters=0@kafka-controller-0.kafka-controller-headless.default.svc.cluster.local:9093,1@kafka-controller-1.kafka-controller-headless.default.svc.cluster.local:9093,2@kafka-controller-2.kafka-controller-headless.default.svc.cluster.local:9093# Kafka data logs directorylog.dir=/bitnami/kafka/data# Kafka application logs directorylogs.dir=/opt/bitnami/kafka/logs# Common Kafka Configuration# Interbroker configurationinter.broker.listener.name=INTERNAL# Custom Kafka Configuration
---
# Source: kafka/templates/scripts-configmap.yaml
apiVersion: v1
kind: ConfigMap
metadata:name: kafka-scriptsnamespace: "default"labels:app.kubernetes.io/instance: kafkaapp.kubernetes.io/managed-by: Helmapp.kubernetes.io/name: kafkaapp.kubernetes.io/version: 3.9.0helm.sh/chart: kafka-31.1.1
data:kafka-init.sh: |-#!/bin/bashset -o errexitset -o nounsetset -o pipefailerror(){local message="${1:?missing message}"echo "ERROR: ${message}"exit 1}retry_while() {local -r cmd="${1:?cmd is missing}"local -r retries="${2:-12}"local -r sleep_time="${3:-5}"local return_value=1read -r -a command <<< "$cmd"for ((i = 1 ; i <= retries ; i+=1 )); do"${command[@]}" && return_value=0 && breaksleep "$sleep_time"donereturn $return_value}replace_in_file() {local filename="${1:?filename is required}"local match_regex="${2:?match regex is required}"local substitute_regex="${3:?substitute regex is required}"local posix_regex=${4:-true}local result# We should avoid using 'sed in-place' substitutions# 1) They are not compatible with files mounted from ConfigMap(s)# 2) We found incompatibility issues with Debian10 and "in-place" substitutionslocal -r del=$'\001' # Use a non-printable character as a 'sed' delimiter to avoid issuesif [[ $posix_regex = true ]]; thenresult="$(sed -E "s${del}${match_regex}${del}${substitute_regex}${del}g" "$filename")"elseresult="$(sed "s${del}${match_regex}${del}${substitute_regex}${del}g" "$filename")"fiecho "$result" > "$filename"}kafka_conf_set() {local file="${1:?missing file}"local key="${2:?missing key}"local value="${3:?missing value}"# Check if the value was set beforeif grep -q "^[#\\s]*$key\s*=.*" "$file"; then# Update the existing keyreplace_in_file "$file" "^[#\\s]*${key}\s*=.*" "${key}=${value}" falseelse# Add a new keyprintf '\n%s=%s' "$key" "$value" >>"$file"fi}replace_placeholder() {local placeholder="${1:?missing placeholder value}"local password="${2:?missing password value}"local -r del=$'\001' # Use a non-printable character as a 'sed' delimiter to avoid issues with delimiter symbols in sed stringsed -i "s${del}$placeholder${del}$password${del}g" "$KAFKA_CONFIG_FILE"}append_file_to_kafka_conf() {local file="${1:?missing source file}"local conf="${2:?missing kafka conf file}"cat "$1" >> "$2"}configure_external_access() {# Configure external hostnameif [[ -f "/shared/external-host.txt" ]]; thenhost=$(cat "/shared/external-host.txt")elif [[ -n "${EXTERNAL_ACCESS_HOST:-}" ]]; thenhost="$EXTERNAL_ACCESS_HOST"elif [[ -n "${EXTERNAL_ACCESS_HOSTS_LIST:-}" ]]; thenread -r -a hosts <<<"$(tr ',' ' ' <<<"${EXTERNAL_ACCESS_HOSTS_LIST}")"host="${hosts[$POD_ID]}"elif [[ "$EXTERNAL_ACCESS_HOST_USE_PUBLIC_IP" =~ ^(yes|true)$ ]]; thenhost=$(curl -s https://ipinfo.io/ip)elseerror "External access hostname not provided"fi# Configure external portif [[ -f "/shared/external-port.txt" ]]; thenport=$(cat "/shared/external-port.txt")elif [[ -n "${EXTERNAL_ACCESS_PORT:-}" ]]; thenif [[ "${EXTERNAL_ACCESS_PORT_AUTOINCREMENT:-}" =~ ^(yes|true)$ ]]; thenport="$((EXTERNAL_ACCESS_PORT + POD_ID))"elseport="$EXTERNAL_ACCESS_PORT"fielif [[ -n "${EXTERNAL_ACCESS_PORTS_LIST:-}" ]]; thenread -r -a ports <<<"$(tr ',' ' ' <<<"${EXTERNAL_ACCESS_PORTS_LIST}")"port="${ports[$POD_ID]}"elseerror "External access port not provided"fi# Configure Kafka advertised listenerssed -i -E "s|^(advertised\.listeners=\S+)$|\1,EXTERNAL://${host}:${port}|" "$KAFKA_CONFIG_FILE"}export KAFKA_CONFIG_FILE=/config/server.propertiescp /configmaps/server.properties $KAFKA_CONFIG_FILE# Get pod ID and role, last and second last fields in the pod name respectivelyPOD_ID=$(echo "$MY_POD_NAME" | rev | cut -d'-' -f 1 | rev)POD_ROLE=$(echo "$MY_POD_NAME" | rev | cut -d'-' -f 2 | rev)# Configure node.id and/or broker.idif [[ -f "/bitnami/kafka/data/meta.properties" ]]; thenif grep -q "broker.id" /bitnami/kafka/data/meta.properties; thenID="$(grep "broker.id" /bitnami/kafka/data/meta.properties | awk -F '=' '{print $2}')"kafka_conf_set "$KAFKA_CONFIG_FILE" "node.id" "$ID"elseID="$(grep "node.id" /bitnami/kafka/data/meta.properties | awk -F '=' '{print $2}')"kafka_conf_set "$KAFKA_CONFIG_FILE" "node.id" "$ID"fielseID=$((POD_ID + KAFKA_MIN_ID))kafka_conf_set "$KAFKA_CONFIG_FILE" "node.id" "$ID"fireplace_placeholder "advertised-address-placeholder" "${MY_POD_NAME}.kafka-${POD_ROLE}-headless.default.svc.cluster.local"if [[ "${EXTERNAL_ACCESS_ENABLED:-false}" =~ ^(yes|true)$ ]]; thenconfigure_external_accessfiif [ -f /secret-config/server-secret.properties ]; thenappend_file_to_kafka_conf /secret-config/server-secret.properties $KAFKA_CONFIG_FILEfi
---
# Source: kafka/templates/controller-eligible/svc-external-access.yaml
apiVersion: v1
kind: Service
metadata:name: kafka-controller-0-externalnamespace: "default"labels:app.kubernetes.io/instance: kafkaapp.kubernetes.io/managed-by: Helmapp.kubernetes.io/name: kafkaapp.kubernetes.io/version: 3.9.0helm.sh/chart: kafka-31.1.1app.kubernetes.io/component: kafkapod: kafka-controller-0
spec:type: NodePortpublishNotReadyAddresses: falseports:- name: tcp-kafkaport: 9094nodePort: 30091targetPort: externalselector:app.kubernetes.io/instance: kafkaapp.kubernetes.io/name: kafkaapp.kubernetes.io/part-of: kafkaapp.kubernetes.io/component: controller-eligiblestatefulset.kubernetes.io/pod-name: kafka-controller-0
---
# Source: kafka/templates/controller-eligible/svc-external-access.yaml
apiVersion: v1
kind: Service
metadata:name: kafka-controller-1-externalnamespace: "default"labels:app.kubernetes.io/instance: kafkaapp.kubernetes.io/managed-by: Helmapp.kubernetes.io/name: kafkaapp.kubernetes.io/version: 3.9.0helm.sh/chart: kafka-31.1.1app.kubernetes.io/component: kafkapod: kafka-controller-1
spec:type: NodePortpublishNotReadyAddresses: falseports:- name: tcp-kafkaport: 9094nodePort: 30092targetPort: externalselector:app.kubernetes.io/instance: kafkaapp.kubernetes.io/name: kafkaapp.kubernetes.io/part-of: kafkaapp.kubernetes.io/component: controller-eligiblestatefulset.kubernetes.io/pod-name: kafka-controller-1
---
# Source: kafka/templates/controller-eligible/svc-external-access.yaml
apiVersion: v1
kind: Service
metadata:name: kafka-controller-2-externalnamespace: "default"labels:app.kubernetes.io/instance: kafkaapp.kubernetes.io/managed-by: Helmapp.kubernetes.io/name: kafkaapp.kubernetes.io/version: 3.9.0helm.sh/chart: kafka-31.1.1app.kubernetes.io/component: kafkapod: kafka-controller-2
spec:type: NodePortpublishNotReadyAddresses: falseports:- name: tcp-kafkaport: 9094nodePort: 30093targetPort: externalselector:app.kubernetes.io/instance: kafkaapp.kubernetes.io/name: kafkaapp.kubernetes.io/part-of: kafkaapp.kubernetes.io/component: controller-eligiblestatefulset.kubernetes.io/pod-name: kafka-controller-2
---
# Source: kafka/templates/controller-eligible/svc-headless.yaml
apiVersion: v1
kind: Service
metadata:name: kafka-controller-headlessnamespace: "default"labels:app.kubernetes.io/instance: kafkaapp.kubernetes.io/managed-by: Helmapp.kubernetes.io/name: kafkaapp.kubernetes.io/version: 3.9.0helm.sh/chart: kafka-31.1.1app.kubernetes.io/component: controller-eligibleapp.kubernetes.io/part-of: kafka
spec:type: ClusterIPclusterIP: NonepublishNotReadyAddresses: trueports:- name: tcp-interbrokerport: 9094protocol: TCPtargetPort: interbroker- name: tcp-clientport: 9092protocol: TCPtargetPort: client- name: tcp-controllerprotocol: TCPport: 9093targetPort: controllerselector:app.kubernetes.io/instance: kafkaapp.kubernetes.io/name: kafkaapp.kubernetes.io/component: controller-eligibleapp.kubernetes.io/part-of: kafka
---
# Source: kafka/templates/svc.yaml
apiVersion: v1
kind: Service
metadata:name: kafkanamespace: "default"labels:app.kubernetes.io/instance: kafkaapp.kubernetes.io/managed-by: Helmapp.kubernetes.io/name: kafkaapp.kubernetes.io/version: 3.9.0helm.sh/chart: kafka-31.1.1app.kubernetes.io/component: kafka
spec:type: ClusterIPsessionAffinity: Noneports:- name: tcp-clientport: 9092protocol: TCPtargetPort: clientnodePort: null- name: tcp-externalport: 9095protocol: TCPtargetPort: externalselector:app.kubernetes.io/instance: kafkaapp.kubernetes.io/name: kafkaapp.kubernetes.io/part-of: kafka
---
# Source: kafka/templates/controller-eligible/statefulset.yaml
apiVersion: apps/v1
kind: StatefulSet
metadata:name: kafka-controllernamespace: "default"labels:app.kubernetes.io/instance: kafkaapp.kubernetes.io/managed-by: Helmapp.kubernetes.io/name: kafkaapp.kubernetes.io/version: 3.9.0helm.sh/chart: kafka-31.1.1app.kubernetes.io/component: controller-eligibleapp.kubernetes.io/part-of: kafka
spec:podManagementPolicy: Parallelreplicas: 3selector:matchLabels:app.kubernetes.io/instance: kafkaapp.kubernetes.io/name: kafkaapp.kubernetes.io/component: controller-eligibleapp.kubernetes.io/part-of: kafkaserviceName: kafka-controller-headlessupdateStrategy:type: RollingUpdatetemplate:metadata:labels:app.kubernetes.io/instance: kafkaapp.kubernetes.io/managed-by: Helmapp.kubernetes.io/name: kafkaapp.kubernetes.io/version: 3.9.0helm.sh/chart: kafka-31.1.1app.kubernetes.io/component: controller-eligibleapp.kubernetes.io/part-of: kafkaannotations:checksum/configuration: 84a30ef8698d80825ae7ffe45fae93a0d18c8861e2dfc64b4b809aa92065dfffspec:automountServiceAccountToken: falsehostNetwork: falsehostIPC: falseaffinity:podAffinity:podAntiAffinity:preferredDuringSchedulingIgnoredDuringExecution:- podAffinityTerm:labelSelector:matchLabels:app.kubernetes.io/instance: kafkaapp.kubernetes.io/name: kafkaapp.kubernetes.io/component: controller-eligibletopologyKey: kubernetes.io/hostnameweight: 1nodeAffinity:securityContext:fsGroup: 1001fsGroupChangePolicy: AlwaysseccompProfile:type: RuntimeDefaultsupplementalGroups: []sysctls: []serviceAccountName: kafkaenableServiceLinks: trueinitContainers:- name: kafka-initimage: docker.io/bitnami/kafka:3.9.0-debian-12-r4imagePullPolicy: IfNotPresentsecurityContext:allowPrivilegeEscalation: falsecapabilities:drop:- ALLreadOnlyRootFilesystem: truerunAsGroup: 1001runAsNonRoot: truerunAsUser: 1001seLinuxOptions: {}resources:limits: {}requests: {} command:- /bin/bashargs:- -ec- |/scripts/kafka-init.shenv:- name: BITNAMI_DEBUGvalue: "false"- name: MY_POD_NAMEvalueFrom:fieldRef:fieldPath: metadata.name- name: KAFKA_VOLUME_DIRvalue: "/bitnami/kafka"- name: KAFKA_MIN_IDvalue: "0"- name: EXTERNAL_ACCESS_ENABLEDvalue: "true"- name: HOST_IPvalueFrom:fieldRef:fieldPath: status.hostIP- name: EXTERNAL_ACCESS_HOSTvalue: "$(HOST_IP)"- name: EXTERNAL_ACCESS_PORTS_LISTvalue: "30091,30092,30093"volumeMounts:- name: datamountPath: /bitnami/kafka- name: kafka-configmountPath: /config- name: kafka-configmapsmountPath: /configmaps- name: kafka-secret-configmountPath: /secret-config- name: scriptsmountPath: /scripts- name: tmpmountPath: /tmpcontainers:- name: kafkaimage: docker.io/bitnami/kafka:3.9.0-debian-12-r4imagePullPolicy: "IfNotPresent"securityContext:allowPrivilegeEscalation: falsecapabilities:drop:- ALLreadOnlyRootFilesystem: truerunAsGroup: 1001runAsNonRoot: truerunAsUser: 1001seLinuxOptions: {}env:- name: BITNAMI_DEBUGvalue: "false"- name: KAFKA_HEAP_OPTSvalue: "-Xmx4096m -Xms2048m"- name: KAFKA_KRAFT_CLUSTER_IDvalueFrom:secretKeyRef:name: kafka-kraft-cluster-idkey: kraft-cluster-idports:- name: controllercontainerPort: 9093- name: clientcontainerPort: 9092- name: interbrokercontainerPort: 9094- name: externalcontainerPort: 9095livenessProbe:failureThreshold: 3initialDelaySeconds: 10periodSeconds: 10successThreshold: 1timeoutSeconds: 5exec:command:- pgrep- -f- kafkareadinessProbe:failureThreshold: 6initialDelaySeconds: 5periodSeconds: 10successThreshold: 1timeoutSeconds: 5tcpSocket:port: "controller"resources:limits:cpu: 4memory: 8Girequests:cpu: 500mmemory: 512MivolumeMounts:- name: datamountPath: /bitnami/kafka- name: logsmountPath: /opt/bitnami/kafka/logs- name: kafka-configmountPath: /opt/bitnami/kafka/config/server.propertiessubPath: server.properties- name: tmpmountPath: /tmpvolumes:- name: kafka-configmapsconfigMap:name: kafka-controller-configuration- name: kafka-secret-configemptyDir: {}- name: kafka-configemptyDir: {}- name: tmpemptyDir: {}- name: scriptsconfigMap:name: kafka-scriptsdefaultMode: 493- name: logsemptyDir: {}volumeClaimTemplates:- apiVersion: v1kind: PersistentVolumeClaimmetadata:name: dataspec:accessModes:- "ReadWriteOnce"resources:requests:storage: "10Gi"storageClassName: "local-path"