feat(helm): add filesystem checkpoint storage mode

helm/chart/templates/flink/config.yaml

@@ -18,9 +18,13 @@
     high-availability.type: kubernetes
     kubernetes.namespace: {{ .Release.Namespace }}
     kubernetes.cluster-id: {{ .Values.clusterId | default (print .Release.Name "-cluster") }}
-    execution.checkpointing.interval: {{ .Values.flink.checkpoint.interval }}
-    execution.checkpointing.mode: {{ .Values.flink.checkpoint.mode }}
+    execution.checkpointing.interval: {{ .Values.flink.state.checkpoint.interval }}
+    execution.checkpointing.mode: {{ .Values.flink.state.checkpoint.mode }}
+    {{- if eq .Values.flink.state.checkpoint.storageType "filesystem" }}
+    state.checkpoints.dir: /opt/flink/checkpoints/
+    {{- else if eq .Values.flink.state.checkpoint.storageType "s3" }}
     state.checkpoints.dir: s3://flink/checkpoints/
+    {{- end }}
     state.backend.rocksdb.localdir: /opt/flink/rocksdb
     high-availability.storageDir: /opt/flink/ha
     state.savepoints.dir: s3://flink/savepoints/

helm/chart/templates/flink/ha.pvc.yaml

@@ -1,7 +1,7 @@
 apiVersion: v1
 kind: PersistentVolumeClaim
 metadata:
-  name: {{ .Release.Name }}-{{ .Values.flink.state.ha.pvcName }}
+  name: {{ .Release.Name }}-flink-ha-pvc
 spec:
   accessModes:
     - ReadWriteOnce

helm/chart/templates/flink/job-manager-deploy.yaml

@@ -64,12 +64,19 @@ spec:
           volumeMounts:
             - name: flink-ha
               mountPath: {{ .Values.flink.state.ha.dir }}
-
+            {{- if eq .Values.flink.state.checkpoint.storageType "filesystem" }}
+            - name: flink-checkpoint
+              mountPath: /opt/flink/checkpoints
+            {{- end }}
       volumes:
         - name: flink-ha
           persistentVolumeClaim:
-            claimName: {{ .Release.Name }}-{{ .Values.flink.state.ha.pvcName }}
-
+            claimName: {{ .Release.Name }}-flink-ha-pvc
+        {{- if eq .Values.flink.state.checkpoint.storageType "filesystem" }}
+        - name: flink-checkpoint
+          persistentVolumeClaim:
+            claimName: {{ .Release.Name }}-flink-checkpoint-pvc
+        {{- end }}
       {{- with .Values.nodeSelector }}
       nodeSelector:
         {{- toYaml . | nindent 8 }}

helm/chart/templates/flink/savepoint.pvc.yaml

@@ -0,0 +1,10 @@
+apiVersion: v1
+kind: PersistentVolumeClaim
+metadata:
+  name: {{ .Release.Name }}-flink-checkpoint-pvc
+spec:
+  accessModes:
+    - ReadWriteMany
+  resources:
+    requests:
+      storage: {{ .Values.flink.state.checkpoint.size }}

helm/chart/templates/flink/task-manager-statefulset.yaml

@@ -46,8 +46,18 @@ spec:
         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