{{- if .Values.kafka.create }}
---
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: {{ include "deepfence-console.fullname" . }}-kafka-broker
  labels:
    {{- include "deepfence-console.labels" . | nindent 4 }}
    name: {{ include "deepfence-console.fullname" . }}-kafka-broker
spec:
  serviceName: {{ include "deepfence-console.fullname" . }}-kafka-broker
  replicas: {{ .Values.kafka.replicaCount }}
  selector:
    matchLabels:
      {{- include "deepfence-console.selectorLabels" . | nindent 6 }}
      name: {{ include "deepfence-console.fullname" . }}-kafka-broker
  template:
    metadata:
      {{- with .Values.kafka.podAnnotations }}
      annotations:
        {{- toYaml . | nindent 8 }}
      {{- end }}
      labels:
        {{- include "deepfence-console.selectorLabels" . | nindent 8 }}
        name: {{ include "deepfence-console.fullname" . }}-kafka-broker
    spec:
      serviceAccountName: {{ include "deepfence-console.serviceAccountName" . }}
      securityContext:
        {{- toYaml .Values.kafka.podSecurityContext | nindent 8 }}
      affinity:
        podAntiAffinity:
          preferredDuringSchedulingIgnoredDuringExecution:
            - weight: 1
              podAffinityTerm:
                labelSelector:
                  matchExpressions:
                    - key: name
                      operator: In
                      values:
                        - kafka-broker
                topologyKey: kubernetes.io/hostname
      initContainers:
        - name: init
          image: busybox
          imagePullPolicy: IfNotPresent
          securityContext:
            privileged: true
          command:  ["/bin/chmod","-R","777", "/data"]
          volumeMounts:
            - name: "kafka-broker"
              mountPath: /data
      containers:
        - name: kafka
          image: "{{ .Values.global.imageRepoPrefix }}/{{ .Values.kafka.image.repository }}:{{ default .Values.global.imageTag .Values.kafka.image.tag }}"
          imagePullPolicy: {{ .Values.kafka.image.pullPolicy }}
          ports:
            - containerPort: 9092
            - containerPort: 9093
          env:
            - name: POD_NAME
              valueFrom:
                fieldRef:
                  fieldPath: metadata.name
            - name: NAMESPACE
              valueFrom:
                fieldRef:
                  fieldPath: metadata.namespace
            - name: REPLICAS
              value: "{{ .Values.kafka.replicaCount }}"
            - name: SERVICE
              value: {{ include "deepfence-console.fullname" . }}-kafka-broker
            - name: KAFKA_LOG_DIRS
              value: /data/kafka
          envFrom:
            - configMapRef:
                name: {{ include "deepfence-console.fullname" . }}-kafka-broker-config
          livenessProbe:
            tcpSocket:
              port: 9092
            initialDelaySeconds: 120
            periodSeconds: 30
          # readinessProbe:
          #   tcpSocket:
          #     port: 9092
          #   initialDelaySeconds: 120
          #   periodSeconds: 30
          resources:
            {{- toYaml .Values.kafka.resources | nindent 12 }}
          volumeMounts:
          - name: "kafka-broker"
            mountPath: /data
      {{- with .Values.kafka.nodeSelector }}
      nodeSelector:
        {{- toYaml . | nindent 8 }}
      {{- end }}
      {{- with .Values.kafka.tolerations }}
      tolerations:
        {{- toYaml . | nindent 8 }}
      {{- end }}
  volumeClaimTemplates:
    - metadata:
        name: "kafka-broker"
      spec:
        accessModes:
          - "ReadWriteOnce"
        resources:
          requests:
            storage: "{{ .Values.kafka.volumeSize }}"
        storageClassName: {{ default .Values.global.storageClass .Values.kafka.storageClass }}
---
apiVersion: v1
kind: Service
metadata:
  name: {{ include "deepfence-console.fullname" . }}-kafka-broker
  labels:
    {{- include "deepfence-console.labels" . | nindent 4 }}
    name: {{ include "deepfence-console.fullname" . }}-kafka-broker
spec:
  type: ClusterIP
  clusterIP: None
  selector:
      {{- include "deepfence-console.selectorLabels" . | nindent 6 }}
      name: {{ include "deepfence-console.fullname" . }}-kafka-broker
  ports:
    - name: '9092'
      port: 9092
      protocol: TCP
      targetPort: 9092
    - name: '9093'
      port: 9093
      protocol: TCP
      targetPort: 9093
{{- end }}