{{- if (eq (include "kafka.combinedMode" .) "false") }}
{{- $thisValues := .Values.controller }}
{{- $thisFullName := include "kafka.controller.fullname" . }}
{{- $controllerFullName := include "kafka.controller.fullname" . }}
{{- $componet := "controller" }}
{{- $antiAffinityTopologyKey := default "kubernetes.io/hostname" $thisValues.antiAffinityTopologyKey }}
---
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: {{ $thisFullName }}
  labels:
    {{- include "kafka.labels" . | nindent 4 }}
    component: {{ $componet | quote }}
spec:
  {{- if $thisValues.replicaCount }}
  replicas: {{ $thisValues.replicaCount }}
  {{- end }}
  selector:
    matchLabels:
      {{- include "kafka.selectorLabels" . | nindent 6 }}
      component: {{ $componet | quote }}
  serviceName: {{ include "kafka.controller.headless.serviceName" . }}
  updateStrategy:
    type: {{ ( $thisValues.updateStrategy | default dict ).type | default "RollingUpdate" }}
  template:
    metadata:
      {{- if or $.Values.podAnnotations $thisValues.podAnnotations }}
      annotations:
        {{- with $.Values.podAnnotations }}
        {{- toYaml . | nindent 8 }}
        {{- end }}
        {{- with $thisValues.podAnnotations }}
        {{- toYaml . | nindent 8 }}
        {{- end }}
      {{- end }}
      labels:
        {{- include "kafka.selectorLabels" . | nindent 8 }}
        component: {{ $componet | quote }}
        {{- with $thisValues.podLabels }}
        {{- toYaml . | nindent 8 }}
        {{- end }}
    spec:
      {{- with .Values.imagePullSecrets }}
      imagePullSecrets:
        {{- toYaml . | nindent 8 }}
      {{- end }}
      serviceAccountName: {{ include "kafka.serviceAccountName" . }}
      {{- with $thisValues.podSecurityContext }}
      securityContext:
        {{- toYaml $thisValues.podSecurityContext | nindent 8 }}
      {{- end }}
      {{- if or $.Values.nodeSelector $thisValues.nodeSelector }}
      nodeSelector:
        {{- if $thisValues.nodeSelector }}
        {{- toYaml $thisValues.nodeSelector | nindent 8 }}
        {{- else if $.Values.nodeSelector }}
        {{- toYaml $.Values.nodeSelector | nindent 8 }}
        {{- end }}
      {{- end }}
      {{- if or (eq $thisValues.antiAffinity "hard") (eq $thisValues.antiAffinity "soft") $thisValues.nodeAffinity }}
      affinity:
      {{- end }}
      {{- if eq $thisValues.antiAffinity "hard" }}
        podAntiAffinity:
          requiredDuringSchedulingIgnoredDuringExecution:
          - labelSelector:
              matchLabels:
                {{- include "kafka.selectorLabels" . | nindent 18 }}
                component: {{ $componet | quote }}
            topologyKey: {{ $antiAffinityTopologyKey | quote }}
      {{- else if eq $thisValues.antiAffinity "soft" }}
        podAntiAffinity:
          preferredDuringSchedulingIgnoredDuringExecution:
          - weight: 5
            podAffinityTerm:
              labelSelector:
                matchLabels:
                  {{- include "kafka.selectorLabels" . | nindent 18 }}
                  component: {{ $componet | quote }}
              topologyKey: {{ $antiAffinityTopologyKey | quote }}
      {{- end }}
      {{- with $thisValues.podAffinity }}
        podAffinity:
        {{ toYaml . | indent 10 }}
      {{- end }}
      {{- with $thisValues.nodeAffinity }}
        nodeAffinity:
        {{ toYaml . | indent 10 }}
      {{- end }}
      containers:
      - name: kafka
        image: {{ include "kafka.kafkaImage" . | quote }}
      {{- if .Values.image.pullPolicy }}
        imagePullPolicy: {{ .Values.image.pullPolicy | quote }}
      {{- end }}
        env:
        {{- include "kafka.controller.containerEnv" . | nindent 8 }}
        ports:
        - containerPort: {{ .Values.containerPort.controller }}
          name: controller
          protocol: TCP
        {{- with $thisValues.livenessProbe }}
        livenessProbe:
          {{- toYaml . | nindent 10 }}
        {{- end }}
        {{- with $thisValues.readinessProbe }}
        readinessProbe:
          {{- toYaml . | nindent 10 }}
        {{- end }}
        {{- with $thisValues.resources }}
        resources:
          {{- toYaml . | nindent 10 }}
        {{- end }}
        volumeMounts:
        - mountPath: /opt/kafka/data
          name: data
          subPath: data
        - mountPath: /opt/kafka/logs
          name: data
          subPath: logs
        - mountPath: /entrypoint.sh
          name: entrypoint-sh
          subPath: entrypoint.sh
        lifecycle:
          preStop:
            exec:
              command: ["sh", "-c", "sleep 10; bin/kafka-server-stop.sh"]
      terminationGracePeriodSeconds: {{ $thisValues.terminationGracePeriodSeconds }}
      volumes:
      - name: entrypoint-sh
        configMap:
          items:
          - key: entrypoint.sh
            path: entrypoint.sh
          name: {{ include "kafka.entrypoint.configmapName" . }}
          defaultMode: 0744
  {{- if not $thisValues.persistence.enabled }}
      - name: data
        emptyDir: {}
  {{- else }}
  volumeClaimTemplates:
    - metadata:
        name: data
        {{- with $thisValues.persistence.annotations }}
        annotations:
          {{- . | toYaml | nindent 10 }}
        {{- end }}
      spec:
        accessModes:
          - ReadWriteOnce
        {{- with $thisValues.persistence.storageClass }}
        storageClassName: {{ if (eq "-" .) }}""{{ else }}{{ . }}{{ end }}
        {{- end }}
        resources:
          requests:
            storage: {{ $thisValues.persistence.size | quote }}
  {{- end }}
{{- end }}