apiVersion: apps/v1
kind: Deployment
metadata:
  name: {{ .Release.Name }}-flink  # Adding the flink prefix to the name
  labels:
    app.kubernetes.io/name: {{ .Release.Name }}-flink  # Adding the flink prefix to the labels
    app.kubernetes.io/instance: {{ .Release.Name }}  # Using the release name for instance
    app.kubernetes.io/managed-by: Helm
spec:
  replicas: 1
  selector:
    matchLabels:
      app.kubernetes.io/name: {{ .Release.Name }}-flink  # Adding the flink prefix to the selector
      app.kubernetes.io/instance: {{ .Release.Name }}
  template:
    metadata:
      labels:
        app.kubernetes.io/name: {{ .Release.Name }}-flink  # Adding the flink prefix to the template labels
        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 }}  # Savepoints path
                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
          volumeMounts:
            - name: flink-data
              mountPath: /opt/flink/data
              subPath: data
            - name: flink-data
              mountPath: /opt/flink/web-upload
              subPath: web-upload
            - name: flink-savepoints
              mountPath: /opt/flink/savepoints

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