apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  name:  {{ include "flink-cdc-session.fullname" . }}-session
  namespace: {{ .Release.Namespace }}
  labels:
    {{- include "flink-cdc-session.labels" . | nindent 4 }}
spec:
  image: {{ .Values.image.repository }}:{{ .Values.image.tag }}
  imagePullPolicy: {{ .Values.image.pullPolicy }}
  flinkVersion: {{ .Values.flinkVersion }}
  flinkConfiguration:
    classloader.resolve-order: parent-first
    taskmanager.numberOfTaskSlots: {{ .Values.taskManager.taskSlots | quote }}
    execution.checkpointing.interval: {{ .Values.checkpoint.interval }}
    execution.checkpointing.mode: {{ .Values.checkpoint.mode }}
    execution.checkpointing.dir: {{ .Values.checkpointStorage.checkpointDir }}
    execution.checkpointing.savepoint-dir: {{ .Values.checkpointStorage.savepointDir }}
    classloader.parent-first-patterns.additional: org.apache.iceberg.
    {{- if eq .Values.checkpointStorage.storageType "s3" }}
    {{- if .Values.checkpointStorage.s3.endpoint }}
    s3.endpoint: {{ .Values.checkpointStorage.s3.endpoint }}
    s3.path-style-access: {{ .Values.checkpointStorage.s3.pathStyleAccess | quote }}
    {{- end }}
    {{- end }}
    {{- if .Values.truststore.enabled }}
    env.java.opts.jobmanager: >-
      -Djavax.net.ssl.trustStore=/opt/flink/certs/truststore.jks
      -Djavax.net.ssl.trustStorePassword={{ .Values.truststore.password }}
      -Djavax.net.ssl.trustStoreType=JKS
    env.java.opts.taskmanager: >-
      -Djavax.net.ssl.trustStore=/opt/flink/certs/truststore.jks
      -Djavax.net.ssl.trustStorePassword={{ .Values.truststore.password }}
      -Djavax.net.ssl.trustStoreType=JKS
    {{- end }}
  serviceAccount: {{ include "flink-cdc-session.serviceAccountName" . }}
  jobManager:
    replicas: {{ .Values.jobManager.replicas }}
    resource:
      memory: {{ .Values.jobManager.memory }}
      cpu: {{ .Values.jobManager.cpu }}
  taskManager:
    replicas: {{ .Values.taskManager.replicas }}
    resource:
      memory: {{ .Values.taskManager.memory }}
      cpu: {{ .Values.taskManager.cpu }}
  podTemplate:
    spec:
      containers:
        - name: flink-main-container
          {{- if eq .Values.checkpointStorage.storageType "gcs" }}
          env:
            - name: GOOGLE_APPLICATION_CREDENTIALS
              value: {{ .Values.checkpointStorage.gcs.credentialPath }}
          {{- else if eq .Values.checkpointStorage.storageType "s3" }}
          {{- if or .Values.checkpointStorage.s3.existingSecret .Values.checkpointStorage.s3.accessKeyId }}
          env:
            {{- if .Values.checkpointStorage.s3.existingSecret }}
            - name: AWS_ACCESS_KEY_ID
              valueFrom:
                secretKeyRef:
                  name: {{ .Values.checkpointStorage.s3.existingSecret }}
                  key: {{ .Values.checkpointStorage.s3.accessKeyIdKey }}
            - name: AWS_SECRET_ACCESS_KEY
              valueFrom:
                secretKeyRef:
                  name: {{ .Values.checkpointStorage.s3.existingSecret }}
                  key: {{ .Values.checkpointStorage.s3.secretAccessKeyKey }}
            {{- else }}
            - name: AWS_ACCESS_KEY_ID
              value: {{ .Values.checkpointStorage.s3.accessKeyId }}
            - name: AWS_SECRET_ACCESS_KEY
              value: {{ .Values.checkpointStorage.s3.secretAccessKey }}
            {{- end }}
          {{- end }}
          {{- end }}
          volumeMounts:
            - name: pipeline-config
              mountPath: /opt/flink/flink-cdc-3.5.0/conf
            {{- if .Values.truststore.enabled }}
            - name: truststore
              mountPath: /opt/flink/certs
              readOnly: true
            {{- end }}
            {{- if eq .Values.checkpointStorage.storageType "gcs" }}
            - name: gcs-key
              mountPath: /opt/flink/gcs-key
              readOnly: true
            {{- end }}
      volumes:
        - name: pipeline-config
          configMap:
            name: {{ include "flink-cdc-session.fullname" . }}-pipeline-config
        {{- if .Values.truststore.enabled }}
        - name: truststore
          secret:
            secretName: {{ include "flink-cdc-session.truststoreSecretName" . }}
        {{- end }}
        {{- if eq .Values.checkpointStorage.storageType "gcs" }}
        - name: gcs-key
          secret:
            secretName: {{ .Values.checkpointStorage.gcs.credentialSecret }}
        {{- end }}