apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: kafka
  namespace: jolt
spec:
  selector:
    matchLabels:
      app: kafka
  serviceName: "broker"
  replicas: 3
  updateStrategy:
    type: RollingUpdate
  podManagementPolicy: Parallel
  template:
    metadata:
      labels:
        app: kafka
      annotations:
        ad.datadoghq.com/kafka.check_names: '["kafka"]'
        ad.datadoghq.com/kafka.init_configs: |
          [
            {
              "is_jmx": true,
              "collect_default_metrics": true
            }
          ]
        ad.datadoghq.com/kafka.logs: '[{"source":"kafka", "service":"broker"}]'
        ad.datadoghq.com/kafka.instances: |
          [
            {
              "host": "%%host%%",
              "port": "9999"
            }
          ]
    spec:
      affinity:
        podAntiAffinity:
          requiredDuringSchedulingIgnoredDuringExecution:
          - labelSelector:
              matchExpressions:
              - key: app
                operator: In
                values:
                - kafka
            topologyKey: "kubernetes.io/hostname"
        podAffinity:
          requiredDuringSchedulingIgnoredDuringExecution:
          - labelSelector:
              matchExpressions:
              - key: appalias
                operator: In
                values:
                - pzoo
            topologyKey: "kubernetes.io/hostname"
      terminationGracePeriodSeconds: 30
      initContainers:
      - name: init-config
        image: solsson/kafka-initutils@sha256:f6d9850c6c3ad5ecc35e717308fddb47daffbde18eb93e98e031128fe8b899ef
        env:
        - name: NODE_NAME
          valueFrom:
            fieldRef:
              fieldPath: spec.nodeName
        - name: POD_NAME
          valueFrom:
            fieldRef:
              fieldPath: metadata.name
        - name: POD_NAMESPACE
          valueFrom:
            fieldRef:
              fieldPath: metadata.namespace
        command: ['/bin/bash', '/etc/kafka-configmap/init.sh']
        volumeMounts:
        - name: configmap
          mountPath: /etc/kafka-configmap
        - name: config
          mountPath: /etc/kafka
        - name: extensions
          mountPath: /opt/kafka/libs/extensions
      containers:
      - name: kafka
        image: solsson/kafka:2.4.1@sha256:79761e15919b4fe9857ec00313c9df799918ad0340b684c0163ab7035907bb5a
        ports:
        - name: inside
          containerPort: 9092
        - name: outside
          containerPort: 9094
          hostPort: 9094
        - name: jmx
          containerPort: 9999
        env:
        - name: CLASSPATH
          value: /opt/kafka/libs/extensions/*
        - name: KAFKA_LOG4J_OPTS
          value: -Dlog4j.configuration=file:/etc/kafka/log4j.properties
        - name: JMX_PORT
          value: "9999"
        - name: POD_IP
          valueFrom:
            fieldRef:
              fieldPath: status.podIP
        - name: KAFKA_HEAP_OPTS
          value: >-
            -Xmx4G
            -Xms4G
        - name: KAFKA_JMX_OPTS
          value: >-
            -Dcom.sun.management.jmxremote=true
            -Dcom.sun.management.jmxremote.authenticate=false
            -Dcom.sun.management.jmxremote.ssl=false
            -Dcom.sun.management.jmxremote.local.only=false
            -Dcom.sun.management.jmxremote.port=9999
            -Dcom.sun.management.jmxremote.rmi.port=9999
            -Djava.rmi.server.hostname=$(POD_IP)
        command:
        - ./bin/kafka-server-start.sh
        - /etc/kafka/server.properties
        lifecycle:
          preStop:
            exec:
             command: ["sh", "-ce", "kill -s TERM 1; while $(kill -0 1 2>/dev/null); do sleep 1; done"]
        resources:
          requests:
            cpu: 200m
            memory: 4Gi
          limits:
            cpu: 400m
            memory: 8Gi
        readinessProbe:
          tcpSocket:
            port: 9092
          timeoutSeconds: 1
        volumeMounts:
        - name: config
          mountPath: /etc/kafka
        - name: data
          mountPath: /var/lib/kafka/data
        - name: extensions
          mountPath: /opt/kafka/libs/extensions
      volumes:
      - name: configmap
        configMap:
          name: broker-config
      - name: config
        emptyDir: {}
      - name: extensions
        emptyDir: {}
  volumeClaimTemplates:
  - metadata:
      name: data
    spec:
      accessModes: [ "ReadWriteOnce" ]
      storageClassName: aws-efs
      resources:
        requests:
          storage: 512Gi
