apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: kafka-broker
  namespace: kafka
  labels:
    app: kafka-broker
spec:
  serviceName: kafka-broker-headless
  replicas: 3
  selector:
    matchLabels:
      app: kafka-broker
  template:
    metadata:
      labels:
        app: kafka-broker
    spec:
      containers:
      - name: kafka-broker
        image: xxx.xuanyuan.run/apache/kafka:latest
        ports:
        - containerPort: 9092
          name: client
        - containerPort: 19092
          name: internal
        env:
        - name: KAFKA_NODE_ID
          valueFrom:
            fieldRef:
              fieldPath: metadata.name
        - name: KAFKA_PROCESS_ROLES
          value: "broker"
        - name: KAFKA_LISTENERS
          value: "PLAINTEXT://:19092,PLAINTEXT_HOST://:9092"
        - name: KAFKA_ADVERTISED_LISTENERS
          value: "PLAINTEXT://$(MY_POD_NAME).kafka-broker-headless.kafka.svc.cluster.local:19092,PLAINTEXT_HOST://localhost:9092"
        - name: MY_POD_NAME
          valueFrom:
            fieldRef:
              fieldPath: metadata.name
        - name: KAFKA_INTER_BROKER_LISTENER_NAME
          value: "PLAINTEXT"
        - name: KAFKA_CONTROLLER_LISTENER_NAMES
          value: "CONTROLLER"
        - name: KAFKA_LISTENER_SECURITY_PROTOCOL_MAP
          value: "CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT"
        - name: KAFKA_CONTROLLER_QUORUM_VOTERS
          value: "1@kafka-controller-0.kafka-controller-headless.kafka.svc.cluster.local:9093,2@kafka-controller-1.kafka-controller-headless.kafka.svc.cluster.local:9093,3@kafka-controller-2.kafka-controller-headless.kafka.svc.cluster.local:9093"
        - name: KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS
          value: "0"
        volumeMounts:
        - name: kafka-data
          mountPath: /tmp/kafka-logs
        readinessProbe:
          tcpSocket:
            port: 9092
          initialDelaySeconds: 15
          periodSeconds: 10
        livenessProbe:
          tcpSocket:
            port: 9092
          initialDelaySeconds: 30
          periodSeconds: 10
  volumeClaimTemplates:
  - metadata:
      name: kafka-data
    spec:
      accessModes: [ "ReadWriteOnce" ]
      storageClassName: "dynamic-storage-class"
      resources:
        requests:
          storage: 10Gi

---
apiVersion: v1
kind: Service
metadata:
  name: kafka-broker-headless
  namespace: kafka
  labels:
    app: kafka-broker
spec:
  ports:
  - port: 9092
    name: client
  - port: 19092
    name: internal
  clusterIP: None
  selector:
    app: kafka-broker

---
apiVersion: v1
kind: Service
metadata:
  name: kafka-broker-external
  namespace: kafka
  labels:
    app: kafka-broker
spec:
  type: NodePort
  ports:
  - port: 9092
    targetPort: 9092
    nodePort: 30092
    name: client
  selector:
    app: kafka-broker