feat(helm): add filesystem savepoint storage mode

helm/chart/templates/flink/checkpoint-pvc.yaml

@@ -1,3 +1,4 @@
+{{- if eq .Values.flink.state.checkpoint.storageType "filesystem" }}
 apiVersion: v1
 kind: PersistentVolumeClaim
 metadata:
@@ -8,3 +9,4 @@ spec:
   resources:
     requests:
       storage: {{ .Values.flink.state.checkpoint.size }}
+{{- end }}

helm/chart/templates/flink/config.yaml

@@ -27,7 +27,11 @@
     {{- end }}
     state.backend.rocksdb.localdir: /opt/flink/rocksdb
     high-availability.storageDir: /opt/flink/ha
+    {{- if eq .Values.flink.state.savepoint.storageType "filesystem" }}
+    state.savepoints.dir: file:///opt/flink/checkpoints/
+    {{- else if eq .Values.flink.state.savepoint.storageType "s3" }}
     state.savepoints.dir: s3://flink/savepoints/
+    {{- end }}
     state.backend.incremental: {{ .Values.flink.state.incremental }}
     rest.profiling.enabled: true
     s3.endpoint: http://{{ .Release.Name }}-minio:9000

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

@@ -68,6 +68,10 @@ spec:
             - name: flink-checkpoint
               mountPath: /opt/flink/checkpoints
             {{- end }}
+            {{- if eq .Values.flink.state.savepoint.storageType "filesystem" }}
+            - name: flink-savepoint
+              mountPath: /opt/flink/savepoint
+            {{- end }}
       volumes:
         - name: flink-ha
           persistentVolumeClaim:
@@ -77,6 +81,11 @@ spec:
           persistentVolumeClaim:
             claimName: {{ .Release.Name }}-flink-checkpoint-pvc
         {{- end }}
+        {{- if eq .Values.flink.state.savepoint.storageType "filesystem" }}
+        - name: flink-savepoint
+          persistentVolumeClaim:
+            claimName: {{ .Release.Name }}-flink-savepoint-pvc
+        {{- end }}
       {{- with .Values.nodeSelector }}
       nodeSelector:
         {{- toYaml . | nindent 8 }}

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

@@ -0,0 +1,12 @@
+{{- if eq .Values.flink.state.savepoint.storageType "filesystem" }}
+apiVersion: v1
+kind: PersistentVolumeClaim
+metadata:
+  name: {{ .Release.Name }}-flink-savepoint-pvc
+spec:
+  accessModes:
+    - ReadWriteMany
+  resources:
+    requests:
+      storage: {{ .Values.flink.state.savepoint.size }}
+{{- end }}