apiVersion: sparkoperator.k8s.io/v1beta2
kind: SparkApplication
metadata:
  name: cdc-{{ .Values.customer }}
  namespace: {{ .Release.Namespace }}
spec:
  type: Python
  mode: cluster
  image: "{{ .Values.image.repository }}:{{ .Values.image.tag }}"
  imagePullPolicy: {{ .Values.image.pullPolicy }}
  imagePullSecrets:
    - {{ .Values.imagePullSecret }}
  mainApplicationFile: "local:///opt/spark/work-dir/cdc_crm_to_raw.py"
  sparkVersion: "{{ .Values.spark.version }}"
  # streaming job chạy mãi -> restart nếu chết (thay cho việc submit tay foreground)
  restartPolicy:
    type: Always
    onFailureRetryInterval: 30
    onSubmissionFailureRetryInterval: 30
  sparkConf:
    # Iceberg catalog (Polaris REST) + MinIO
    "spark.sql.extensions": "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions"
    "spark.sql.catalog.iceberg": "org.apache.iceberg.spark.SparkCatalog"
    "spark.sql.catalog.iceberg.type": "rest"
    "spark.sql.catalog.iceberg.uri": "{{ .Values.catalog.uri }}"
    "spark.sql.catalog.iceberg.warehouse": "{{ .Values.catalog.warehouse }}"
    "spark.sql.catalog.iceberg.header.X-Iceberg-Access-Delegation": "vended-credentials"
    "spark.sql.catalog.iceberg.io-impl": "org.apache.iceberg.aws.s3.S3FileIO"
    "spark.sql.catalog.iceberg.s3.endpoint": "{{ .Values.s3.endpoint }}"
    "spark.sql.catalog.iceberg.s3.path-style-access": "true"
    "spark.sql.catalog.iceberg.s3.region": "{{ .Values.s3.region }}"
    "spark.sql.defaultCatalog": "iceberg"
  driver:
    cores: {{ .Values.spark.driverCores }}
    memory: "{{ .Values.spark.driverMemory }}"
    labels: { app: cdc-{{ .Values.customer }} }
    env:
      - name: KAFKA_BOOTSTRAP
        value: "{{ .Values.kafka.bootstrap }}"
      - name: TOPIC_PATTERN
        value: "{{ .Values.topicPattern }}"
      - name: TARGET_TABLE
        value: "{{ .Values.targetTable }}"
      - name: TRIGGER_INTERVAL
        value: "{{ .Values.triggerInterval }}"
      - name: MAX_OFFSETS_PER_TRIGGER
        value: "{{ .Values.maxOffsetsPerTrigger }}"
      - name: CHECKPOINT
        value: "file:///checkpoints/cdc_{{ .Values.customer }}"
    volumeMounts:
      - name: checkpoint
        mountPath: /checkpoints
  executor:
    instances: {{ .Values.spark.executorInstances }}
    cores: {{ .Values.spark.executorCores }}
    memory: "{{ .Values.spark.executorMemory }}"
    volumeMounts:
      - name: checkpoint
        mountPath: /checkpoints
  volumes:
    - name: checkpoint
      persistentVolumeClaim:
        claimName: cdc-{{ .Values.customer }}-checkpoint
  deps:
    packages:
      - org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.9.1
      - org.apache.iceberg:iceberg-aws-bundle:1.9.1
      - org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.6