[FLINK-35265] Address PR comments

apache · Jul 5, 2024 · 5c78025 · 5c78025
1 parent 87b1c3d
commit 5c78025
Show file tree

Hide file tree

Showing 4 changed files with 29 additions and 30 deletions.
diff --git a/...c/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkStateSnapshotStatus.java b/...c/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkStateSnapshotStatus.java
@@ -41,7 +41,7 @@
 public class FlinkStateSnapshotStatus implements Diffable<FlinkStateSnapshotStatus> {
 
     /** Current state of the snapshot. */
-    @PrinterColumn(name = "Snapshot Status")
+    @PrinterColumn(name = "Snapshot State")
     private FlinkStateSnapshotState state = FlinkStateSnapshotState.TRIGGER_PENDING;
 
     /** Trigger ID of the snapshot. */

diff --git a/...in/java/org/apache/flink/kubernetes/operator/controller/FlinkStateSnapshotController.java b/...in/java/org/apache/flink/kubernetes/operator/controller/FlinkStateSnapshotController.java
@@ -91,7 +91,7 @@ public UpdateControl<FlinkStateSnapshot> reconcile(
         observer.observe(ctx);
 
         // validate
-        if (!validateSavepoint(ctx)) {
+        if (!validateSnapshot(ctx)) {
             statusRecorder.patchAndCacheStatus(flinkStateSnapshot, ctx.getKubernetesClient());
             UpdateControl<FlinkStateSnapshot> updateControl = UpdateControl.noUpdate();
             return updateControl.rescheduleAfter(
@@ -186,7 +186,7 @@ public Map<String, EventSource> prepareEventSources(
                 EventSourceUtils.getFlinkStateSnapshotInformerEventSources(context));
     }
 
-    private boolean validateSavepoint(FlinkStateSnapshotContext ctx) {
+    private boolean validateSnapshot(FlinkStateSnapshotContext ctx) {
         var savepoint = ctx.getResource();
         for (var validator : validators) {
             var validationError =
@@ -199,7 +199,7 @@ private boolean validateSavepoint(FlinkStateSnapshotContext ctx) {
                         EventRecorder.Component.Operator,
                         validationError.get(),
                         ctx.getKubernetesClient());
-                return true;
+                return false;
             }
         }
         return true;

diff --git a/...ache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java b/...ache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java
@@ -124,19 +124,7 @@ public void reconcile(FlinkResourceContext<CR> ctx) throws Exception {
             updateStatusBeforeFirstDeployment(
                     cr, spec, deployConfig, status, ctx.getKubernetesClient());
 
-            Optional<String> savepointPathOpt =
-                    Optional.ofNullable(spec.getJob()).map(JobSpec::getInitialSavepointPath);
-            if (spec.getJob() != null && spec.getJob().getFlinkStateSnapshotReference() != null) {
-                var snapshotRef = spec.getJob().getFlinkStateSnapshotReference();
-                if (snapshotRef.getName() != null) {
-                    savepointPathOpt =
-                            Optional.of(
-                                    FlinkStateSnapshotUtils.getAndValidateFlinkStateSnapshotPath(
-                                            ctx.getKubernetesClient(), snapshotRef));
-                }
-            }
-
-            deploy(ctx, spec, deployConfig, savepointPathOpt, false);
+            deploy(ctx, spec, deployConfig, getInitialSnapshotPath(ctx, spec), false);
 
             ReconciliationUtils.updateStatusForDeployedSpec(cr, deployConfig, clock);
             return;
@@ -192,6 +180,22 @@ public void reconcile(FlinkResourceContext<CR> ctx) throws Exception {
         }
     }
 
+    private Optional<String> getInitialSnapshotPath(
+            FlinkResourceContext<CR> ctx, AbstractFlinkSpec spec) {
+        if (spec.getJob() == null || spec.getJob().getFlinkStateSnapshotReference() == null) {
+            return Optional.ofNullable(spec.getJob()).map(JobSpec::getInitialSavepointPath);
+        }
+
+        var snapshotRef = spec.getJob().getFlinkStateSnapshotReference();
+        if (snapshotRef.getName() != null) {
+            return Optional.of(
+                    FlinkStateSnapshotUtils.getAndValidateFlinkStateSnapshotPath(
+                            ctx.getKubernetesClient(), snapshotRef));
+        }
+
+        return Optional.empty();
+    }
+
     private void applyAutoscaler(FlinkResourceContext<CR> ctx) throws Exception {
         var autoScalerCtx = ctx.getJobAutoScalerContext();
         boolean autoscalerEnabled =

diff --git a/...ava/org/apache/flink/kubernetes/operator/controller/FlinkStateSnapshotControllerTest.java b/...ava/org/apache/flink/kubernetes/operator/controller/FlinkStateSnapshotControllerTest.java
@@ -125,24 +125,19 @@ public void testReconcileSnapshotDeploymentDoesNotExist() {
         var deployment = createDeployment();
         context = TestUtils.createSnapshotContext(client, null);
         var snapshot = createSavepoint(deployment);
+        controller.reconcile(snapshot, context);
 
-        var exception =
-                assertThrows(
-                        ReconciliationException.class,
-                        () -> controller.reconcile(snapshot, context));
-        assertThat(exception.getCause()).hasMessageContaining("not found");
         assertThat(snapshot.getStatus().getState())
                 .isEqualTo(FlinkStateSnapshotState.TRIGGER_PENDING);
 
         assertThat(flinkStateSnapshotEventCollector.events)
-                .anySatisfy(
-                        event -> {
-                            assertThat(event.getReason())
-                                    .isEqualTo(EventRecorder.Reason.SnapshotError.name());
-                            assertThat(event.getType())
-                                    .isEqualTo(EventRecorder.Type.Warning.name());
-                            assertThat(event.getMessage())
-                                    .isEqualTo(exception.getCause().getMessage());
+                .hasSize(1)
+                .allSatisfy(
+                        e -> {
+                            assertThat(e.getReason())
+                                    .isEqualTo(EventRecorder.Reason.ValidationError.name());
+                            assertThat(e.getType()).isEqualTo(EventRecorder.Type.Warning.name());
+                            assertThat(e.getMessage()).contains("was not found");
                         });
     }