Move exception handling to getLastCheckpoint in the service

Author: luca-p-castelli

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/SnapshotObserver.java

@@ -37,9 +37,7 @@
 import org.apache.flink.kubernetes.operator.utils.EventRecorder;
 import org.apache.flink.kubernetes.operator.utils.FlinkStateSnapshotUtils;
 import org.apache.flink.kubernetes.operator.utils.SnapshotUtils;
-import org.apache.flink.runtime.rest.util.RestClientException;
 import org.apache.flink.util.CollectionUtil;
-import org.apache.flink.util.ExceptionUtils;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -446,30 +444,17 @@ private void observeLatestCheckpoint(FlinkResourceContext<CR> ctx, String jobId)
         var status = ctx.getResource().getStatus();
         var jobStatus = status.getJobStatus();
 
-        try {
-            ctx.getFlinkService()
-                    .getLastCheckpoint(JobID.fromHexString(jobId), ctx.getObserveConfig())
-                    .ifPresentOrElse(
-                            snapshot -> jobStatus.setUpgradeSavepointPath(snapshot.getLocation()),
-                            () -> {
-                                if (ReconciliationUtils.isJobCancelled(status)) {
-                                    // For cancelled jobs the observed savepoint is always definite,
-                                    // so if empty we know the job doesn't have any
-                                    // checkpoints/savepoints
-                                    jobStatus.setUpgradeSavepointPath(null);
-                                }
-                            });
-        } catch (Exception e) {
-            if (ExceptionUtils.findThrowable(e, RestClientException.class)
-                    .map(ex -> ex.getMessage().contains("Checkpointing has not been enabled"))
-                    .orElse(false)) {
-                LOG.warn(
-                        "Checkpointing not enabled for job {}, skipping checkpoint observation",
-                        jobId,
-                        e);
-                return;
-            }
-            throw e;
-        }
+        ctx.getFlinkService()
+                .getLastCheckpoint(JobID.fromHexString(jobId), ctx.getObserveConfig())
+                .ifPresentOrElse(
+                        snapshot -> jobStatus.setUpgradeSavepointPath(snapshot.getLocation()),
+                        () -> {
+                            if (ReconciliationUtils.isJobCancelled(status)) {
+                                // For cancelled jobs the observed savepoint is always definite,
+                                // so if empty we know the job doesn't have any
+                                // checkpoints/savepoints
+                                jobStatus.setUpgradeSavepointPath(null);
+                            }
+                        });
     }
 }

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java

@@ -546,6 +546,12 @@ public Optional<Savepoint> getLastCheckpoint(JobID jobId, Configuration conf) {
         try {
             latestCheckpointOpt = getCheckpointInfo(jobId, conf).f0;
         } catch (Exception e) {
+            if (e instanceof RestClientException
+                    && e.getMessage() != null
+                    && e.getMessage().contains("Checkpointing has not been enabled")) {
+                LOG.warn("Checkpointing not enabled for job {}", jobId, e);
+                return Optional.empty();
+            }
             throw new ReconciliationException("Could not observe latest savepoint information", e);
         }
 

flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java

@@ -43,7 +43,6 @@
 import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration;
 import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
 import org.apache.flink.kubernetes.operator.controller.FlinkResourceContext;
-import org.apache.flink.kubernetes.operator.exception.ReconciliationException;
 import org.apache.flink.kubernetes.operator.exception.UpgradeFailureException;
 import org.apache.flink.kubernetes.operator.observer.CheckpointFetchResult;
 import org.apache.flink.kubernetes.operator.observer.CheckpointStatsResult;
@@ -576,13 +575,6 @@ public void disposeSavepoint(String savepointPath, Configuration conf) throws Ex
 
     @Override
     public Optional<Savepoint> getLastCheckpoint(JobID jobId, Configuration conf) {
-        if (throwCheckpointingDisabledError) {
-            throw new ReconciliationException(
-                    "Could not observe latest savepoint information",
-                    new RestClientException(
-                            "Checkpointing has not been enabled", HttpResponseStatus.BAD_REQUEST));
-        }
-
         jobs.stream()
                 .filter(js -> js.f1.getJobId().equals(jobId))
                 .findAny()
@@ -602,6 +594,10 @@ public Optional<Savepoint> getLastCheckpoint(JobID jobId, Configuration conf) {
                     Optional<CheckpointHistoryWrapper.CompletedCheckpointInfo>,
                     Optional<CheckpointHistoryWrapper.PendingCheckpointInfo>>
             getCheckpointInfo(JobID jobId, Configuration conf) throws Exception {
+        if (throwCheckpointingDisabledError) {
+            throw new RestClientException(
+                    "Checkpointing has not been enabled", HttpResponseStatus.BAD_REQUEST);
+        }
 
         if (checkpointInfo != null) {
             return checkpointInfo;

flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserverTest.java

@@ -900,7 +900,7 @@ public void jobStatusNotOverwrittenWhenTerminal() throws Exception {
     }
 
     @Test
-    public void observeLatestCheckpointShouldSkipWhenCheckpointingDisabled() throws Exception {
+    public void getLastCheckpointShouldHandleCheckpointingNotEnabled() throws Exception {
         Configuration conf =
                 configManager.getDeployConfig(deployment.getMetadata(), deployment.getSpec());
         flinkService.submitApplicationCluster(deployment.getSpec().getJob(), conf, false);