helm/chart/Chart.yaml

@@ -2,5 +2,5 @@ apiVersion: v2
 name: flink-kube-operator
 description: Helm chart for flink kube operator
 type: application
-version: 0.1.10
+version: 0.1.9
 appVersion: "0.1.0"

helm/chart/templates/flink/deploy.yaml

@@ -18,15 +18,13 @@
     high-availability.type: kubernetes
     kubernetes.namespace: {{ .Release.Namespace }}
     kubernetes.cluster-id: cluster-one
-    execution.checkpointing.interval: 5min
-    execution.checkpointing.mode: EXACTLY_ONCE
     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 }}
     state.backend.incremental: false
-    rest.profiling.enabled: true 
+    rest.profiling.enabled: true
 {{- end }}
 
 {{- define "flink.volumeMounts" -}}

internal/managed_job/savepoint.go

@@ -89,7 +89,3 @@ func (job ManagedJob) TriggerSavepoint() error {
 	err = job.trackSavepoint()
 	return err
 }
-
-func (job ManagedJob) GetLastSavepointPath() *string {
-	return job.def.Status.LastSavepointPath
-}

internal/rest/controller/crd.go

@@ -14,7 +14,6 @@ import (
 	"io"
 	"os"
 	"path/filepath"
-	"strings"
 
 	"github.com/danielgtaylor/huma/v2"
 	"go.uber.org/zap"
@@ -129,14 +128,7 @@ func TriggerSavepoint(ctx context.Context, req *JobTriggerSavepointReq) (*JobTri
 }
 
 func DownloadSavepoint(ctx context.Context, req *types.SavepointDownloadReq) (*huma.StreamResponse, error) {
-	mgr := managed_job.GetManager()
+	folderPath := req.SavepointPath // Change this to your folder path
-	job := mgr.GetJob(k8sTypes.UID(req.JobUID))
-	lastSavepointPath := job.GetLastSavepointPath()
-
-	if lastSavepointPath == nil {
-		return nil, huma.Error404NotFound("there is no savepoint path is registered for the job")
-	}
-	folderPath := strings.TrimLeft(*lastSavepointPath, "file:") // Change this to your folder path
 
 	// Create a temporary zip file
 	zipFilePath, err := filepath.Abs("./savepoint.zip")

internal/rest/types/savepoint.go

@@ -1,7 +1,7 @@
 package types
 
 type SavepointDownloadReq struct {
-	JobUID string `query:"jobUID"`
+	SavepointPath string `query:"savepoint-path"`
 }
 
 type SavepointDownloadResp struct {