[FLINK-38033] Fix accidental upgrade snapshot dispose bug

Author: gyfora

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

@@ -409,25 +409,30 @@ protected void setUpgradeSavepointPath(
                         conf.get(KubernetesOperatorConfigOptions.OPERATOR_SAVEPOINT_FORMAT_TYPE)
                                 .name());
 
-        FlinkStateSnapshotUtils.createUpgradeSnapshotResource(
-                conf,
-                ctx.getOperatorConfig(),
-                ctx.getKubernetesClient(),
-                ctx.getResource(),
-                savepointFormatType,
-                savepointLocation);
+        var snapshotCrOpt =
+                FlinkStateSnapshotUtils.createUpgradeSnapshotResource(
+                        conf,
+                        ctx.getOperatorConfig(),
+                        ctx.getKubernetesClient(),
+                        ctx.getResource(),
+                        savepointFormatType,
+                        savepointLocation);
         var jobStatus = ctx.getResource().getStatus().getJobStatus();
         jobStatus.setUpgradeSavepointPath(savepointLocation);
 
-        // Register created savepoint in the now deprecated savepoint info and history
-        var savepoint =
-                new Savepoint(
-                        cancelTs.toEpochMilli(),
-                        savepointLocation,
-                        SnapshotTriggerType.UPGRADE,
-                        savepointFormatType,
-                        null);
-        jobStatus.getSavepointInfo().updateLastSavepoint(savepoint);
+        if (snapshotCrOpt.isEmpty()) {
+            // Register created savepoint in the now deprecated savepoint info and history
+            // only if snapshot CR was not created, otherwise it would be double recorded
+            // and disposed immediately
+            var savepoint =
+                    new Savepoint(
+                            cancelTs.toEpochMilli(),
+                            savepointLocation,
+                            SnapshotTriggerType.UPGRADE,
+                            savepointFormatType,
+                            null);
+            jobStatus.getSavepointInfo().updateLastSavepoint(savepoint);
+        }
     }
 
     /**

flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java

@@ -90,6 +90,7 @@
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.function.ThrowingConsumer;
 import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.EnumSource;
 import org.junit.jupiter.params.provider.MethodSource;
 import org.junit.jupiter.params.provider.ValueSource;
@@ -112,6 +113,7 @@
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Predicate;
+import java.util.stream.Stream;
 
 import static org.apache.flink.api.common.JobStatus.FINISHED;
 import static org.apache.flink.api.common.JobStatus.RECONCILING;
@@ -137,6 +139,7 @@
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
+import static org.junit.jupiter.params.provider.Arguments.arguments;
 
 /**
  * @link JobStatusObserver unit tests
@@ -235,9 +238,12 @@ public void testSubmitAndCleanUpWithSavepointOnResource(FlinkVersion flinkVersio
     }
 
     @ParameterizedTest
-    @MethodSource("org.apache.flink.kubernetes.operator.TestUtils#flinkVersions")
-    public void testUpgrade(FlinkVersion flinkVersion) throws Exception {
+    @MethodSource("upgradeArgs")
+    public void testUpgrade(FlinkVersion flinkVersion, boolean snapshotResource) throws Exception {
         FlinkDeployment deployment = TestUtils.buildApplicationCluster(flinkVersion);
+        conf.set(SNAPSHOT_RESOURCE_ENABLED, snapshotResource);
+        configManager.updateDefaultConfig(conf);
+        operatorConfig = configManager.getOperatorConfiguration();
 
         reconciler.reconcile(deployment, context);
         var runningJobs = flinkService.listJobs();
@@ -305,26 +311,35 @@ public void testUpgrade(FlinkVersion flinkVersion) throws Exception {
         assertEquals(0, flinkService.getRunningCount());
 
         var spInfo = statefulUpgrade.getStatus().getJobStatus().getSavepointInfo();
-        assertEquals("savepoint_0", spInfo.getLastSavepoint().getLocation());
-        assertEquals(SnapshotTriggerType.UPGRADE, spInfo.getLastSavepoint().getTriggerType());
-        assertEquals(
-                spInfo.getLastSavepoint(),
-                new LinkedList<>(spInfo.getSavepointHistory()).getLast());
+        if (snapshotResource) {
+            assertNull(spInfo.getLastSavepoint());
+        } else {
+            assertEquals("savepoint_0", spInfo.getLastSavepoint().getLocation());
+            assertEquals(SnapshotTriggerType.UPGRADE, spInfo.getLastSavepoint().getTriggerType());
+            assertEquals(
+                    spInfo.getLastSavepoint(),
+                    new LinkedList<>(spInfo.getSavepointHistory()).getLast());
+        }
 
         reconciler.reconcile(statefulUpgrade, context);
 
         runningJobs = flinkService.listJobs();
         assertEquals(1, flinkService.getRunningCount());
         var snapshots = TestUtils.getFlinkStateSnapshotsForResource(kubernetesClient, deployment);
-        assertThat(snapshots).isNotEmpty();
-        assertThat(snapshots.get(0).getSpec().getSavepoint().getPath()).isEqualTo("savepoint_0");
-        assertEquals(
-                SnapshotTriggerType.UPGRADE.name(),
-                snapshots
-                        .get(0)
-                        .getMetadata()
-                        .getLabels()
-                        .get(CrdConstants.LABEL_SNAPSHOT_TRIGGER_TYPE));
+        if (snapshotResource) {
+            assertThat(snapshots).isNotEmpty();
+            assertThat(snapshots.get(0).getSpec().getSavepoint().getPath())
+                    .isEqualTo("savepoint_0");
+            assertEquals(
+                    SnapshotTriggerType.UPGRADE.name(),
+                    snapshots
+                            .get(0)
+                            .getMetadata()
+                            .getLabels()
+                            .get(CrdConstants.LABEL_SNAPSHOT_TRIGGER_TYPE));
+        } else {
+            assertThat(snapshots).isEmpty();
+        }
 
         // Make sure jobId rotated on savepoint
         verifyNewJobId(runningJobs.get(0).f1, runningJobs.get(0).f2, jobId);
@@ -370,6 +385,13 @@ public void testUpgrade(FlinkVersion flinkVersion) throws Exception {
         verifyNewJobId(runningJobs.get(0).f1, runningJobs.get(0).f2, jobId);
     }
 
+    private static Stream<Arguments> upgradeArgs() {
+        return Stream.of(
+                arguments(FlinkVersion.v1_16, true),
+                arguments(FlinkVersion.v1_20, true),
+                arguments(FlinkVersion.v1_20, false));
+    }
+
     private void verifyJobId(
             FlinkDeployment deployment, JobStatusMessage status, Configuration conf, JobID jobId) {
         // jobId set by operator