feat: Allow specifying savepoint format type

apis/flinkcluster/v1beta1/assets/test/flinkcluster_type_expected.yaml

@@ -40,6 +40,7 @@ spec:
   job:
     jarFile: ./examples/streaming/StateMachineExample.jar
     allowNonRestoredState: false
+    savepointFormatType: CANONICAL
     noLoggingToStdout: false
     restartPolicy: Never
     cleanupPolicy:

apis/flinkcluster/v1beta1/flinkcluster_types.go

@@ -137,6 +137,15 @@ const (
 	SavepointReasonUpdate        SavepointReason = "update"
 )
 
+// SavepointFormatType specifies the binary format of a savepoint.
+// +kubebuilder:validation:Enum=CANONICAL;NATIVE
+type SavepointFormatType string
+
+const (
+	SavepointFormatTypeCanonical SavepointFormatType = "CANONICAL"
+	SavepointFormatTypeNative    SavepointFormatType = "NATIVE"
+)
+
 // ImageSpec defines Flink image of JobManager and TaskManager containers.
 type ImageSpec struct {
 	// Flink image name.
@@ -568,6 +577,11 @@ type JobSpec struct {
 	// _(Optional)_ Savepoints dir where to store savepoints of the job.
 	SavepointsDir *string `json:"savepointsDir,omitempty"`
 
+	// _(Optional)_ Savepoint format type, "CANONICAL" or "NATIVE", default: "CANONICAL".
+	// +kubebuilder:validation:Enum=CANONICAL;NATIVE
+	// +kubebuilder:default:=CANONICAL
+	SavepointFormatType *SavepointFormatType `json:"savepointFormatType,omitempty"`
+
 	// _(Optional)_ Should take savepoint before updating job, default: `true`.
 	// If this is set as false, maxStateAgeToRestoreSeconds must be provided to limit the savepoint age to restore.
 	TakeSavepointOnUpdate *bool `json:"takeSavepointOnUpdate,omitempty"`
@@ -946,6 +960,9 @@ type SavepointStatus struct {
 	// Savepoint state.
 	State string `json:"state"`
 
+	// Savepoint format type, "CANONICAL" or "NATIVE".
+	FormatType string `json:"formatType,omitempty"`
+
 	// Savepoint message.
 	Message string `json:"message,omitempty"`
 }

config/crd/bases/flinkoperator.k8s.io_flinkclusters.yaml

@@ -1539,6 +1539,16 @@ spec:
                         - Never
                         - FromSavepointOnFailure
                       type: string
+                    savepointFormatType:
+                      allOf:
+                        - enum:
+                            - CANONICAL
+                            - NATIVE
+                        - enum:
+                            - CANONICAL
+                            - NATIVE
+                      default: CANONICAL
+                      type: string
                     savepointGeneration:
                       format: int32
                       type: integer
@@ -9717,6 +9727,8 @@ spec:
                   type: object
                 savepoint:
                   properties:
+                    formatType:
+                      type: string
                     jobID:
                       type: string
                     message:

controllers/flinkcluster/flinkcluster_converter.go

@@ -70,8 +70,9 @@ var (
 		"query.server.port":      {},
 		"rest.port":              {},
 	}
-	v10, _ = version.NewVersion("1.10")
-	v20, _ = version.NewVersion("2.0")
+	v10, _  = version.NewVersion("1.10")
+	v115, _ = version.NewVersion("1.15")
+	v20, _  = version.NewVersion("2.0")
 )
 
 // Gets the desired state of a cluster.

controllers/flinkcluster/flinkcluster_reconciler.go

@@ -29,6 +29,7 @@ import (
 	"k8s.io/client-go/util/retry"
 
 	"github.com/go-logr/logr"
+	"github.com/hashicorp/go-version"
 	v1beta1 "github.com/spotify/flink-on-k8s-operator/apis/flinkcluster/v1beta1"
 	"github.com/spotify/flink-on-k8s-operator/internal/batchscheduler"
 	schedulerTypes "github.com/spotify/flink-on-k8s-operator/internal/batchscheduler/types"
@@ -812,10 +813,10 @@ func (reconciler *ClusterReconciler) triggerSavepoint(
 	var savepointTriggerID *flink.SavepointTriggerID
 	var triggerID string
 	var message string
+	var formatType = savepointFormatType(cluster)
 	var err error
-
 	log.Info(fmt.Sprintf("Trigger savepoint for %s", triggerReason), "jobID", jobID)
-	savepointTriggerID, err = reconciler.flinkClient.TriggerSavepoint(apiBaseURL, jobID, *cluster.Spec.Job.SavepointsDir, cancel)
+	savepointTriggerID, err = reconciler.flinkClient.TriggerSavepoint(apiBaseURL, jobID, *cluster.Spec.Job.SavepointsDir, cancel, formatType)
 	if err != nil {
 		// limit message size to 1KiB
 		if message = err.Error(); len(message) > 1024 {
@@ -828,7 +829,7 @@ func (reconciler *ClusterReconciler) triggerSavepoint(
 		triggerID = savepointTriggerID.RequestID
 		log.Info("Successfully savepoint triggered", "jobID", jobID, "triggerID", triggerID)
 	}
-	newSavepointStatus := reconciler.getNewSavepointStatus(triggerID, triggerReason, message, triggerSuccess)
+	newSavepointStatus := reconciler.getNewSavepointStatus(triggerID, triggerReason, message, triggerSuccess, formatType)
 
 	return newSavepointStatus, err
 }
@@ -839,7 +840,7 @@ func (reconciler *ClusterReconciler) takeSavepoint(ctx context.Context, jobID st
 	apiBaseURL := getFlinkAPIBaseURL(reconciler.observed.cluster)
 
 	log.Info("Taking savepoint.", "jobID", jobID)
-	status, err := reconciler.flinkClient.TakeSavepoint(apiBaseURL, jobID, *reconciler.observed.cluster.Spec.Job.SavepointsDir)
+	status, err := reconciler.flinkClient.TakeSavepoint(apiBaseURL, jobID, *reconciler.observed.cluster.Spec.Job.SavepointsDir, savepointFormatType(reconciler.observed.cluster))
 	log.Info("Savepoint status.", "status", status, "error", err)
 
 	if err == nil && len(status.FailureCause.StackTrace) > 0 {
@@ -853,6 +854,19 @@ func (reconciler *ClusterReconciler) takeSavepoint(ctx context.Context, jobID st
 	return err
 }
 
+// savepointFormatType returns the format type to pass to the Flink REST API,
+// or empty string for Flink < 1.15 which does not support the formatType parameter.
+func savepointFormatType(cluster *v1beta1.FlinkCluster) string {
+	appVersion, _ := version.NewVersion(cluster.Spec.FlinkVersion)
+	if appVersion == nil || appVersion.LessThan(v115) {
+		return ""
+	}
+	if cluster.Spec.Job == nil || cluster.Spec.Job.SavepointFormatType == nil {
+		return ""
+	}
+	return string(*cluster.Spec.Job.SavepointFormatType)
+}
+
 func (reconciler *ClusterReconciler) updateStatus(
 	ctx context.Context, ss **v1beta1.SavepointStatus, cs **v1beta1.FlinkClusterControlStatus) {
 	log := logr.FromContextOrDiscard(ctx)
@@ -942,7 +956,7 @@ func (reconciler *ClusterReconciler) updateJobDeployStatus(ctx context.Context)
 }
 
 // getNewSavepointStatus returns newly triggered savepoint status.
-func (reconciler *ClusterReconciler) getNewSavepointStatus(triggerID string, triggerReason v1beta1.SavepointReason, message string, triggerSuccess bool) *v1beta1.SavepointStatus {
+func (reconciler *ClusterReconciler) getNewSavepointStatus(triggerID string, triggerReason v1beta1.SavepointReason, message string, triggerSuccess bool, formatType string) *v1beta1.SavepointStatus {
 	var jobID = reconciler.getFlinkJobID()
 	var savepointState string
 	var now string
@@ -961,6 +975,7 @@ func (reconciler *ClusterReconciler) getNewSavepointStatus(triggerID string, tri
 		UpdateTime:    now,
 		Message:       message,
 		State:         savepointState,
+		FormatType:    formatType,
 	}
 	return savepointStatus
 }

controllers/flinkcluster/flinkcluster_util.go

@@ -202,6 +202,7 @@ func newRevisionDataPatch(cluster *v1beta1.FlinkCluster) ([]byte, error) {
 		c.Spec.Job.RestartPolicy = nil
 		c.Spec.Job.CancelRequested = nil
 		c.Spec.Job.SavepointGeneration = 0
+		c.Spec.Job.SavepointFormatType = nil
 	} else {
 		c = cluster
 	}

controllers/flinkcluster/flinkcluster_util_test.go

@@ -407,3 +407,65 @@ func TestGetFlinkJobSubmitLog(t *testing.T) {
 	submit = getFlinkJobSubmitLogFromString("")
 	assert.Equal(t, submit.jobID, "")
 }
+
+func TestSavepointFormatType(t *testing.T) {
+	native := v1beta1.SavepointFormatTypeNative
+	canonical := v1beta1.SavepointFormatTypeCanonical
+
+	tests := []struct {
+		name         string
+		flinkVersion string
+		job          *v1beta1.JobSpec
+		expected     string
+	}{
+		{
+			name:         "empty version returns empty",
+			flinkVersion: "",
+			job:          &v1beta1.JobSpec{},
+			expected:     "",
+		},
+		{
+			name:         "unparseable version returns empty",
+			flinkVersion: "latest",
+			job:          &v1beta1.JobSpec{SavepointFormatType: &native},
+			expected:     "",
+		},
+		{
+			name:         "Flink 1.14.6 always returns empty",
+			flinkVersion: "1.14.6",
+			job:          &v1beta1.JobSpec{SavepointFormatType: &canonical},
+			expected:     "",
+		},
+		{
+			name:         "Flink 1.15.0 returns NATIVE",
+			flinkVersion: "1.15.0",
+			job:          &v1beta1.JobSpec{SavepointFormatType: &native},
+			expected:     "NATIVE",
+		},
+		{
+			name:         "Flink 1.15.0 returns CANONICAL",
+			flinkVersion: "1.15.0",
+			job:          &v1beta1.JobSpec{SavepointFormatType: &canonical},
+			expected:     "CANONICAL",
+		},
+		{
+			name:         "Flink 1.15 nil formatType returns empty",
+			flinkVersion: "1.15.0",
+			job:          &v1beta1.JobSpec{},
+			expected:     "",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			cluster := &v1beta1.FlinkCluster{
+				Spec: v1beta1.FlinkClusterSpec{
+					FlinkVersion: tt.flinkVersion,
+					Job:          tt.job,
+				},
+			}
+			got := savepointFormatType(cluster)
+			assert.Equal(t, got, tt.expected)
+		})
+	}
+}

docs/savepoints_guide.md

@@ -241,3 +241,23 @@ Note that
 
 Usually you want to store savepoints in remote storages, see this [doc](../images/flink/README.md) on how you can store
 savepoints in GCS.
+
+## Savepoint format
+
+Flink supports two savepoint formats: `CANONICAL` and `NATIVE`. You can control which format is used by setting the
+`savepointFormatType` property in the job spec. If not specified, Flink uses the canonical format by default.
+
+For more details on the differences between the two formats and when to use each, see the
+[Flink documentation on savepoint formats](https://nightlies.apache.org/flink/flink-docs-stable/docs/ops/state/savepoints/#savepoint-format).
+
+```yaml
+apiVersion: flinkoperator.k8s.io/v1beta1
+kind: FlinkCluster
+metadata:
+  name: flinkjobcluster-sample
+spec:
+  ...
+  job:
+    savepointFormatType: NATIVE
+    ...
+```

internal/flink/client.go

@@ -177,12 +177,22 @@ func (c *Client) StopJob(
 }
 
 // TriggerSavepoint triggers an async savepoint operation.
-func (c *Client) TriggerSavepoint(apiBaseURL string, jobID string, dir string, cancel bool) (*SavepointTriggerID, error) {
+// formatType is optional: pass an empty string to omit it (required for Flink < 1.15).
+func (c *Client) TriggerSavepoint(apiBaseURL string, jobID string, dir string, cancel bool, formatType string) (*SavepointTriggerID, error) {
 	url := fmt.Sprintf("%s/jobs/%s/savepoints", apiBaseURL, jobID)
-	jsonStr := fmt.Sprintf(`{
-		"target-directory" : "%s",
-		"cancel-job" : %v
-	}`, dir, cancel)
+	var jsonStr string
+	if formatType != "" {
+		jsonStr = fmt.Sprintf(`{
+			"target-directory" : "%s",
+			"cancel-job" : %v,
+			"formatType" : "%s"
+		}`, dir, cancel, formatType)
+	} else {
+		jsonStr = fmt.Sprintf(`{
+			"target-directory" : "%s",
+			"cancel-job" : %v
+		}`, dir, cancel)
+	}
 	resp, err := c.httpClient.Post(url, "application/json", strings.NewReader(jsonStr))
 	if err != nil {
 		return nil, err
@@ -271,10 +281,10 @@ func (c *Client) GetSavepointStatus(
 }
 
 // TakeSavepoint takes savepoint, blocks until it succeeds or fails.
-func (c *Client) TakeSavepoint(apiBaseURL string, jobID string, dir string) (*SavepointStatus, error) {
+func (c *Client) TakeSavepoint(apiBaseURL string, jobID string, dir string, formatType string) (*SavepointStatus, error) {
 	status := &SavepointStatus{JobID: jobID}
 
-	triggerID, err := c.TriggerSavepoint(apiBaseURL, jobID, dir, false)
+	triggerID, err := c.TriggerSavepoint(apiBaseURL, jobID, dir, false, formatType)
 	if err != nil {
 		return nil, err
 	}
@@ -290,8 +300,8 @@ func (c *Client) TakeSavepoint(apiBaseURL string, jobID string, dir string) (*Sa
 	return status, err
 }
 
-func (c *Client) TakeSavepointAsync(apiBaseURL string, jobID string, dir string) (string, error) {
-	triggerID, err := c.TriggerSavepoint(apiBaseURL, jobID, dir, false)
+func (c *Client) TakeSavepointAsync(apiBaseURL string, jobID string, dir string, formatType string) (string, error) {
+	triggerID, err := c.TriggerSavepoint(apiBaseURL, jobID, dir, false, formatType)
 	if err != nil {
 		return "", err
 	}