apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: kafka-statefulset
  labels:
    app: kafka
spec:
  replicas: 2
  selector:
    matchLabels:
      app: kafka
  serviceName: "kafka-service"
  template:
    metadata:
      labels:
        app: kafka
    spec:
      containers:
        - name: kafka
          image: bitnami/kafka:3.5.1
          imagePullPolicy: IfNotPresent
          resources:
            limits:
              memory: "2Gi"
              cpu: "1000m"
            requests:
              memory: "1Gi"
              cpu: "500m"
          ports:
            - containerPort: 9092 # PLAINTEXT
            - containerPort: 9093 # CONTROLLER
            - containerPort: 9094 # EXTERNAL
          env:
            - name: TZ
              value: "Asia/Shanghai"
            - name: KAFKA_CFG_NODE_ID
              value: "0"
            - name: KAFKA_CFG_PROCESS_ROLES
              value: "controller,broker"
            - name: KAFKA_CFG_CONTROLLER_QUORUM_VOTERS
              value: "0@kafka-service:9093"
            - name: KAFKA_CFG_LISTENERS
              value: "PLAINTEXT://:9092,CONTROLLER://:9093,EXTERNAL://:9094"
            - name: KAFKA_CFG_ADVERTISED_LISTENERS
              value: "PLAINTEXT://kafka-service:9092,EXTERNAL://kafka-service:19094"
            - name: KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP
              value: "CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT,PLAINTEXT:PLAINTEXT"
            - name: KAFKA_CFG_CONTROLLER_LISTENER_NAMES
              value: "CONTROLLER"
            - name: KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE
              value: "true"
          volumeMounts:
            - name: kafka-data
              mountPath: /bitnami/kafka

      volumes:
        - name: kafka-data
          persistentVolumeClaim:
            claimName: kafka-pvc

---
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: kafka-pvc
spec:
  accessModes:
    - ReadWriteOnce
  resources:
    requests:
      storage: 10Gi