[FLINK-35265] Remove type field from CheckpointSpec

Author: mateczagany

docs/content/docs/custom-resource/reference.md

@@ -51,11 +51,11 @@ This serves as a full reference for FlinkDeployment and FlinkSessionJob custom r
 ### CheckpointSpec
 **Class**: org.apache.flink.kubernetes.operator.api.spec.CheckpointSpec
 
-**Description**: Spec for checkpoint state snapshots.
+**Description**: Spec for checkpoint state snapshots. This is an empty class, used to instruct the operator to
+ trigger a checkpoint.
 
 | Parameter | Type | Docs |
 | ----------| ---- | ---- |
-| checkpointType | org.apache.flink.kubernetes.operator.api.status.CheckpointType | Type of checkpoint to take. |
 
 ### FlinkDeploymentSpec
 **Class**: org.apache.flink.kubernetes.operator.api.spec.FlinkDeploymentSpec

flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/CheckpointSpec.java

@@ -18,22 +18,19 @@
 package org.apache.flink.kubernetes.operator.api.spec;
 
 import org.apache.flink.annotation.Experimental;
-import org.apache.flink.kubernetes.operator.api.status.CheckpointType;
 
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
-import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
 import lombok.NoArgsConstructor;
 
-/** Spec for checkpoint state snapshots. */
+/**
+ * Spec for checkpoint state snapshots. This is an empty class, used to instruct the operator to
+ * trigger a checkpoint.
+ */
 @Experimental
 @Data
 @NoArgsConstructor
-@AllArgsConstructor
 @Builder
 @JsonIgnoreProperties(ignoreUnknown = true)
-public class CheckpointSpec {
-    /** Type of checkpoint to take. */
-    private CheckpointType checkpointType = CheckpointType.FULL;
-}
+public class CheckpointSpec {}

flink-kubernetes-operator-api/src/test/java/org/apache/flink/kubernetes/operator/api/utils/BaseTestUtils.java

@@ -249,11 +249,10 @@ public static FlinkStateSnapshot buildFlinkStateSnapshotCheckpoint(
             String namespace,
             CheckpointType checkpointType,
             JobReference jobReference) {
-        var checkpointSpec = CheckpointSpec.builder().checkpointType(checkpointType).build();
         var spec =
                 FlinkStateSnapshotSpec.builder()
                         .jobReference(jobReference)
-                        .checkpoint(checkpointSpec)
+                        .checkpoint(new CheckpointSpec())
                         .build();
         var snapshot = new FlinkStateSnapshot();
         snapshot.setSpec(spec);

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractJobReconciler.java

@@ -412,15 +412,15 @@ private boolean triggerSnapshotIfNeeded(FlinkResourceContext<CR> ctx, SnapshotTy
 
                 break;
             case CHECKPOINT:
-                var checkpointType =
-                        conf.get(KubernetesOperatorConfigOptions.OPERATOR_CHECKPOINT_TYPE);
                 if (createSnapshotResource) {
                     FlinkStateSnapshotUtils.createCheckpointResource(
-                            ctx.getKubernetesClient(), resource, checkpointType, triggerType);
+                            ctx.getKubernetesClient(), resource, triggerType);
 
                     ReconciliationUtils.updateLastReconciledSnapshotTriggerNonce(
                             triggerType, resource, CHECKPOINT);
                 } else {
+                    var checkpointType =
+                            conf.get(KubernetesOperatorConfigOptions.OPERATOR_CHECKPOINT_TYPE);
                     var triggerId =
                             ctx.getFlinkService()
                                     .triggerCheckpoint(

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/snapshot/StateSnapshotReconciler.java

@@ -18,6 +18,7 @@
 package org.apache.flink.kubernetes.operator.reconciler.snapshot;
 
 import org.apache.flink.configuration.CheckpointingOptions;
+import org.apache.flink.core.execution.CheckpointType;
 import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
 import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
 import org.apache.flink.kubernetes.operator.api.spec.FlinkStateSnapshotSpec;
@@ -216,12 +217,7 @@ private Optional<String> triggerCheckpointOrSavepoint(
                 throw new IllegalArgumentException(
                         "Manual checkpoint triggering is not supported for this Flink job (requires Flink 1.17+)");
             }
-            return Optional.of(
-                    flinkService.triggerCheckpoint(
-                            jobId,
-                            org.apache.flink.core.execution.CheckpointType.valueOf(
-                                    spec.getCheckpoint().getCheckpointType().name()),
-                            conf));
+            return Optional.of(flinkService.triggerCheckpoint(jobId, CheckpointType.FULL, conf));
         } else {
             throw new IllegalArgumentException(
                     "Snapshot must specify either savepoint or checkpoint spec");

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkStateSnapshotUtils.java

@@ -27,7 +27,6 @@
 import org.apache.flink.kubernetes.operator.api.spec.FlinkStateSnapshotSpec;
 import org.apache.flink.kubernetes.operator.api.spec.JobReference;
 import org.apache.flink.kubernetes.operator.api.spec.SavepointSpec;
-import org.apache.flink.kubernetes.operator.api.status.CheckpointType;
 import org.apache.flink.kubernetes.operator.api.status.SavepointFormatType;
 import org.apache.flink.kubernetes.operator.api.status.SnapshotTriggerType;
 import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration;
@@ -176,21 +175,17 @@ public static FlinkStateSnapshot createSavepointResource(
      *
      * @param kubernetesClient kubernetes client
      * @param resource Flink resource associated
-     * @param checkpointType type of checkpoint
      * @param triggerType trigger type
      * @return created snapshot
      */
     public static FlinkStateSnapshot createCheckpointResource(
             KubernetesClient kubernetesClient,
             AbstractFlinkResource<?, ?> resource,
-            CheckpointType checkpointType,
             SnapshotTriggerType triggerType) {
-        var checkpointSpec = CheckpointSpec.builder().checkpointType(checkpointType).build();
-
         var snapshotSpec =
                 FlinkStateSnapshotSpec.builder()
                         .jobReference(JobReference.fromFlinkResource(resource))
-                        .checkpoint(checkpointSpec)
+                        .checkpoint(new CheckpointSpec())
                         .build();
 
         var resourceName = getFlinkStateSnapshotName(CHECKPOINT, triggerType, resource);

flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/FlinkStateSnapshotUtilsTest.java

@@ -27,7 +27,6 @@
 import org.apache.flink.kubernetes.operator.api.spec.FlinkVersion;
 import org.apache.flink.kubernetes.operator.api.spec.JobKind;
 import org.apache.flink.kubernetes.operator.api.spec.JobReference;
-import org.apache.flink.kubernetes.operator.api.status.CheckpointType;
 import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus;
 import org.apache.flink.kubernetes.operator.api.status.JobManagerDeploymentStatus;
 import org.apache.flink.kubernetes.operator.api.status.SavepointFormatType;
@@ -173,14 +172,12 @@ public void testGetFlinkStateSnapshotsForResource() {
     @Test
     public void testCreatePeriodicCheckpointResource() {
         var deployment = initDeployment();
-        var checkpointType = CheckpointType.FULL;
 
         var snapshot =
                 FlinkStateSnapshotUtils.createCheckpointResource(
-                        client, deployment, checkpointType, SnapshotTriggerType.PERIODIC);
+                        client, deployment, SnapshotTriggerType.PERIODIC);
 
         assertTrue(snapshot.getSpec().isCheckpoint());
-        assertEquals(checkpointType, snapshot.getSpec().getCheckpoint().getCheckpointType());
         assertEquals(
                 deployment.getMetadata().getName(), snapshot.getSpec().getJobReference().getName());
         assertEquals(JobKind.FLINK_DEPLOYMENT, snapshot.getSpec().getJobReference().getKind());
@@ -238,9 +235,8 @@ public void testCreateCheckpointResource() {
 
         var snapshot =
                 FlinkStateSnapshotUtils.createCheckpointResource(
-                        client, deployment, CheckpointType.FULL, SnapshotTriggerType.MANUAL);
-        assertCheckpointResource(
-                snapshot, deployment, SnapshotTriggerType.MANUAL, CheckpointType.FULL);
+                        client, deployment, SnapshotTriggerType.MANUAL);
+        assertCheckpointResource(snapshot, deployment, SnapshotTriggerType.MANUAL);
     }
 
     @ParameterizedTest
@@ -389,13 +385,11 @@ private void assertSavepointResource(
     private void assertCheckpointResource(
             FlinkStateSnapshot snapshot,
             FlinkDeployment deployment,
-            SnapshotTriggerType triggerType,
-            CheckpointType checkpointType) {
+            SnapshotTriggerType triggerType) {
         assertEquals(
                 triggerType.name(),
                 snapshot.getMetadata().getLabels().get(CrdConstants.LABEL_SNAPSHOT_TYPE));
         assertTrue(snapshot.getSpec().isCheckpoint());
-        assertEquals(checkpointType, snapshot.getSpec().getCheckpoint().getCheckpointType());
 
         assertEquals(
                 deployment.getMetadata().getName(), snapshot.getSpec().getJobReference().getName());

helm/flink-kubernetes-operator/crds/flinkstatesnapshots.flink.apache.org-v1.yml

@@ -38,13 +38,6 @@ spec:
               backoffLimit:
                 type: integer
               checkpoint:
-                properties:
-                  checkpointType:
-                    enum:
-                    - FULL
-                    - INCREMENTAL
-                    - UNKNOWN
-                    type: string
                 type: object
               jobReference:
                 properties: