---
apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  name: flow
spec:
  image: local.azurecr.io/flink:2.1.1
  imagePullPolicy: Always
  flinkVersion: v2_1
  flinkConfiguration:
    # JobManager settings
    jobmanager.rpc.port: 6123
    jobmanager.bind-host: 0.0.0.0
    jobmanager.memory.process.size: 1024m
    jobmanager.memory.jvm-metaspace.size: 384m
    jobmanager.memory.enable-jvm-direct-memory-limit: true

    # TaskManager settings
    taskmanager.network.detailed-metrics: false
    taskmanager.bind-host: 0.0.0.0
    taskmanager.numberOfTaskSlots: 8
    taskmanager.memory.process.size: 7168m
    taskmanager.memory.task.heap.size: 1536mb
    taskmanager.memory.managed.fraction: 0.30
    taskmanager.memory.network.fraction: 0.15

    # Execution settings
    pipeline.max-parallelism: 256
    parallelism.default: 8
    execution.buffer-timeout.interval: 10ms
    pipeline.object-reuse: true
    execution.checkpointing.mode: EXACTLY_ONCE

    # Checkpoint interval must be less than Kafka max.poll.interval.ms (30 min)
    # Set to 3 minutes for high-throughput processing
    # (must be < max.poll.interval.ms)
    execution.checkpointing.interval: 3min
    execution.checkpointing.timeout: 8min

    # Minimum pause between checkpoints prevents rapid consecutive checkpoints
    # Must be less than checkpoint interval
    execution.checkpointing.min-pause: 1min

    # Unaligned checkpoints for high throughput
    execution.checkpointing.unaligned.enabled: true
    execution.checkpointing.unaligned.max-subtask-alignment-time: 30s
    execution.checkpointing.aligned-checkpoint-timeout: 30s

    # Tolerate checkpoint failures without failing the job (for transient issues)
    execution.checkpointing.tolerable-failed-checkpoints: 3
    execution.checkpointing.max-concurrent-checkpoints: 1
    execution.checkpointing.num-retained: 1
    execution.checkpointing.incremental: true
    execution.checkpointing.storage: filesystem
    execution.checkpointing.dir: file:///data/checkpoints
    execution.checkpointing.savepoint-dir: file:///data/savepoint
    execution.checkpointing.snapshot-compression: true
    # In K8s with PVCs, recovery happens automatically from checkpoint storage
    # execution.state-recovery.from-local: true
    # execution.state-recovery.ignore-unclaimed-state: true
    # execution.state-recovery.path: file:///data/savepoint

    # State backend settings
    state.backend.type: rocksdb
    state.backend.rocksdb.block.cache-size: 192m
    state.backend.rocksdb.writebuffer.size: 48m
    state.backend.rocksdb.writebuffer.count: 4
    state.backend.rocksdb.predefined-options: FLASH_SSD_OPTIMIZED
    state.backend.incremental: true
    state.backend.rocksdb.thread.num: 4
    state.backend.rocksdb.memory.write-buffer-ratio: 0.5
    state.backend.rocksdb.memory.high-prio-pool-ratio: 0.1
    state.backend.rocksdb.files.open: -1
    state.backend.rocksdb.cache-index-and-filter-blocks: true
    state.backend.rocksdb.use-partitioned-index-filters: true
    state.backend.rocksdb.pin-l0-filter-and-index-blocks: true
    state.backend.rocksdb.metrics.background-errors: true
    state.backend.rocksdb.metrics.estimate-live-data-size: true
    state.backend.rocksdb.metrics.estimate-pending-compaction-bytes: true
    state.backend.rocksdb.metrics.num-running-compactions: true
    state.backend.rocksdb.metrics.compaction-pending: true
    state.backend.rocksdb.metrics.is-write-stopped: true
    state.backend.rocksdb.metrics.num-running-flushes: true
    state.backend.rocksdb.metrics.mem-table-flush-pending: true
    # Standard checkpointing (more forgiving than changelog)
    state.changelog.enabled: false
    state.changelog.storage: filesystem
    state.changelog.dstl.dfs.base-path: file:///data/checkpoints/changelog

    # 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

    # Kafka connector settings
    table.exec.source.idle-timeout: 10s

    # Network buffer optimization for high-throughput Kafka streams
    taskmanager.network.memory.buffer-debloat.enabled: true
    taskmanager.network.memory.buffer-debloat.period: 1s
    taskmanager.network.memory.buffer-debloat.target: 1s
    taskmanager.network.memory.buffers-per-channel: 4
    taskmanager.network.memory.floating-buffers-per-gate: 128
    taskmanager.network.memory.max-buffers-per-channel: 16
    taskmanager.network.request-backoff.initial: 100
    taskmanager.network.request-backoff.max: 10000
    taskmanager.network.blocking-shuffle.type: mmap
    taskmanager.network.sort-shuffle.min-parallelism: 1

    # Observability
    rest.address: jobmanager
    rest.port: 8081
    rest.flamegraph.enabled: true
    rest.profiling.enabled: true
    metrics.reporters: prometheus
    metrics.reporter.prometheus.factory.class: org.apache.flink.metrics.prometheus.PrometheusReporterFactory
    metrics.reporter.prometheus.port: 9249
    metrics.reporter.prometheus.host: 0.0.0.0
    metrics.system-resource: true

    # Python execution optimizations
    env.java.opts.taskmanager: -Dpython.fn-execution.memory.managed.size=1024m
    python.execution-mode: multi_thread
    python.operator-chaining.enabled: true
    python.state.cache-size: 25000
    python.fn-execution.bundle.size: 2500
    python.fn-execution.bundle.time: 2000
    python.fn-execution.arrow.batch.size: 5000

    high-availability.type: kubernetes
    high-availability.storageDir: file:///data/ha
    # high-availability.vvp-kubernetes.job-graph-store.enabled: true

    # Blob storage configuration (for log files and artifacts)
    blob.storage.directory: /data/blob
    blob.server.port: 6130

    # k8s AutoScaling
    jobmanager.scheduler: adaptive
    jobmanager.adaptive-scheduler.executing.resource-stabilization-timeout: 15s
    jobmanager.adaptive-scheduler.executing.cooldown-after-rescaling: 30s
    job.autoscaler.memory.tuning.enabled: true
    job.autoscaler.memory.tuning.maximize-managed-memory: false
    job.autoscaler.memory.tuning.overhead: 0.3
    job.autoscaler.enabled: true
    job.autoscaler.scaling.enabled: true
    # job.autoscaler.observed-scalability.enabled: true
    # Asymmetric scaling: aggressive scale-up, conservative scale-down
    job.autoscaler.stabilization.interval: 5m
    job.autoscaler.metrics.window: 3m
    job.autoscaler.target.utilization: 0.70
    job.autoscaler.target.utilization.boundary: 0.15
    job.autoscaler.restart.time: 1m
    job.autoscaler.restart.time-tracking.enabled: true
    job.autoscaler.catch-up.duration: 10m
    job.autoscaler.scale-down.max-factor: 0.5
    job.autoscaler.scale-down.interval: 10m
  serviceAccount: flink
  jobManager:
    replicas: 1
    resource:
      cpu: 1
      memory: 1536m
    podTemplate:
      spec:
        containers:
          - name: flink-main-container
            env:
              - name: TZ
                value: Europe/Paris
              - name: ROOT_LOG_LEVEL
                value: WARN
            volumeMounts:
              - name: flink-checkpoints
                mountPath: /data/checkpoints
              - name: flink-savepoint
                mountPath: /data/savepoint
              - name: flink-ha
                mountPath: /data/ha
              - name: flink-blob
                mountPath: /data/blob
              - name: flink-logs
                mountPath: /opt/flink/log
        volumes:
          - name: flink-checkpoints
            persistentVolumeClaim:
              claimName: flink-checkpoints-flow
          - name: flink-savepoint
            persistentVolumeClaim:
              claimName: flink-savepoint-flow
          - name: flink-ha
            persistentVolumeClaim:
              claimName: flink-ha-flow
          - name: flink-blob
            emptyDir: {}
          - name: flink-logs
            emptyDir: {}
  taskManager:
    resource:
      cpu: 3
      memory: 8Gi
    podTemplate:
      spec:
        containers:
          - name: flink-main-container
            # TODO: Investigate not working
            resources:
              limits:
                cpu: 3.5
                memory: "12Gi"
            env:
              - name: TZ
                value: Europe/Paris
              - name: ROOT_LOG_LEVEL
                value: WARN
            ports:
              - name: metrics
                containerPort: 9249
            volumeMounts:
              - name: flink-checkpoints
                mountPath: /data/checkpoints
              - name: flink-savepoint
                mountPath: /data/savepoint
              - name: flink-ha
                mountPath: /data/ha
                # readOnly: true
              - name: flink-blob
                mountPath: /data/blob
        volumes:
          - name: flink-checkpoints
            persistentVolumeClaim:
              claimName: flink-checkpoints-flow
          - name: flink-savepoint
            persistentVolumeClaim:
              claimName: flink-savepoint-flow
          - name: flink-ha
            persistentVolumeClaim:
              claimName: flink-ha-flow
              # readOnly: true
          - name: flink-blob
            emptyDir: {}
  job:
    entryClass: "org.apache.flink.client.python.PythonDriver"
    args:
      ["-pyclientexec", "/usr/bin/python3.12", "-py", "/jobs/processor_flow.py"]
    parallelism: 8
    upgradeMode: last-state
    state: running
  podTemplate:
    spec:
      containers:
        - name: flink-main-container
          # TODO: Investigate not working
          resources:
            limits:
              cpu: 3.5
              memory: "12Gi"
          env:
            - name: TZ
              value: Europe/Paris
          ports:
            - name: metrics
              containerPort: 9249
