#version: kafka-cluster, 1.0.0
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaNodePool
metadata:
  name: controller
  namespace: "kafka-cluster"
  labels:
    strimzi.io/cluster: "kafka-cluster"
spec:
  replicas: 3
  roles:
    - controller
  storage:
    type: jbod
    volumes:
      - id: 0
        type: persistent-claim
        size: 5Gi
        kraftMetadata: shared
        class: standard-rwo
---
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaNodePool
metadata:
  name: broker
  namespace: "kafka-cluster"
  labels:
    strimzi.io/cluster: "kafka-cluster"
spec:
  replicas: 3
  roles:
    - broker
  storage:
    type: jbod
    volumes:
      - id: 0
        type: persistent-claim
        size: 5Gi
        class: standard-rwo
        kraftMetadata: shared
---
apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: "kafka-cluster"
  namespace: "kafka-cluster"
  annotations:
    strimzi.io/node-pools: enabled
    strimzi.io/kraft: enabled
spec:
  kafka:
    version: 4.0.0
    metadataVersion: "4.0"
    template:
      kafkaContainer:
        env:
        - name: KAFKA_OPTS
          value: "-Duser.timezone=Asia/Seoul"
    listeners:
      - name: tls
        type: cluster-ip
        port: 9093
        tls: false
        authentication:
          type: oauth
          clientId: "service-kafka-cluster"
          clientSecret:
            key: client-secret
            secretName: kafka-cluster-oauth-secret
          checkIssuer: true
          checkAccessTokenType: true
          accessTokenIsJwt: true
          checkAudience: false
          enableOauthBearer: true
          validIssuerUri: https://keycloak.gke.paasup.io/realms/paasup
          jwksEndpointUri: https://keycloak.gke.paasup.io/realms/paasup/protocol/openid-connect/certs
          userNameClaim: preferred_username
          customClaimCheck: '''service-kafka-cluster'' in @.realm_access.roles'

    config:
      offsets.topic.replication.factor: 3
      transaction.state.log.replication.factor: 3
      transaction.state.log.min.isr: 2
      default.replication.factor: 3
      min.insync.replicas: 2
      auto.create.topics.enable: false
      num.partitions: 3
      delete.topic.enable: true
    authorization:
      type: simple
    logging:
      type: inline
      loggers:
        kafka.root.logger.level: INFO
  entityOperator:
    topicOperator: { }
    userOperator: { }

---
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: "kafka-cluster"
  namespace: "kafka-cluster"
  labels:
    strimzi.io/cluster: "kafka-cluster"
  annotations:
    strimzi.io/use-connector-resources: 'true'
spec:
  image: paasup/kafka-connect:0.2
  replicas: 1
  bootstrapServers: "kafka-cluster-kafka-tls-bootstrap.kafka-cluster.svc.cluster.local:9093"
  config:
    group.id: "kafka-cluster-default-connect"
    offset.storage.topic: "kafka-cluster.connect-offsets"
    config.storage.topic: "kafka-cluster.connect-configs"
    status.storage.topic: "kafka-cluster.connect-status"
    key.converter: org.apache.kafka.connect.json.JsonConverter
    value.converter: org.apache.kafka.connect.json.JsonConverter
    plugin.path: /opt/kafka/plugins
    topic.creation.enable: "true"
  authentication:
    type: oauth
    tokenEndpointUri: https://keycloak.gke.paasup.io/realms/paasup/protocol/openid-connect/token
    clientId: "service-kafka-cluster-kafka-connect"
    clientSecret:
      secretName: kafka-connect-oauth-secret
      key: client-secret


  template:
    connectContainer:
      volumeMounts:
        - name: truststore-volume
          mountPath: /mnt/truststore/truststore.jks
          subPath: truststore.jks
          readOnly: true
    pod:
      volumes:
        - name: truststore-volume
          secret:
            secretName: truststore
  
  logging:
    type: inline
    loggers:
      rootLogger.level: INFO

---
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  name: service-account-service-kafka-cluster-kafka-connect
  namespace: "kafka-cluster"
  labels:
    strimzi.io/cluster: "kafka-cluster"
spec:
  authorization:
    type: simple
    acls:
      - resource:
          type: topic
          name: "*"
          patternType: literal
        operations:
          - Read
          - Describe
          - DescribeConfigs
          - Write
      - resource:
          type: group
          name: "*"
          patternType: literal
        operations:
          - Read
          - Write
          - Describe     
         
---
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaTopic
metadata:
  labels:
    strimzi.io/cluster: kafka-cluster
  name: "kafka-cluster-connect-offsets"
  namespace: "kafka-cluster"
spec:
  partitions: 1
  replicas: 3
  topicName: "kafka-cluster.connect-offsets"
  config:
    cleanup.policy: compact
    retention.ms: 604800000
    segment.bytes: 1073741824

---
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaTopic
metadata:
  labels:
    strimzi.io/cluster: kafka-cluster
  name: "kafka-cluster-connect-configs"
  namespace: "kafka-cluster"
spec:
  partitions: 1
  replicas: 3
  topicName: "kafka-cluster.connect-configs"
  config:
    cleanup.policy: compact
    retention.ms: 604800000
    segment.bytes: 1073741824

---
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaTopic
metadata:
  labels:
    strimzi.io/cluster: kafka-cluster
  name: "kafka-cluster-connect-status"
  namespace: "kafka-cluster"
spec:
  partitions: 1
  replicas: 3
  topicName: "kafka-cluster.connect-status"
  config:
    cleanup.policy: compact
    retention.ms: 604800000
    segment.bytes: 1073741824