{{- if .Values.job.enabled }}
apiVersion: batch/v1
kind: Job
metadata:
  name:  {{ include "flink-cdc-session.fullname" . }}-{{ .Values.job.name }}
  namespace: {{ .Release.Namespace }}
  labels:
    {{- include "flink-cdc-session.labels" . | nindent 4 }}
spec:
  backoffLimit: 0
  template:
    spec:
      serviceAccountName: {{ include "flink-cdc-session.serviceAccountName" . }}
      containers:
        - name: submit
          image: {{ .Values.image.repository }}:{{ .Values.image.tag }}
          {{- 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 }}
          command: ["/bin/sh", "-c"]
          args:
            - |
              cd /opt/flink/flink-cdc-3.5.0
              ./bin/flink-cdc.sh \
                {{ .Values.job.pipelineFile }} \
                -Dexecution.checkpointing.interval={{ .Values.pipeline.checkpointInterval }} \
                -Drest.address={{ .Values.job.restAddress }} \
                -Drest.port={{ .Values.job.restPort }}
          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 }}
      restartPolicy: Never
{{- end }}