apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: kafka
spec:
  selector:
    matchLabels:
      app: kafka
  serviceName: kafka-svc-hs
  replicas: 1
  updateStrategy:
    type: RollingUpdate
  podManagementPolicy: Parallel
  template:
    metadata:
      labels:
        app: kafka
    spec:
      initContainers:
        - name: zookeeper-wait
          image: busybox
          command:
            - sh
            - -ec
            - |
              echo "Waiting for Zookeeper"
              until [ "imok" = "$(echo ruok | nc -w 1 zookeeper-svc 2181)" ]; do
                sleep 1
              done
      containers:
        - name: kafka
          image: wurstmeister/kafka:2.12-2.3.0
          env:
            - name: POD_NAME
              valueFrom:
                fieldRef:
                  fieldPath: metadata.name
            - name: POD_NAMESPACE
              valueFrom:
                fieldRef:
                  fieldPath: metadata.namespace
            - name: EXTERNAL_PORT
              value: "9092"
            - name: INTERNAL_PORT
              value: "9094"
            - name: KAFKA_ZOOKEEPER_CONNECT
              value: "zookeeper-svc:2181"
            - name: KAFKA_LISTENERS
              value: "INTERNAL://:$(INTERNAL_PORT),EXTERNAL://:$(EXTERNAL_PORT)"
            - name: KAFKA_INTER_BROKER_LISTENER_NAME
              value: "INTERNAL"
            - name: KAFKA_LISTENER_SECURITY_PROTOCOL_MAP
              value: "INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT"
            - name: KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR
              value: "1"
          command:
            - sh
            - -exc
            - |
              export KAFKA_BROKER_ID=${HOSTNAME##*-} 
              export KAFKA_ADVERTISED_LISTENERS="${KAFKA_ADVERTISED_LISTENERS},INTERNAL://${POD_NAME}.kafka-svc-hs.${POD_NAMESPACE}:${INTERNAL_PORT}"
              exec start-kafka.sh
          readinessProbe:
            tcpSocket:
              port: 9092
          volumeMounts:
            - name: kafka-data
              mountPath: /kafka
  volumeClaimTemplates:
    - metadata:
        name: kafka-data
      spec:
        accessModes: [ "ReadWriteOnce" ]
        storageClassName: standard
        resources:
          requests:
            storage: 2Gi
