yaml文件方式只需要对存储类名、名空间、节点IP和镜像地址进行调整就可以在k8s部署起来的kafka集群Kraft模式该yaml适用于bitnami版本镜像该yaml支持集群内外访问。--- apiVersion: v1 kind: Service metadata: name: kafka-headless namespace: test spec: clusterIP: None publishNotReadyAddresses: true selector: app: kafka ports: - name: plaintext port: 9092 targetPort: 9092 - name: controller port: 9093 targetPort: 9093 --- apiVersion: v1 kind: Service metadata: name: kafka-svc namespace: test spec: type: ClusterIP selector: statefulset.kubernetes.io/pod-name: kafka-0 ports: - protocol: TCP port: 9092 targetPort: 9092 --- apiVersion: v1 kind: Service metadata: name: kafka-0 namespace: test spec: type: NodePort selector: statefulset.kubernetes.io/pod-name: kafka-0 ports: - protocol: TCP port: 39092 targetPort: 9092 nodePort: 39092 --- apiVersion: v1 kind: Service metadata: name: kafka-1 namespace: test spec: type: NodePort selector: statefulset.kubernetes.io/pod-name: kafka-1 ports: - protocol: TCP port: 39093 targetPort: 9092 nodePort: 39093 --- apiVersion: v1 kind: Service metadata: name: kafka-2 namespace: test spec: type: NodePort selector: statefulset.kubernetes.io/pod-name: kafka-2 ports: - protocol: TCP port: 39094 targetPort: 9092 nodePort: 39094 --- apiVersion: apps/v1 kind: StatefulSet metadata: name: kafka namespace: test spec: selector: matchLabels: app: kafka serviceName: kafka-headless replicas: 3 updateStrategy: type: RollingUpdate podManagementPolicy: OrderedReady template: metadata: labels: app: kafka spec: affinity: podAntiAffinity: requiredDuringSchedulingIgnoredDuringExecution: - labelSelector: matchExpressions: - key: app operator: In values: [kafka] topologyKey: kubernetes.io/hostname securityContext: runAsUser: 1000 fsGroup: 1000 containers: - name: kafka image: kafka:3.9.0-debian-12-r13 imagePullPolicy: Always command: - bash - -ec - | HOSTNAMEhostname -s if [[ $HOSTNAME ~ (.*)-([0-9])$ ]]; then ORD${BASH_REMATCH[2]} PORT$((ORD 39092)) export KAFKA_CFG_NODE_ID$ORD export KAFKA_CFG_ADVERTISED_LISTENERSPLAINTEXT://192.168.100.100:$PORT else echo Failed to get index from hostname $HOSTNAME exit 1 fi exec /entrypoint.sh /run.sh env: - {name: ALLOW_PLAINTEXT_LISTENER, value: yes} - {name: KAFKA_KRAFT_CLUSTER_ID, value: MkU3OEVBNTcwNTJENDM2Qk} - {name: KAFKA_CFG_PROCESS_ROLES, value: broker,controller} - {name: KAFKA_CFG_LISTENERS, value: PLAINTEXT://:9092,CONTROLLER://:9093} - {name: KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP, value: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT} - {name: KAFKA_CFG_INTER_BROKER_LISTENER_NAME, value: PLAINTEXT} - {name: KAFKA_CFG_CONTROLLER_LISTENER_NAMES, value: CONTROLLER} - {name: KAFKA_CFG_CONTROLLER_QUORUM_VOTERS, value: 0kafka-0.kafka-headless.test.svc.cluster.local:9093,1kafka-1.kafka-headless.test.svc.cluster.local:9093,2kafka-2.kafka-headless.test.svc.cluster.local:9093} - {name: KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR, value: 3} - {name: KAFKA_CFG_TRANSACTION_STATE_LOG_MIN_ISR, value: 3} - {name: KAFKA_CFG_TRANSACTION_STATE_LOG_REPLICATION_FACTOR, value: 3} - {name: KAFKA_CFG_DEFAULT_REPLICATION_FACTOR, value: 3} - {name: TZ, value: Asia/Shanghai} ports: - {containerPort: 9092, name: plaintext} - {containerPort: 9093, name: controller} resources: limits: {cpu: 500m, memory: 1Gi} requests: {cpu: 100m, memory: 500Mi} volumeMounts: - {name: kafka-data, mountPath: /bitnami/kafka} volumeClaimTemplates: - metadata: {name: kafka-data} spec: accessModes: [ReadWriteOnce] storageClassName: managed-nfs-storage resources: requests: {storage: 1000Gi}仅供参考。官方kafka版本镜像apache/kakfa# # Apache Kafka KRaft 模式 · Kubernetes StatefulSet 部署Combined 模式 / 3 节点 # 镜像apache/kafka:4.3.1Apache 官方 JVM 镜像自 3.7.0 起提供 # # 拓扑3 个 Pod 同时承担 broker controller 角色可容忍 1 个节点故障。 # 适用中小规模生产、测试/预发环境、希望用最少组件跑通 KRaft 的场景。 # 用法kubectl apply -f kafka-kraft-combined.yaml # # 部署前必须确认/修改 # 1) CLUSTER_ID —— 换成自己生成的集群 ID见 README「部署步骤」第 1 步 # 2) storageClassName —— 换成集群中实际存在的 StorageClass # 3) 存储容量、CPU/内存 requests/limits 按实际环境调整 # # kafka-kraft-combined.yaml --- apiVersion: v1 kind: Namespace metadata: name: kafka labels: app.kubernetes.io/name: kafka --- # 无头 Service为每个 Pod 提供稳定 DNS # kafka-0.kafka-headless.kafka.svc.cluster.local # kafka-1.kafka-headless.kafka.svc.cluster.local # kafka-2.kafka-headless.kafka.svc.cluster.local # publishNotReadyAddresses 必须为 trueKRaft 组建 quorum 时 Pod 之间需要互相解析 # 若等到 Ready 才注册 DNS会因互相解析不到而永远无法 Ready死锁。 apiVersion: v1 kind: Service metadata: name: kafka-headless namespace: kafka labels: app.kubernetes.io/name: kafka spec: clusterIP: None publishNotReadyAddresses: true selector: app.kubernetes.io/name: kafka ports: - name: plaintext port: 9092 targetPort: 9092 protocol: TCP - name: controller port: 9093 targetPort: 9093 protocol: TCP --- # 普通 ClusterIP Service给集群内客户端做 bootstrap 入口 # 客户端使用 kafka-bootstrap.kafka.svc.cluster.local:9092 apiVersion: v1 kind: Service metadata: name: kafka-bootstrap namespace: kafka labels: app.kubernetes.io/name: kafka spec: type: ClusterIP selector: app.kubernetes.io/name: kafka ports: - name: plaintext port: 9092 targetPort: 9092 protocol: TCP --- apiVersion: apps/v1 kind: StatefulSet metadata: name: kafka namespace: kafka labels: app.kubernetes.io/name: kafka spec: serviceName: kafka-headless replicas: 3 # 关键Parallel。KRaft 的 quorum 需要多个 Pod 同时起来 # 若用默认的 OrderedReadykafka-0 会因 quorum 不完整而无法 Ready # 导致 kafka-1/kafka-2 永远不被创建。 podManagementPolicy: Parallel updateStrategy: type: RollingUpdate selector: matchLabels: app.kubernetes.io/name: kafka template: metadata: labels: app.kubernetes.io/name: kafka spec: # Kafka 关停需要时间做数据落盘留足优雅退出时间 terminationGracePeriodSeconds: 60 securityContext: runAsUser: 1000 # 官方镜像内建 appuseruid1000, gid1000 runAsGroup: 1000 runAsNonRoot: true fsGroup: 1000 # 让 PVC 挂载后对 appuser 可写缺了会报权限错误 # 尽量把 3 个副本打散到不同节点软约束节点不足时仍可调度 affinity: podAntiAffinity: preferredDuringSchedulingIgnoredDuringExecution: - weight: 100 podAffinityTerm: topologyKey: kubernetes.io/hostname labelSelector: matchLabels: app.kubernetes.io/name: kafka containers: - name: kafka image: apache/kafka:4.3.1 imagePullPolicy: IfNotPresent # 官方镜像的启动命令是 /etc/kafka/docker/runCMD非 ENTRYPOINT可被覆盖。 # 这里包一层 bash用 Pod 名推导本节点的 node.id 与 advertised.listeners # 因为这两个值每个 Pod 都不同无法用静态 env 写死。 # 注意${POD_NAME##*-} 是运行时由 bash 解析的Kubernetes 只替换 $(VAR) 形式不会误伤。 command: - bash - -c - | set -euo pipefail export KAFKA_NODE_ID${POD_NAME##*-} export KAFKA_ADVERTISED_LISTENERSPLAINTEXT://${POD_NAME}.kafka-headless.kafka.svc.cluster.local:9092 exec /etc/kafka/docker/run ports: - name: plaintext containerPort: 9092 - name: controller containerPort: 9093 env: - name: POD_NAME valueFrom: fieldRef: fieldPath: metadata.name # ---- 集群标识 ---------------------------------------------------- # 示例值务必替换为自己生成的集群 IDkafka-storage.sh random-uuid。 # 同一集群所有节点必须一致集群 ID 只在首次格式化数据目录时写入 # 之后想更换必须清空数据卷重新格式化。 - name: CLUSTER_ID value: 4L6g3nShT-eMCtK--X86sw # ---- KRaft 角色与监听 -------------------------------------------- - name: KAFKA_PROCESS_ROLES value: broker,controller - name: KAFKA_LISTENERS value: PLAINTEXT://:9092,CONTROLLER://:9093 - name: KAFKA_LISTENER_SECURITY_PROTOCOL_MAP value: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT - name: KAFKA_INTER_BROKER_LISTENER_NAME value: PLAINTEXT - name: KAFKA_CONTROLLER_LISTENER_NAMES value: CONTROLLER # 控制器 quorum 成员nodeIdPod 的稳定 DNS - name: KAFKA_CONTROLLER_QUORUM_VOTERS value: 0kafka-0.kafka-headless.kafka.svc.cluster.local:9093,1kafka-1.kafka-headless.kafka.svc.cluster.local:9093,2kafka-2.kafka-headless.kafka.svc.cluster.local:9093 # ---- 存储 -------------------------------------------------------- - name: KAFKA_LOG_DIRS value: /var/lib/kafka/data # ---- 副本因子单节点测试环境要改成 1 ----------------------------- - name: KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR value: 3 - name: KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR value: 3 - name: KAFKA_TRANSACTION_STATE_LOG_MIN_ISR value: 2 # Kafka 4.x 新增的 share group队列协调器内部主题 - name: KAFKA_SHARE_COORDINATOR_STATE_TOPIC_REPLICATION_FACTOR value: 3 - name: KAFKA_SHARE_COORDINATOR_STATE_TOPIC_MIN_ISR value: 2 - name: KAFKA_DEFAULT_REPLICATION_FACTOR value: 3 - name: KAFKA_MIN_INSYNC_REPLICAS value: 2 - name: KAFKA_NUM_PARTITIONS value: 3 # ---- 常规参数 ---------------------------------------------------- - name: KAFKA_AUTO_CREATE_TOPICS_ENABLE value: false - name: KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS value: 0 - name: KAFKA_LOG_RETENTION_HOURS value: 168 # JVM 堆按容器内存 limits 的 1/2 左右设置留出页缓存和直接内存 - name: KAFKA_HEAP_OPTS value: -Xms1g -Xmx1g # 官方镜像里 bin 脚本齐全用 broker API 探测更贴近真的能服务请求 # 比单纯 TCP 端口探测更可靠。 startupProbe: exec: command: - /bin/sh - -c - /opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092 /dev/null 21 initialDelaySeconds: 20 periodSeconds: 10 timeoutSeconds: 10 failureThreshold: 30 readinessProbe: exec: command: - /bin/sh - -c - /opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092 /dev/null 21 periodSeconds: 10 timeoutSeconds: 10 failureThreshold: 6 # 刻意不配 livenessProbeKafka 启动慢且重启不能自愈多为 quorum/存储问题 # 探活失败重启只会造成滚动失败循环靠 startupProbe readinessProbe 已足够。 resources: requests: cpu: 1 memory: 2Gi limits: cpu: 2 memory: 4Gi volumeMounts: - name: data mountPath: /var/lib/kafka/data volumeClaimTemplates: - metadata: name: data spec: accessModes: - ReadWriteOnce storageClassName: standard # TODO 改成集群实际的 StorageClass resources: requests: storage: 20Gi --- # 滚动升级/节点维护时保证最多只有 1 个 Kafka Pod 不可用 apiVersion: policy/v1 kind: PodDisruptionBudget metadata: name: kafka namespace: kafka spec: maxUnavailable: 1 selector: matchLabels: app.kubernetes.io/name: kafka# # 可选把 Kafka 暴露到集群外NodePort 方案 # 镜像apache/kafka:4.3.1Apache 官方 JVM 镜像 # # 原理客户端连接 broker 后会拿到 broker 自报的 advertised.listeners 地址并直连 # 所以每个 broker 必须有自己的对外入口不能共用一个 Service/NodePort # 否则客户端会被转发到错误的 broker。因此这里给每个 Pod 建一个独立 Service # 用 selector: statefulset.kubernetes.io/pod-name 精确绑定到单个 Pod # 并固定 nodePort使地址可预测。 # # 使用前需要同时修改 StatefulSetkafka-kraft-combined.yaml # # 1) 增加一个 EXTERNAL 监听端口与安全协议映射 # - name: KAFKA_LISTENERS # value: PLAINTEXT://:9092,CONTROLLER://:9093,EXTERNAL://:9094 # - name: KAFKA_LISTENER_SECURITY_PROTOCOL_MAP # value: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,EXTERNAL:PLAINTEXT # # 2) 增加容器端口 9094containerPort: 9094, name: external # # 3) 增加下行 API 环境变量拿到 Pod 所在节点 IP # - name: NODE_IP # valueFrom: # fieldRef: # fieldPath: status.hostIP # # 4) 改写 command 中的 advertised.listeners把 EXTERNAL 指向节点 IP 固定 nodePort # nodePort 与下方 Service 的 30092/30093/30094 一一对应 # export KAFKA_ADVERTISED_LISTENERSPLAINTEXT://${POD_NAME}.kafka-headless.kafka.svc.cluster.local:9092,EXTERNAL://${NODE_IP}:$((30092 ${POD_NAME##*-})) # # 5) 客户端 bootstrap 用 任一节点IP:30092,另一节点IP:30093,第三节点IP:30094 # 若集群前面有负载均衡/LB建议把三个 nodePort 都注册到同一个对外地址上。 # # 安全提醒PLAINTEXT 暴露到公网等于无认证无加密生产环境务必改用 # SASL_SSL / SSL 监听镜像支持挂载 /etc/kafka/secrets 下的证书与 JAAS 配置。 # --- apiVersion: v1 kind: Service metadata: name: kafka-external-0 namespace: kafka labels: app.kubernetes.io/name: kafka app.kubernetes.io/component: external spec: type: NodePort selector: statefulset.kubernetes.io/pod-name: kafka-0 ports: - name: external port: 9094 targetPort: 9094 nodePort: 30092 protocol: TCP --- apiVersion: v1 kind: Service metadata: name: kafka-external-1 namespace: kafka labels: app.kubernetes.io/name: kafka app.kubernetes.io/component: external spec: type: NodePort selector: statefulset.kubernetes.io/pod-name: kafka-1 ports: - name: external port: 9094 targetPort: 9094 nodePort: 30093 protocol: TCP --- apiVersion: v1 kind: Service metadata: name: kafka-external-2 namespace: kafka labels: app.kubernetes.io/name: kafka app.kubernetes.io/component: external spec: type: NodePort selector: statefulset.kubernetes.io/pod-name: kafka-2 ports: - name: external port: 9094 targetPort: 9094 nodePort: 30094 protocol: TCP# # Apache Kafka KRaft 模式 · Kubernetes StatefulSet 部署Isolated 模式 # 镜像apache/kafka:4.3.1Apache 官方 JVM 镜像 # # 拓扑3 个专用 controller 3 个专用 broker角色分离共 6 个 Pod。 # 适用生产环境。controller 只跑元数据共识不受 broker 流量影响职责清晰、可独立扩缩容。 # 用法kubectl apply -f kafka-kraft-isolated.yaml # 与 kafka-kraft-combined.yaml 二选一不要同时部署 # # 部署前必须确认/修改 # 1) CLUSTER_ID、2) storageClassName、3) 容量与资源配置 # # --- apiVersion: v1 kind: Namespace metadata: name: kafka labels: app.kubernetes.io/name: kafka # Controller 层 --- apiVersion: v1 kind: Service metadata: name: kafka-controller-headless namespace: kafka labels: app.kubernetes.io/name: kafka app.kubernetes.io/component: controller spec: clusterIP: None publishNotReadyAddresses: true selector: app.kubernetes.io/name: kafka app.kubernetes.io/component: controller ports: - name: controller port: 9093 targetPort: 9093 protocol: TCP --- apiVersion: apps/v1 kind: StatefulSet metadata: name: kafka-controller namespace: kafka labels: app.kubernetes.io/name: kafka app.kubernetes.io/component: controller spec: serviceName: kafka-controller-headless replicas: 3 podManagementPolicy: Parallel # quorum 组建要求多个 Pod 同时启动 selector: matchLabels: app.kubernetes.io/name: kafka app.kubernetes.io/component: controller template: metadata: labels: app.kubernetes.io/name: kafka app.kubernetes.io/component: controller spec: terminationGracePeriodSeconds: 60 securityContext: runAsUser: 1000 runAsGroup: 1000 runAsNonRoot: true fsGroup: 1000 affinity: podAntiAffinity: preferredDuringSchedulingIgnoredDuringExecution: - weight: 100 podAffinityTerm: topologyKey: kubernetes.io/hostname labelSelector: matchLabels: app.kubernetes.io/name: kafka app.kubernetes.io/component: controller containers: - name: kafka image: apache/kafka:4.3.1 imagePullPolicy: IfNotPresent # controller 节点只需要 node.id注意 controller-only 节点严禁设置 # KAFKA_ADVERTISED_LISTENERS镜像启动脚本会直接报错退出。 command: - bash - -c - | set -euo pipefail export KAFKA_NODE_ID${POD_NAME##*-} exec /etc/kafka/docker/run ports: - name: controller containerPort: 9093 env: - name: POD_NAME valueFrom: fieldRef: fieldPath: metadata.name - name: CLUSTER_ID value: 4L6g3nShT-eMCtK--X86sw # TODO 替换为自己生成的集群 ID - name: KAFKA_PROCESS_ROLES value: controller - name: KAFKA_LISTENERS value: CONTROLLER://:9093 - name: KAFKA_LISTENER_SECURITY_PROTOCOL_MAP value: CONTROLLER:PLAINTEXT - name: KAFKA_CONTROLLER_LISTENER_NAMES value: CONTROLLER - name: KAFKA_CONTROLLER_QUORUM_VOTERS value: 0kafka-controller-0.kafka-controller-headless.kafka.svc.cluster.local:9093,1kafka-controller-1.kafka-controller-headless.kafka.svc.cluster.local:9093,2kafka-controller-2.kafka-controller-headless.kafka.svc.cluster.local:9093 - name: KAFKA_LOG_DIRS value: /var/lib/kafka/data # controller 也承载元数据日志保留策略可略短于业务数据 - name: KAFKA_LOG_RETENTION_HOURS value: 168 - name: KAFKA_HEAP_OPTS value: -Xms1g -Xmx1g # controller 节点不提供 broker 端口用 TCP 探测其 CONTROLLER 监听端口 # 更精确的 quorum 状态可用 kafka-metadata-quorum.sh 人工检查见部署手册「验证」。 startupProbe: tcpSocket: port: 9093 initialDelaySeconds: 20 periodSeconds: 10 timeoutSeconds: 5 failureThreshold: 30 readinessProbe: tcpSocket: port: 9093 periodSeconds: 15 timeoutSeconds: 5 failureThreshold: 6 resources: requests: cpu: 500m memory: 2Gi limits: cpu: 1 memory: 4Gi volumeMounts: - name: data mountPath: /var/lib/kafka/data volumeClaimTemplates: - metadata: name: data spec: accessModes: - ReadWriteOnce storageClassName: standard # TODO 改成集群实际的 StorageClass resources: requests: storage: 10Gi # Broker 层 --- apiVersion: v1 kind: Service metadata: name: kafka-broker-headless namespace: kafka labels: app.kubernetes.io/name: kafka app.kubernetes.io/component: broker spec: clusterIP: None publishNotReadyAddresses: true selector: app.kubernetes.io/name: kafka app.kubernetes.io/component: broker ports: - name: plaintext port: 9092 targetPort: 9092 protocol: TCP --- # 客户端 bootstrap 入口kafka-bootstrap.kafka.svc.cluster.local:9092 apiVersion: v1 kind: Service metadata: name: kafka-bootstrap namespace: kafka labels: app.kubernetes.io/name: kafka app.kubernetes.io/component: broker spec: type: ClusterIP selector: app.kubernetes.io/name: kafka app.kubernetes.io/component: broker ports: - name: plaintext port: 9092 targetPort: 9092 protocol: TCP --- apiVersion: apps/v1 kind: StatefulSet metadata: name: kafka-broker namespace: kafka labels: app.kubernetes.io/name: kafka app.kubernetes.io/component: broker spec: serviceName: kafka-broker-headless replicas: 3 podManagementPolicy: Parallel selector: matchLabels: app.kubernetes.io/name: kafka app.kubernetes.io/component: broker template: metadata: labels: app.kubernetes.io/name: kafka app.kubernetes.io/component: broker spec: terminationGracePeriodSeconds: 60 securityContext: runAsUser: 1000 runAsGroup: 1000 runAsNonRoot: true fsGroup: 1000 affinity: podAntiAffinity: preferredDuringSchedulingIgnoredDuringExecution: - weight: 100 podAffinityTerm: topologyKey: kubernetes.io/hostname labelSelector: matchLabels: app.kubernetes.io/name: kafka app.kubernetes.io/component: broker containers: - name: kafka image: apache/kafka:4.3.1 imagePullPolicy: IfNotPresent # broker 的 node.id 从 3 开始避免与 controller 的 0/1/2 冲突 command: - bash - -c - | set -euo pipefail export KAFKA_NODE_ID$(( ${POD_NAME##*-} 3 )) export KAFKA_ADVERTISED_LISTENERSPLAINTEXT://${POD_NAME}.kafka-broker-headless.kafka.svc.cluster.local:9092 exec /etc/kafka/docker/run ports: - name: plaintext containerPort: 9092 env: - name: POD_NAME valueFrom: fieldRef: fieldPath: metadata.name - name: CLUSTER_ID value: 4L6g3nShT-eMCtK--X86sw # TODO 与 controller 保持一致 - name: KAFKA_PROCESS_ROLES value: broker - name: KAFKA_LISTENERS value: PLAINTEXT://:9092 - name: KAFKA_LISTENER_SECURITY_PROTOCOL_MAP value: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT - name: KAFKA_INTER_BROKER_LISTENER_NAME value: PLAINTEXT - name: KAFKA_CONTROLLER_LISTENER_NAMES value: CONTROLLER - name: KAFKA_CONTROLLER_QUORUM_VOTERS value: 0kafka-controller-0.kafka-controller-headless.kafka.svc.cluster.local:9093,1kafka-controller-1.kafka-controller-headless.kafka.svc.cluster.local:9093,2kafka-controller-2.kafka-controller-headless.kafka.svc.cluster.local:9093 - name: KAFKA_LOG_DIRS value: /var/lib/kafka/data - name: KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR value: 3 - name: KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR value: 3 - name: KAFKA_TRANSACTION_STATE_LOG_MIN_ISR value: 2 - name: KAFKA_SHARE_COORDINATOR_STATE_TOPIC_REPLICATION_FACTOR value: 3 - name: KAFKA_SHARE_COORDINATOR_STATE_TOPIC_MIN_ISR value: 2 - name: KAFKA_DEFAULT_REPLICATION_FACTOR value: 3 - name: KAFKA_MIN_INSYNC_REPLICAS value: 2 - name: KAFKA_NUM_PARTITIONS value: 3 - name: KAFKA_AUTO_CREATE_TOPICS_ENABLE value: false - name: KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS value: 0 - name: KAFKA_LOG_RETENTION_HOURS value: 168 - name: KAFKA_HEAP_OPTS value: -Xms2g -Xmx2g startupProbe: exec: command: - /bin/sh - -c - /opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092 /dev/null 21 initialDelaySeconds: 20 periodSeconds: 10 timeoutSeconds: 10 failureThreshold: 30 readinessProbe: exec: command: - /bin/sh - -c - /opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092 /dev/null 21 periodSeconds: 10 timeoutSeconds: 10 failureThreshold: 6 resources: requests: cpu: 1 memory: 4Gi limits: cpu: 2 memory: 8Gi volumeMounts: - name: data mountPath: /var/lib/kafka/data volumeClaimTemplates: - metadata: name: data spec: accessModes: - ReadWriteOnce storageClassName: standard # TODO 改成集群实际的 StorageClass resources: requests: storage: 50Gi --- apiVersion: policy/v1 kind: PodDisruptionBudget metadata: name: kafka-controller namespace: kafka spec: maxUnavailable: 1 selector: matchLabels: app.kubernetes.io/name: kafka app.kubernetes.io/component: controller --- apiVersion: policy/v1 kind: PodDisruptionBudget metadata: name: kafka-broker namespace: kafka spec: maxUnavailable: 1 selector: matchLabels: app.kubernetes.io/name: kafka app.kubernetes.io/component: broker K8S yaml文件部署kafka集群（Kraft模式）——筑梦之路
yaml文件方式只需要对存储类名、名空间、节点IP和镜像地址进行调整就可以在k8s部署起来的kafka集群Kraft模式该yaml适用于bitnami版本镜像该yaml支持集群内外访问。--- apiVersion: v1 kind: Service metadata: name: kafka-headless namespace: test spec: clusterIP: None publishNotReadyAddresses: true selector: app: kafka ports: - name: plaintext port: 9092 targetPort: 9092 - name: controller port: 9093 targetPort: 9093 --- apiVersion: v1 kind: Service metadata: name: kafka-svc namespace: test spec: type: ClusterIP selector: statefulset.kubernetes.io/pod-name: kafka-0 ports: - protocol: TCP port: 9092 targetPort: 9092 --- apiVersion: v1 kind: Service metadata: name: kafka-0 namespace: test spec: type: NodePort selector: statefulset.kubernetes.io/pod-name: kafka-0 ports: - protocol: TCP port: 39092 targetPort: 9092 nodePort: 39092 --- apiVersion: v1 kind: Service metadata: name: kafka-1 namespace: test spec: type: NodePort selector: statefulset.kubernetes.io/pod-name: kafka-1 ports: - protocol: TCP port: 39093 targetPort: 9092 nodePort: 39093 --- apiVersion: v1 kind: Service metadata: name: kafka-2 namespace: test spec: type: NodePort selector: statefulset.kubernetes.io/pod-name: kafka-2 ports: - protocol: TCP port: 39094 targetPort: 9092 nodePort: 39094 --- apiVersion: apps/v1 kind: StatefulSet metadata: name: kafka namespace: test spec: selector: matchLabels: app: kafka serviceName: kafka-headless replicas: 3 updateStrategy: type: RollingUpdate podManagementPolicy: OrderedReady template: metadata: labels: app: kafka spec: affinity: podAntiAffinity: requiredDuringSchedulingIgnoredDuringExecution: - labelSelector: matchExpressions: - key: app operator: In values: [kafka] topologyKey: kubernetes.io/hostname securityContext: runAsUser: 1000 fsGroup: 1000 containers: - name: kafka image: kafka:3.9.0-debian-12-r13 imagePullPolicy: Always command: - bash - -ec - | HOSTNAMEhostname -s if [[ $HOSTNAME ~ (.*)-([0-9])$ ]]; then ORD${BASH_REMATCH[2]} PORT$((ORD 39092)) export KAFKA_CFG_NODE_ID$ORD export KAFKA_CFG_ADVERTISED_LISTENERSPLAINTEXT://192.168.100.100:$PORT else echo Failed to get index from hostname $HOSTNAME exit 1 fi exec /entrypoint.sh /run.sh env: - {name: ALLOW_PLAINTEXT_LISTENER, value: yes} - {name: KAFKA_KRAFT_CLUSTER_ID, value: MkU3OEVBNTcwNTJENDM2Qk} - {name: KAFKA_CFG_PROCESS_ROLES, value: broker,controller} - {name: KAFKA_CFG_LISTENERS, value: PLAINTEXT://:9092,CONTROLLER://:9093} - {name: KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP, value: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT} - {name: KAFKA_CFG_INTER_BROKER_LISTENER_NAME, value: PLAINTEXT} - {name: KAFKA_CFG_CONTROLLER_LISTENER_NAMES, value: CONTROLLER} - {name: KAFKA_CFG_CONTROLLER_QUORUM_VOTERS, value: 0kafka-0.kafka-headless.test.svc.cluster.local:9093,1kafka-1.kafka-headless.test.svc.cluster.local:9093,2kafka-2.kafka-headless.test.svc.cluster.local:9093} - {name: KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR, value: 3} - {name: KAFKA_CFG_TRANSACTION_STATE_LOG_MIN_ISR, value: 3} - {name: KAFKA_CFG_TRANSACTION_STATE_LOG_REPLICATION_FACTOR, value: 3} - {name: KAFKA_CFG_DEFAULT_REPLICATION_FACTOR, value: 3} - {name: TZ, value: Asia/Shanghai} ports: - {containerPort: 9092, name: plaintext} - {containerPort: 9093, name: controller} resources: limits: {cpu: 500m, memory: 1Gi} requests: {cpu: 100m, memory: 500Mi} volumeMounts: - {name: kafka-data, mountPath: /bitnami/kafka} volumeClaimTemplates: - metadata: {name: kafka-data} spec: accessModes: [ReadWriteOnce] storageClassName: managed-nfs-storage resources: requests: {storage: 1000Gi}仅供参考。官方kafka版本镜像apache/kakfa# # Apache Kafka KRaft 模式 · Kubernetes StatefulSet 部署Combined 模式 / 3 节点 # 镜像apache/kafka:4.3.1Apache 官方 JVM 镜像自 3.7.0 起提供 # # 拓扑3 个 Pod 同时承担 broker controller 角色可容忍 1 个节点故障。 # 适用中小规模生产、测试/预发环境、希望用最少组件跑通 KRaft 的场景。 # 用法kubectl apply -f kafka-kraft-combined.yaml # # 部署前必须确认/修改 # 1) CLUSTER_ID —— 换成自己生成的集群 ID见 README「部署步骤」第 1 步 # 2) storageClassName —— 换成集群中实际存在的 StorageClass # 3) 存储容量、CPU/内存 requests/limits 按实际环境调整 # # kafka-kraft-combined.yaml --- apiVersion: v1 kind: Namespace metadata: name: kafka labels: app.kubernetes.io/name: kafka --- # 无头 Service为每个 Pod 提供稳定 DNS # kafka-0.kafka-headless.kafka.svc.cluster.local # kafka-1.kafka-headless.kafka.svc.cluster.local # kafka-2.kafka-headless.kafka.svc.cluster.local # publishNotReadyAddresses 必须为 trueKRaft 组建 quorum 时 Pod 之间需要互相解析 # 若等到 Ready 才注册 DNS会因互相解析不到而永远无法 Ready死锁。 apiVersion: v1 kind: Service metadata: name: kafka-headless namespace: kafka labels: app.kubernetes.io/name: kafka spec: clusterIP: None publishNotReadyAddresses: true selector: app.kubernetes.io/name: kafka ports: - name: plaintext port: 9092 targetPort: 9092 protocol: TCP - name: controller port: 9093 targetPort: 9093 protocol: TCP --- # 普通 ClusterIP Service给集群内客户端做 bootstrap 入口 # 客户端使用 kafka-bootstrap.kafka.svc.cluster.local:9092 apiVersion: v1 kind: Service metadata: name: kafka-bootstrap namespace: kafka labels: app.kubernetes.io/name: kafka spec: type: ClusterIP selector: app.kubernetes.io/name: kafka ports: - name: plaintext port: 9092 targetPort: 9092 protocol: TCP --- apiVersion: apps/v1 kind: StatefulSet metadata: name: kafka namespace: kafka labels: app.kubernetes.io/name: kafka spec: serviceName: kafka-headless replicas: 3 # 关键Parallel。KRaft 的 quorum 需要多个 Pod 同时起来 # 若用默认的 OrderedReadykafka-0 会因 quorum 不完整而无法 Ready # 导致 kafka-1/kafka-2 永远不被创建。 podManagementPolicy: Parallel updateStrategy: type: RollingUpdate selector: matchLabels: app.kubernetes.io/name: kafka template: metadata: labels: app.kubernetes.io/name: kafka spec: # Kafka 关停需要时间做数据落盘留足优雅退出时间 terminationGracePeriodSeconds: 60 securityContext: runAsUser: 1000 # 官方镜像内建 appuseruid1000, gid1000 runAsGroup: 1000 runAsNonRoot: true fsGroup: 1000 # 让 PVC 挂载后对 appuser 可写缺了会报权限错误 # 尽量把 3 个副本打散到不同节点软约束节点不足时仍可调度 affinity: podAntiAffinity: preferredDuringSchedulingIgnoredDuringExecution: - weight: 100 podAffinityTerm: topologyKey: kubernetes.io/hostname labelSelector: matchLabels: app.kubernetes.io/name: kafka containers: - name: kafka image: apache/kafka:4.3.1 imagePullPolicy: IfNotPresent # 官方镜像的启动命令是 /etc/kafka/docker/runCMD非 ENTRYPOINT可被覆盖。 # 这里包一层 bash用 Pod 名推导本节点的 node.id 与 advertised.listeners # 因为这两个值每个 Pod 都不同无法用静态 env 写死。 # 注意${POD_NAME##*-} 是运行时由 bash 解析的Kubernetes 只替换 $(VAR) 形式不会误伤。 command: - bash - -c - | set -euo pipefail export KAFKA_NODE_ID${POD_NAME##*-} export KAFKA_ADVERTISED_LISTENERSPLAINTEXT://${POD_NAME}.kafka-headless.kafka.svc.cluster.local:9092 exec /etc/kafka/docker/run ports: - name: plaintext containerPort: 9092 - name: controller containerPort: 9093 env: - name: POD_NAME valueFrom: fieldRef: fieldPath: metadata.name # ---- 集群标识 ---------------------------------------------------- # 示例值务必替换为自己生成的集群 IDkafka-storage.sh random-uuid。 # 同一集群所有节点必须一致集群 ID 只在首次格式化数据目录时写入 # 之后想更换必须清空数据卷重新格式化。 - name: CLUSTER_ID value: 4L6g3nShT-eMCtK--X86sw # ---- KRaft 角色与监听 -------------------------------------------- - name: KAFKA_PROCESS_ROLES value: broker,controller - name: KAFKA_LISTENERS value: PLAINTEXT://:9092,CONTROLLER://:9093 - name: KAFKA_LISTENER_SECURITY_PROTOCOL_MAP value: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT - name: KAFKA_INTER_BROKER_LISTENER_NAME value: PLAINTEXT - name: KAFKA_CONTROLLER_LISTENER_NAMES value: CONTROLLER # 控制器 quorum 成员nodeIdPod 的稳定 DNS - name: KAFKA_CONTROLLER_QUORUM_VOTERS value: 0kafka-0.kafka-headless.kafka.svc.cluster.local:9093,1kafka-1.kafka-headless.kafka.svc.cluster.local:9093,2kafka-2.kafka-headless.kafka.svc.cluster.local:9093 # ---- 存储 -------------------------------------------------------- - name: KAFKA_LOG_DIRS value: /var/lib/kafka/data # ---- 副本因子单节点测试环境要改成 1 ----------------------------- - name: KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR value: 3 - name: KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR value: 3 - name: KAFKA_TRANSACTION_STATE_LOG_MIN_ISR value: 2 # Kafka 4.x 新增的 share group队列协调器内部主题 - name: KAFKA_SHARE_COORDINATOR_STATE_TOPIC_REPLICATION_FACTOR value: 3 - name: KAFKA_SHARE_COORDINATOR_STATE_TOPIC_MIN_ISR value: 2 - name: KAFKA_DEFAULT_REPLICATION_FACTOR value: 3 - name: KAFKA_MIN_INSYNC_REPLICAS value: 2 - name: KAFKA_NUM_PARTITIONS value: 3 # ---- 常规参数 ---------------------------------------------------- - name: KAFKA_AUTO_CREATE_TOPICS_ENABLE value: false - name: KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS value: 0 - name: KAFKA_LOG_RETENTION_HOURS value: 168 # JVM 堆按容器内存 limits 的 1/2 左右设置留出页缓存和直接内存 - name: KAFKA_HEAP_OPTS value: -Xms1g -Xmx1g # 官方镜像里 bin 脚本齐全用 broker API 探测更贴近真的能服务请求 # 比单纯 TCP 端口探测更可靠。 startupProbe: exec: command: - /bin/sh - -c - /opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092 /dev/null 21 initialDelaySeconds: 20 periodSeconds: 10 timeoutSeconds: 10 failureThreshold: 30 readinessProbe: exec: command: - /bin/sh - -c - /opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092 /dev/null 21 periodSeconds: 10 timeoutSeconds: 10 failureThreshold: 6 # 刻意不配 livenessProbeKafka 启动慢且重启不能自愈多为 quorum/存储问题 # 探活失败重启只会造成滚动失败循环靠 startupProbe readinessProbe 已足够。 resources: requests: cpu: 1 memory: 2Gi limits: cpu: 2 memory: 4Gi volumeMounts: - name: data mountPath: /var/lib/kafka/data volumeClaimTemplates: - metadata: name: data spec: accessModes: - ReadWriteOnce storageClassName: standard # TODO 改成集群实际的 StorageClass resources: requests: storage: 20Gi --- # 滚动升级/节点维护时保证最多只有 1 个 Kafka Pod 不可用 apiVersion: policy/v1 kind: PodDisruptionBudget metadata: name: kafka namespace: kafka spec: maxUnavailable: 1 selector: matchLabels: app.kubernetes.io/name: kafka# # 可选把 Kafka 暴露到集群外NodePort 方案 # 镜像apache/kafka:4.3.1Apache 官方 JVM 镜像 # # 原理客户端连接 broker 后会拿到 broker 自报的 advertised.listeners 地址并直连 # 所以每个 broker 必须有自己的对外入口不能共用一个 Service/NodePort # 否则客户端会被转发到错误的 broker。因此这里给每个 Pod 建一个独立 Service # 用 selector: statefulset.kubernetes.io/pod-name 精确绑定到单个 Pod # 并固定 nodePort使地址可预测。 # # 使用前需要同时修改 StatefulSetkafka-kraft-combined.yaml # # 1) 增加一个 EXTERNAL 监听端口与安全协议映射 # - name: KAFKA_LISTENERS # value: PLAINTEXT://:9092,CONTROLLER://:9093,EXTERNAL://:9094 # - name: KAFKA_LISTENER_SECURITY_PROTOCOL_MAP # value: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,EXTERNAL:PLAINTEXT # # 2) 增加容器端口 9094containerPort: 9094, name: external # # 3) 增加下行 API 环境变量拿到 Pod 所在节点 IP # - name: NODE_IP # valueFrom: # fieldRef: # fieldPath: status.hostIP # # 4) 改写 command 中的 advertised.listeners把 EXTERNAL 指向节点 IP 固定 nodePort # nodePort 与下方 Service 的 30092/30093/30094 一一对应 # export KAFKA_ADVERTISED_LISTENERSPLAINTEXT://${POD_NAME}.kafka-headless.kafka.svc.cluster.local:9092,EXTERNAL://${NODE_IP}:$((30092 ${POD_NAME##*-})) # # 5) 客户端 bootstrap 用 任一节点IP:30092,另一节点IP:30093,第三节点IP:30094 # 若集群前面有负载均衡/LB建议把三个 nodePort 都注册到同一个对外地址上。 # # 安全提醒PLAINTEXT 暴露到公网等于无认证无加密生产环境务必改用 # SASL_SSL / SSL 监听镜像支持挂载 /etc/kafka/secrets 下的证书与 JAAS 配置。 # --- apiVersion: v1 kind: Service metadata: name: kafka-external-0 namespace: kafka labels: app.kubernetes.io/name: kafka app.kubernetes.io/component: external spec: type: NodePort selector: statefulset.kubernetes.io/pod-name: kafka-0 ports: - name: external port: 9094 targetPort: 9094 nodePort: 30092 protocol: TCP --- apiVersion: v1 kind: Service metadata: name: kafka-external-1 namespace: kafka labels: app.kubernetes.io/name: kafka app.kubernetes.io/component: external spec: type: NodePort selector: statefulset.kubernetes.io/pod-name: kafka-1 ports: - name: external port: 9094 targetPort: 9094 nodePort: 30093 protocol: TCP --- apiVersion: v1 kind: Service metadata: name: kafka-external-2 namespace: kafka labels: app.kubernetes.io/name: kafka app.kubernetes.io/component: external spec: type: NodePort selector: statefulset.kubernetes.io/pod-name: kafka-2 ports: - name: external port: 9094 targetPort: 9094 nodePort: 30094 protocol: TCP# # Apache Kafka KRaft 模式 · Kubernetes StatefulSet 部署Isolated 模式 # 镜像apache/kafka:4.3.1Apache 官方 JVM 镜像 # # 拓扑3 个专用 controller 3 个专用 broker角色分离共 6 个 Pod。 # 适用生产环境。controller 只跑元数据共识不受 broker 流量影响职责清晰、可独立扩缩容。 # 用法kubectl apply -f kafka-kraft-isolated.yaml # 与 kafka-kraft-combined.yaml 二选一不要同时部署 # # 部署前必须确认/修改 # 1) CLUSTER_ID、2) storageClassName、3) 容量与资源配置 # # --- apiVersion: v1 kind: Namespace metadata: name: kafka labels: app.kubernetes.io/name: kafka # Controller 层 --- apiVersion: v1 kind: Service metadata: name: kafka-controller-headless namespace: kafka labels: app.kubernetes.io/name: kafka app.kubernetes.io/component: controller spec: clusterIP: None publishNotReadyAddresses: true selector: app.kubernetes.io/name: kafka app.kubernetes.io/component: controller ports: - name: controller port: 9093 targetPort: 9093 protocol: TCP --- apiVersion: apps/v1 kind: StatefulSet metadata: name: kafka-controller namespace: kafka labels: app.kubernetes.io/name: kafka app.kubernetes.io/component: controller spec: serviceName: kafka-controller-headless replicas: 3 podManagementPolicy: Parallel # quorum 组建要求多个 Pod 同时启动 selector: matchLabels: app.kubernetes.io/name: kafka app.kubernetes.io/component: controller template: metadata: labels: app.kubernetes.io/name: kafka app.kubernetes.io/component: controller spec: terminationGracePeriodSeconds: 60 securityContext: runAsUser: 1000 runAsGroup: 1000 runAsNonRoot: true fsGroup: 1000 affinity: podAntiAffinity: preferredDuringSchedulingIgnoredDuringExecution: - weight: 100 podAffinityTerm: topologyKey: kubernetes.io/hostname labelSelector: matchLabels: app.kubernetes.io/name: kafka app.kubernetes.io/component: controller containers: - name: kafka image: apache/kafka:4.3.1 imagePullPolicy: IfNotPresent # controller 节点只需要 node.id注意 controller-only 节点严禁设置 # KAFKA_ADVERTISED_LISTENERS镜像启动脚本会直接报错退出。 command: - bash - -c - | set -euo pipefail export KAFKA_NODE_ID${POD_NAME##*-} exec /etc/kafka/docker/run ports: - name: controller containerPort: 9093 env: - name: POD_NAME valueFrom: fieldRef: fieldPath: metadata.name - name: CLUSTER_ID value: 4L6g3nShT-eMCtK--X86sw # TODO 替换为自己生成的集群 ID - name: KAFKA_PROCESS_ROLES value: controller - name: KAFKA_LISTENERS value: CONTROLLER://:9093 - name: KAFKA_LISTENER_SECURITY_PROTOCOL_MAP value: CONTROLLER:PLAINTEXT - name: KAFKA_CONTROLLER_LISTENER_NAMES value: CONTROLLER - name: KAFKA_CONTROLLER_QUORUM_VOTERS value: 0kafka-controller-0.kafka-controller-headless.kafka.svc.cluster.local:9093,1kafka-controller-1.kafka-controller-headless.kafka.svc.cluster.local:9093,2kafka-controller-2.kafka-controller-headless.kafka.svc.cluster.local:9093 - name: KAFKA_LOG_DIRS value: /var/lib/kafka/data # controller 也承载元数据日志保留策略可略短于业务数据 - name: KAFKA_LOG_RETENTION_HOURS value: 168 - name: KAFKA_HEAP_OPTS value: -Xms1g -Xmx1g # controller 节点不提供 broker 端口用 TCP 探测其 CONTROLLER 监听端口 # 更精确的 quorum 状态可用 kafka-metadata-quorum.sh 人工检查见部署手册「验证」。 startupProbe: tcpSocket: port: 9093 initialDelaySeconds: 20 periodSeconds: 10 timeoutSeconds: 5 failureThreshold: 30 readinessProbe: tcpSocket: port: 9093 periodSeconds: 15 timeoutSeconds: 5 failureThreshold: 6 resources: requests: cpu: 500m memory: 2Gi limits: cpu: 1 memory: 4Gi volumeMounts: - name: data mountPath: /var/lib/kafka/data volumeClaimTemplates: - metadata: name: data spec: accessModes: - ReadWriteOnce storageClassName: standard # TODO 改成集群实际的 StorageClass resources: requests: storage: 10Gi # Broker 层 --- apiVersion: v1 kind: Service metadata: name: kafka-broker-headless namespace: kafka labels: app.kubernetes.io/name: kafka app.kubernetes.io/component: broker spec: clusterIP: None publishNotReadyAddresses: true selector: app.kubernetes.io/name: kafka app.kubernetes.io/component: broker ports: - name: plaintext port: 9092 targetPort: 9092 protocol: TCP --- # 客户端 bootstrap 入口kafka-bootstrap.kafka.svc.cluster.local:9092 apiVersion: v1 kind: Service metadata: name: kafka-bootstrap namespace: kafka labels: app.kubernetes.io/name: kafka app.kubernetes.io/component: broker spec: type: ClusterIP selector: app.kubernetes.io/name: kafka app.kubernetes.io/component: broker ports: - name: plaintext port: 9092 targetPort: 9092 protocol: TCP --- apiVersion: apps/v1 kind: StatefulSet metadata: name: kafka-broker namespace: kafka labels: app.kubernetes.io/name: kafka app.kubernetes.io/component: broker spec: serviceName: kafka-broker-headless replicas: 3 podManagementPolicy: Parallel selector: matchLabels: app.kubernetes.io/name: kafka app.kubernetes.io/component: broker template: metadata: labels: app.kubernetes.io/name: kafka app.kubernetes.io/component: broker spec: terminationGracePeriodSeconds: 60 securityContext: runAsUser: 1000 runAsGroup: 1000 runAsNonRoot: true fsGroup: 1000 affinity: podAntiAffinity: preferredDuringSchedulingIgnoredDuringExecution: - weight: 100 podAffinityTerm: topologyKey: kubernetes.io/hostname labelSelector: matchLabels: app.kubernetes.io/name: kafka app.kubernetes.io/component: broker containers: - name: kafka image: apache/kafka:4.3.1 imagePullPolicy: IfNotPresent # broker 的 node.id 从 3 开始避免与 controller 的 0/1/2 冲突 command: - bash - -c - | set -euo pipefail export KAFKA_NODE_ID$(( ${POD_NAME##*-} 3 )) export KAFKA_ADVERTISED_LISTENERSPLAINTEXT://${POD_NAME}.kafka-broker-headless.kafka.svc.cluster.local:9092 exec /etc/kafka/docker/run ports: - name: plaintext containerPort: 9092 env: - name: POD_NAME valueFrom: fieldRef: fieldPath: metadata.name - name: CLUSTER_ID value: 4L6g3nShT-eMCtK--X86sw # TODO 与 controller 保持一致 - name: KAFKA_PROCESS_ROLES value: broker - name: KAFKA_LISTENERS value: PLAINTEXT://:9092 - name: KAFKA_LISTENER_SECURITY_PROTOCOL_MAP value: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT - name: KAFKA_INTER_BROKER_LISTENER_NAME value: PLAINTEXT - name: KAFKA_CONTROLLER_LISTENER_NAMES value: CONTROLLER - name: KAFKA_CONTROLLER_QUORUM_VOTERS value: 0kafka-controller-0.kafka-controller-headless.kafka.svc.cluster.local:9093,1kafka-controller-1.kafka-controller-headless.kafka.svc.cluster.local:9093,2kafka-controller-2.kafka-controller-headless.kafka.svc.cluster.local:9093 - name: KAFKA_LOG_DIRS value: /var/lib/kafka/data - name: KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR value: 3 - name: KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR value: 3 - name: KAFKA_TRANSACTION_STATE_LOG_MIN_ISR value: 2 - name: KAFKA_SHARE_COORDINATOR_STATE_TOPIC_REPLICATION_FACTOR value: 3 - name: KAFKA_SHARE_COORDINATOR_STATE_TOPIC_MIN_ISR value: 2 - name: KAFKA_DEFAULT_REPLICATION_FACTOR value: 3 - name: KAFKA_MIN_INSYNC_REPLICAS value: 2 - name: KAFKA_NUM_PARTITIONS value: 3 - name: KAFKA_AUTO_CREATE_TOPICS_ENABLE value: false - name: KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS value: 0 - name: KAFKA_LOG_RETENTION_HOURS value: 168 - name: KAFKA_HEAP_OPTS value: -Xms2g -Xmx2g startupProbe: exec: command: - /bin/sh - -c - /opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092 /dev/null 21 initialDelaySeconds: 20 periodSeconds: 10 timeoutSeconds: 10 failureThreshold: 30 readinessProbe: exec: command: - /bin/sh - -c - /opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092 /dev/null 21 periodSeconds: 10 timeoutSeconds: 10 failureThreshold: 6 resources: requests: cpu: 1 memory: 4Gi limits: cpu: 2 memory: 8Gi volumeMounts: - name: data mountPath: /var/lib/kafka/data volumeClaimTemplates: - metadata: name: data spec: accessModes: - ReadWriteOnce storageClassName: standard # TODO 改成集群实际的 StorageClass resources: requests: storage: 50Gi --- apiVersion: policy/v1 kind: PodDisruptionBudget metadata: name: kafka-controller namespace: kafka spec: maxUnavailable: 1 selector: matchLabels: app.kubernetes.io/name: kafka app.kubernetes.io/component: controller --- apiVersion: policy/v1 kind: PodDisruptionBudget metadata: name: kafka-broker namespace: kafka spec: maxUnavailable: 1 selector: matchLabels: app.kubernetes.io/name: kafka app.kubernetes.io/component: broker