[FLINK-35265] Fix cleanup logic for abandoned snapshots

apache · Jul 24, 2024 · 58ed871 · 58ed871
1 parent 0f03c0a
commit 58ed871
Show file tree

Hide file tree

Showing 3 changed files with 17 additions and 9 deletions.
diff --git a/...ava/org/apache/flink/kubernetes/operator/reconciler/snapshot/StateSnapshotReconciler.java b/...ava/org/apache/flink/kubernetes/operator/reconciler/snapshot/StateSnapshotReconciler.java
@@ -118,19 +118,17 @@ public DeleteControl cleanup(FlinkStateSnapshotContext ctx) throws Exception {
                         resourceName);
                 return DeleteControl.noFinalizerRemoval()
                         .rescheduleAfter(ctx.getOperatorConfig().getReconcileInterval().toMillis());
+            case COMPLETED:
+                var flinkDeployment = getFlinkDeployment(ctx);
+                return handleSnapshotCleanup(resource, flinkDeployment, ctx);
             case FAILED:
-                LOG.info(
-                        "Savepoint was not successful, cleaning up resource {} without disposal...",
-                        resourceName);
-                return DeleteControl.defaultDelete();
             case TRIGGER_PENDING:
+            case ABANDONED:
                 LOG.info(
-                        "Savepoint has not started yet, cleaning up resource {} without disposal...",
+                        "Savepoint state is {}, cleaning up resource {} without disposal...",
+                        state.name(),
                         resourceName);
                 return DeleteControl.defaultDelete();
-            case COMPLETED:
-                var flinkDeployment = getFlinkDeployment(ctx);
-                return handleSnapshotCleanup(resource, flinkDeployment, ctx);
             default:
                 LOG.info("Unknown savepoint state for {}: {}", resourceName, state);
                 return DeleteControl.defaultDelete();

diff --git a/...rator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java b/...rator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java
@@ -41,6 +41,7 @@
 import org.apache.flink.kubernetes.operator.api.spec.Resource;
 import org.apache.flink.kubernetes.operator.api.spec.TaskManagerSpec;
 import org.apache.flink.kubernetes.operator.api.spec.UpgradeMode;
+import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus;
 import org.apache.flink.kubernetes.operator.api.status.JobManagerDeploymentStatus;
 import org.apache.flink.kubernetes.operator.config.FlinkConfigBuilder;
 import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
@@ -518,7 +519,11 @@ public Optional<String> validateStateSnapshot(
             return Optional.of("Job reference must be supplied for this snapshot");
         }
 
-        if (target.isEmpty()) {
+        // If the savepoint has already been processed by the operator, we don't need to check the job reference.
+        if (target.isEmpty()
+                && (savepoint.getStatus() == null
+                        || FlinkStateSnapshotStatus.State.TRIGGER_PENDING.equals(
+                                savepoint.getStatus().getState()))) {
             return Optional.of(
                     String.format(
                             "Target for snapshot (%s) in namespace %s was not found",

diff --git a/...ava/org/apache/flink/kubernetes/operator/controller/FlinkStateSnapshotControllerTest.java b/...ava/org/apache/flink/kubernetes/operator/controller/FlinkStateSnapshotControllerTest.java
@@ -238,6 +238,11 @@ public void testReconcileSavepointCleanup() {
         assertDeleteControl(controller.cleanup(snapshot, context), true, null);
         assertThat(flinkService.getDisposedSavepoints()).isEmpty();
 
+        snapshot.getSpec().getSavepoint().setDisposeOnDelete(true);
+        snapshot.getStatus().setState(ABANDONED);
+        assertDeleteControl(controller.cleanup(snapshot, context), true, null);
+        assertThat(flinkService.getDisposedSavepoints()).isEmpty();
+
         snapshot.getSpec().getSavepoint().setDisposeOnDelete(true);
         snapshot.getStatus().setState(IN_PROGRESS);
         assertDeleteControl(