apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: kafka
spec:
  selector:
    matchLabels:
      app: kafka
  serviceName: "kafka"
  replicas: 1
  template:
    metadata:
      labels:
        app: kafka
    spec:
      terminationGracePeriodSeconds: 10
      containers:
      - name: kafka
        image: bitnami/kafka:2.5.0
        imagePullPolicy: Always
        ports:
        - containerPort: 9092
        env:
        - name: ALLOW_PLAINTEXT_LISTENER
          value: "yes"
        - name: KAFKA_CLIENT_USER
          value: "admin"
        - name: KAFKA_CFG_ZOOKEEPER_CONNECT
          value: "zookeeper-service:2181"
        - name: KAFKA_CFG_MESSAGE_MAX_BYTES
          value: "50000000"
        - name: KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP
          value: "INTERNAL_PLAINTEXT:PLAINTEXT"
        - name: KAFKA_CFG_ADVERTISED_LISTENERS
          value: "INTERNAL_PLAINTEXT://kafka-service:9092"
        - name: KAFKA_LISTENERS
          value: "INTERNAL_PLAINTEXT://0.0.0.0:9092"
        - name: KAFKA_INTER_BROKER_LISTENER_NAME
          value: "INTERNAL_PLAINTEXT"
        # command: ["/bin/sh"]
        # args: ["-c", "export KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://kafka-service:9092; \
        #   /opt/bitnami/scripts/kafka/entrypoint.sh /opt/bitnami/scripts/kafka/run.sh"]
        volumeMounts:
        - name: kafka-persistent-storage
          mountPath: /bitnami/kafka
        resources:
          limits:
            memory: "2Gi"
            cpu: "2000m"
          requests:
            memory: "128Mi"
            cpu: "100m"
      nodeSelector:
        intendedfor: kafka
      securityContext:  
        fsGroup: 1001
  volumeClaimTemplates:
  - metadata:
      name: kafka-persistent-storage
    spec:
      accessModes: [ "ReadWriteOnce" ]
      resources:
        requests:
          storage: 100Gi
