apiVersion: v1
kind: ConfigMap
metadata:
  name: kafka-config
  namespace: default
data:
  server.properties: |
    broker.id=0
    listeners=PLAINTEXT://${POD_IP}:9092
    advertised.listeners=PLAINTEXT://${KAFKA_ADVERTISED_HOST_NAME}:${KAFKA_ADVERTISED_PORT}
    zookeeper.connect=${ZOOKEEPER_CONNECT}
    log.dirs=/kafka/data
    num.partitions=1
    default.replication.factor=1
    min.insync.replicas=1
---
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: kafka
  namespace: default
spec:
  serviceName: "kafka"
  replicas: 1
  selector:
    matchLabels:
      app: kafka
  template:
    metadata:
      labels:
        app: kafka
    spec:
      containers:
        - name: kafka
          image: confluentinc/cp-kafka:latest
          ports:
            - containerPort: 9092
          env:
            - name: KAFKA_BROKER_ID
              value: "0"
            - name: POD_IP
              valueFrom:
                fieldRef:
                  fieldPath: status.podIP
            - name: KAFKA_ADVERTISED_HOST_NAME
              valueFrom:
                fieldRef:
                  fieldPath: metadata.name
            - name: KAFKA_ZOOKEEPER_CONNECT
              value: "zookeeper:2181"
            - name: KAFKA_LISTENERS
              value: "PLAINTEXT://:9092"
            - name: KAFKA_ADVERTISED_LISTENERS
              value: "PLAINTEXT://:9092"
          volumeMounts:
            - name: kafka-storage
              mountPath: /kafka/data
  volumeClaimTemplates:
    - metadata:
        name: kafka-storage
      spec:
        accessModes: [ "ReadWriteOnce" ]
        resources:
          requests:
            storage: 10Gi
---
apiVersion: v1
kind: Service
metadata:
  name: kafka-service
  namespace: default
spec:
  ports:
    - port: 9092
      targetPort: 9092
  selector:
    app: kafka
  clusterIP: None