apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: {{ .Release.Name }}-flink-task-manager
  labels:
    app: {{ .Release.Name }}-flink-operator
    component: taskmanager
spec:
  serviceName: {{ .Release.Name }}-flink-task-manager
  replicas: {{ .Values.flink.taskManager.replicas }}
  selector:
    matchLabels:
      app: {{ .Release.Name }}-flink-operator
      component: {{ .Release.Name }}-flink-task-manager
  template:
    metadata:
      labels:
        app: {{ .Release.Name }}-flink-operator
        component: {{ .Release.Name }}-flink-task-manager
    spec:
      serviceAccountName: {{ include "flink-kube-operator.serviceAccountName" . }}
      containers:
      - name: task-manager
        image: {{ .Values.flink.image.repository }}:{{ .Values.flink.image.tag }}
        imagePullPolicy: Always
        args: ["taskmanager"]
        env:
        {{- include "flink.env" . | nindent 8 }}
        - name: POD_IP
          valueFrom:
            fieldRef:
              fieldPath: status.podIP
        - name: S3_ENDPOINT
          value: "http://minio-service:9000"
        - name: AWS_ACCESS_KEY_ID
          valueFrom:
            secretKeyRef:
              name: {{ .Release.Name }}-flink-secrets
              key: minio_access_key
        - name: AWS_SECRET_ACCESS_KEY
          valueFrom:
            secretKeyRef:
              name: {{ .Release.Name }}-flink-secrets
              key: minio_secret_key
        volumeMounts:
        - name: rocksdb-storage
          mountPath: /opt/flink/rocksdb
        {{- if eq .Values.flink.state.checkpoint.storageType "filesystem" }}
        - name: flink-checkpoint
          mountPath: /opt/flink/checkpoints
        {{- end }}
        resources:
          {{- toYaml .Values.flink.taskManager.resources | nindent 10 }}
      {{- if eq .Values.flink.state.checkpoint.storageType "filesystem" }}
      volumes:
        - name: flink-checkpoint
          persistentVolumeClaim:
            claimName: {{ .Release.Name }}-flink-checkpoint-pvc
      {{- end }}
  volumeClaimTemplates:
  - metadata:
      name: rocksdb-storage
    spec:
      accessModes: [ ReadWriteOnce ]
      resources:
        requests:
          storage: {{ .Values.flink.taskManager.storage.rocksDb.size }}