apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  name: ha-flink-cluster-v2
  namespace: freeleaps-data-platform
  labels:
    app: flink
    component: streaming
    cluster-type: ha
spec:
  flinkVersion: v1_19
  image: flink:1.19.0
  flinkConfiguration:
    # High Availability Configuration
    high-availability.type: kubernetes
    high-availability.storageDir: file:///opt/flink/ha-data
    # Checkpointing Configuration
    state.backend.type: filesystem
    state.checkpoints.dir: file:///opt/flink/checkpoints
    state.savepoints.dir: file:///opt/flink/savepoints
    execution.checkpointing.interval: 60s
    execution.checkpointing.min-pause: 5s
    execution.checkpointing.timeout: 10min
    # JobManager Configuration
    jobmanager.rpc.address: ha-flink-cluster-v2-jobmanager
    jobmanager.rpc.port: "6123"
    jobmanager.bind-host: "0.0.0.0"
    # REST Configuration
    rest.address: ha-flink-cluster-v2-jobmanager
    rest.port: "8081"
    rest.bind-address: "0.0.0.0"
    # Blob Server Configuration
    blob.server.port: "6124"
    # TaskManager Configuration
    taskmanager.numberOfTaskSlots: "2"
    # Memory Configuration
    taskmanager.memory.process.size: 2048m
    jobmanager.memory.process.size: 1024m
    # Restart Strategy
    restart-strategy.type: exponential-delay
    restart-strategy.exponential-delay.initial-backoff: 10s
    restart-strategy.exponential-delay.max-backoff: 2min
    restart-strategy.exponential-delay.backoff-multiplier: "2.0"
    restart-strategy.exponential-delay.reset-backoff-threshold: 10min
    restart-strategy.exponential-delay.jitter-factor: "0.1"
  serviceAccount: flink
  jobManager:
    replicas: 2
    resource:
      memory: "1024m"
      cpu: 0.5
    podTemplate:
      spec:
        containers:
        - name: flink-main-container
          volumeMounts:
          - name: ha-data
            mountPath: /opt/flink/ha-data
          - name: checkpoints
            mountPath: /opt/flink/checkpoints
          - name: savepoints
            mountPath: /opt/flink/savepoints
        volumes:
        - name: ha-data
          persistentVolumeClaim:
            claimName: ha-flink-ha-data
        - name: checkpoints
          persistentVolumeClaim:
            claimName: ha-flink-checkpoints
        - name: savepoints
          persistentVolumeClaim:
            claimName: ha-flink-savepoints
  taskManager:
    replicas: 3
    resource:
      memory: "2048m"
      cpu: 0.5
    podTemplate:
      spec:
        containers:
        - name: flink-main-container
          volumeMounts:
          - name: checkpoints
            mountPath: /opt/flink/checkpoints
          - name: savepoints
            mountPath: /opt/flink/savepoints
        volumes:
        - name: checkpoints
          persistentVolumeClaim:
            claimName: ha-flink-checkpoints
        - name: savepoints
          persistentVolumeClaim:
            claimName: ha-flink-savepoints