apiVersion: apps/v1
kind: Deployment
metadata:
  name: {{ .Release.Name }}-flink
  labels:
    app.kubernetes.io/name: {{ .Release.Name }}-flink
    app.kubernetes.io/instance: {{ .Release.Name }}
spec:
  replicas: 1
  selector:
    matchLabels:
      app.kubernetes.io/name: {{ .Release.Name }}-flink
      app.kubernetes.io/instance: {{ .Release.Name }}
  template:
    metadata:
      labels:
        app.kubernetes.io/name: {{ .Release.Name }}-flink
        app.kubernetes.io/instance: {{ .Release.Name }}
    spec:
      serviceAccountName: {{ include "flink-kube-operator.serviceAccountName" . }}

      containers:
        - name: flink
          image: {{ .Values.flink.image.repository }}:{{ .Values.flink.image.tag }}
          imagePullPolicy: Always
          ports:
            - containerPort: 8081  # JobManager Web UI port
            - containerPort: 6121  # TaskManager communication port
            - containerPort: 6122  # TaskManager communication port
          env:
            - name: JOB_MANAGER_RPC_ADDRESS
              value: "localhost"  # JobManager and TaskManager in the same container
            - name: FLINK_PROPERTIES
              value: |
                jobmanager.rpc.address: localhost
                jobmanager.memory.process.size: 2048m
                taskmanager.memory.process.size: 2048m
                taskmanager.data.port: 6125
                taskmanager.numberOfTaskSlots: {{ .Values.flink.taskManager.numberOfTaskSlots }}
                parallelism.default: {{ .Values.flink.parallelism.default }}
                state.backend: {{ .Values.flink.state.backend }}
                state.savepoints.dir: {{ .Values.flink.state.savepoints.dir }}
                rest.port: 8081
                rootLogger.level = DEBUG
                rootLogger.appenderRef.console.ref = ConsoleAppender
                web.upload.dir: /opt/flink/data/web-upload
                state.checkpoints.dir: file:///tmp/flink-checkpoints
                high-availability.type: kubernetes
                high-availability.storageDir: file:///opt/flink/ha
                kubernetes.cluster-id: cluster-one
                kubernetes.namespace: {{ .Release.Namespace }}
          volumeMounts:
            - name: flink-data
              mountPath: /opt/flink/data
              subPath: data
            - name: flink-savepoints
              mountPath: /opt/flink/savepoints
            - name: flink-savepoints
              mountPath: /opt/flink/ha
              subPath: ha

      volumes:
        - name: flink-data
          emptyDir: {}  # Temporary storage for internal data
        - name: flink-savepoints
          persistentVolumeClaim:
            claimName: {{ .Values.flink.state.savepoints.pvcName }}  # PVC for savepoints persistence