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
          livenessProbe:
            httpGet:
              path: /
              port: 8081
          readinessProbe:
            httpGet:
              path: /
              port: 8081
          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 }}
                rest.port: 8081
                rootLogger.level = DEBUG
                rootLogger.appenderRef.console.ref = ConsoleAppender
                high-availability.type: kubernetes
                kubernetes.namespace: {{ .Release.Namespace }}
                kubernetes.cluster-id: cluster-one
                web.upload.dir: {{ .Values.flink.state.data.dir }}/web-upload
                state.checkpoints.dir: file://{{ .Values.flink.state.data.dir }}/checkpoints
                state.backend.rocksdb.localdir: file://{{ .Values.flink.state.data.dir }}/rocksdb
                high-availability.storageDir: file://{{ .Values.flink.state.ha.dir }}
                state.savepoints.dir: file://{{ .Values.flink.state.savepoints.dir }}
          volumeMounts:
            - name: flink-data
              mountPath: {{ .Values.flink.state.data.dir }}/data
            - name: flink-data
              mountPath: {{ .Values.flink.state.data.dir }}/rocksdb
              subPath: rocksdb
            - name: flink-data
              mountPath: {{ .Values.flink.state.data.dir }}/checkpoints
              subPath: checkpoints
            - name: flink-data
              mountPath: {{ .Values.flink.state.data.dir }}/web-upload
              subPath: web-upload
            - name: flink-ha
              mountPath: {{ .Values.flink.state.ha.dir }}
            - name: flink-savepoints
              mountPath: {{ .Values.flink.state.savepoints.dir }}


      volumes:
        - name: flink-data
          persistentVolumeClaim:
            claimName: {{ .Values.flink.state.data.pvcName }}  # PVC for savepoints persistence
        - name: flink-savepoints
          persistentVolumeClaim:
            claimName: {{ .Values.flink.state.savepoints.pvcName }}  # PVC for savepoints persistence
        - name: flink-ha
          persistentVolumeClaim:
            claimName: {{ .Values.flink.state.ha.pvcName }}  # PVC for savepoints persistence