apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: {{ include "kafka-cluster.fullname" . }}
  namespace: {{ include "kafka-cluster.namespace" . }}
  labels:
    {{- include "kafka-cluster.labels" . | nindent 4 }}
  annotations:
    strimzi.io/node-pools: enabled
    strimzi.io/kraft: enabled
spec:
  kafka:
    version: {{ .Values.kafka.version }}
    metadataVersion: {{ .Values.kafka.metadataVersion | quote }}
    template:
      kafkaContainer:
        env:
          - name: KAFKA_OPTS
            value: "-Duser.timezone={{ .Values.kafka.timezone }}"
    listeners:
      - name: tls
        type: cluster-ip
        port: {{ .Values.listener.port }}
        tls: {{ .Values.listener.tls }}
        authentication:
          type: oauth
          clientId: {{ .Values.oauth.keycloakClientId | quote }}
          clientSecret:
            key: {{ .Values.oauth.secretKey }}
            secretName: {{ .Values.oauth.secretName }}
          checkIssuer: true
          checkAccessTokenType: true
          accessTokenIsJwt: true
          checkAudience: false
          enableOauthBearer: true
          validIssuerUri: {{ .Values.oauth.keycloakUrl }}/realms/{{ .Values.oauth.keycloakRealm }}
          jwksEndpointUri: {{ .Values.oauth.keycloakUrl }}/realms/{{ .Values.oauth.keycloakRealm }}/protocol/openid-connect/certs
          userNameClaim: preferred_username
          customClaimCheck: "'{{ .Values.oauth.keycloakClientId }}' in @.realm_access.roles"
          tlsTrustedCertificates:
            - secretName: {{ .Values.oauth.tlsSecretName }}
              certificate: {{ .Values.oauth.tlsCertificate }}
      {{- if .Values.internalListener.enabled }}
      - name: internal
        type: cluster-ip
        port: {{ .Values.internalListener.port }}
        tls: {{ .Values.internalListener.tls }}
        authentication:
          type: scram-sha-512
      {{- end }}
    config:
      offsets.topic.replication.factor: {{ .Values.config.offsetsTopicReplicationFactor }}
      transaction.state.log.replication.factor: {{ .Values.config.transactionStateLogReplicationFactor }}
      transaction.state.log.min.isr: {{ .Values.config.transactionStateLogMinIsr }}
      default.replication.factor: {{ .Values.config.defaultReplicationFactor }}
      min.insync.replicas: {{ .Values.config.minInsyncReplicas }}
      auto.create.topics.enable: {{ .Values.config.autoCreateTopicsEnable }}
      num.partitions: {{ .Values.config.numPartitions }}
      delete.topic.enable: {{ .Values.config.deleteTopicEnable }}
    authorization:
      type: simple
    logging:
      type: inline
      loggers:
        kafka.root.logger.level: {{ .Values.logging.kafka.rootLoggerLevel }}
  entityOperator:
    topicOperator:
      {{- if .Values.entityOperator.topicOperator.resources }}
      resources:
        requests:
          memory: {{ .Values.entityOperator.topicOperator.resources.requests.memory }}
          cpu: {{ .Values.entityOperator.topicOperator.resources.requests.cpu }}
        limits:
          memory: {{ .Values.entityOperator.topicOperator.resources.limits.memory }}
          cpu: {{ .Values.entityOperator.topicOperator.resources.limits.cpu }}
      {{- end }}
    userOperator:
      {{- if .Values.entityOperator.userOperator.resources }}
      resources:
        requests:
          memory: {{ .Values.entityOperator.userOperator.resources.requests.memory }}
          cpu: {{ .Values.entityOperator.userOperator.resources.requests.cpu }}
        limits:
          memory: {{ .Values.entityOperator.userOperator.resources.limits.memory }}
          cpu: {{ .Values.entityOperator.userOperator.resources.limits.cpu }}
      {{- end }}