[FLINK-37766] FlinkSessionJob deletion blocked by finalizer

Author: prshnt

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

@@ -345,9 +345,11 @@ protected CancelResult cancelJob(
                     }
                     break;
                 case CANCEL:
-                    cancelJobOrError(clusterClient, status, false);
-                    // This is async we need to return
-                    return CancelResult.pending();
+                    if (!cancelJobOrError(clusterClient, status, false)) {
+                        // This is async we need to return
+                        return CancelResult.pending();
+                    }
+                    break;
             }
         }
         if (suspendMode.deleteCluster() || deleteCluster) {
@@ -370,9 +372,12 @@ public CancelResult cancelSessionJob(
             switch (suspendMode) {
                 case STATELESS:
                 case CANCEL:
-                    cancelJobOrError(clusterClient, status, suspendMode == SuspendMode.STATELESS);
-                    // This is async we need to return and re-observe
-                    return CancelResult.pending();
+                    if (!cancelJobOrError(
+                            clusterClient, status, suspendMode == SuspendMode.STATELESS)) {
+                        // This is async we need to return and re-observe
+                        return CancelResult.pending();
+                    }
+                    break;
                 case SAVEPOINT:
                     savepointPath = savepointJobOrError(clusterClient, status, conf);
                     break;
@@ -383,14 +388,30 @@ public CancelResult cancelSessionJob(
         return CancelResult.completed(savepointPath);
     }
 
-    public void cancelJobOrError(
+    /**
+     * Attempts to cancel a Flink job given its current status. If the job is already in the process
+     * of being cancelled or has been terminated, the method handles these cases accordingly.
+     *
+     * @param clusterClient the {@code RestClusterClient} instance used to interact with the Flink
+     *     cluster.
+     * @param status the current status of the job, encapsulated in a {@code CommonStatus} object.
+     * @param ignoreMissing a flag indicating whether the absence of the job should be ignored. If
+     *     {@code true}, the method will return {@code true} when the job is missing. If {@code
+     *     false}, an exception will be thrown when the job cannot be found.
+     * @return {@code true} if the job was already missing or terminated, and no further action is
+     *     needed. {@code false} if cancellation was successfully initiated and is still pending
+     *     (the caller should await completion before proceeding).
+     * @throws UpgradeFailureException if the job cannot be cancelled due to an unexpected error, or
+     *     if the job is missing and {@code ignoreMissing} is set to {@code false}.
+     */
+    public boolean cancelJobOrError(
             RestClusterClient<String> clusterClient,
             CommonStatus<?> status,
             boolean ignoreMissing) {
         var jobID = JobID.fromHexString(status.getJobStatus().getJobId());
         if (ReconciliationUtils.isJobCancelling(status)) {
             LOG.info("Job already cancelling");
-            return;
+            return false;
         }
         LOG.info("Cancelling job");
         try {
@@ -402,6 +423,7 @@ public void cancelJobOrError(
             if (isJobMissing(e)) {
                 if (ignoreMissing) {
                     LOG.info("Job already missing");
+                    return true;
                 } else {
                     throw new UpgradeFailureException(
                             "Cannot find job when trying to cancel",
@@ -410,13 +432,15 @@ public void cancelJobOrError(
                 }
             } else if (isJobTerminated(e)) {
                 LOG.info("Job already terminated");
+                return true;
             } else {
                 LOG.warn("Error while cancelling job", e);
                 throw new UpgradeFailureException(
                         "Cancellation Error", EventRecorder.Reason.CleanupFailed.name(), e);
             }
         }
         status.getJobStatus().setState(JobStatus.CANCELLING);
+        return false;
     }
 
     public String savepointJobOrError(
@@ -487,9 +511,16 @@ public static boolean isJobTerminated(Exception e) {
             return true;
         }
 
-        return findThrowable(e, RestClientException.class)
+        if (findThrowable(e, RestClientException.class)
                 .map(RestClientException::getHttpResponseStatus)
                 .map(respCode -> HttpResponseStatus.CONFLICT == respCode)
+                .orElse(false)) {
+            return true;
+        }
+
+        return Optional.ofNullable(ExceptionUtils.getExceptionMessage(e))
+                .map(String::toLowerCase)
+                .map(msg -> msg.contains("already reached another terminal state"))
                 .orElse(false);
     }
 
@@ -1051,6 +1082,7 @@ private static String createEmptyJar() {
         }
     }
 
+    @Override
     public Map<String, String> getMetrics(
             Configuration conf, String jobId, List<String> metricNames) throws Exception {
         try (var clusterClient = getClusterClient(conf)) {

flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkServiceTest.java

@@ -113,6 +113,7 @@
 import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
 import io.fabric8.kubernetes.client.server.mock.KubernetesMockServer;
 import lombok.SneakyThrows;
+import org.jetbrains.annotations.NotNull;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -334,15 +335,7 @@ public void cancelJobWithStatelessUpgradeModeTest() throws Exception {
     @ValueSource(ints = {404, 409, 500})
     public void cancelErrorHandling(int statusCode) throws Exception {
 
-        var testingClusterClient =
-                new TestingClusterClient<>(configuration, TestUtils.TEST_DEPLOYMENT_NAME);
-        testingClusterClient.setCancelFunction(
-                jobID ->
-                        CompletableFuture.failedFuture(
-                                new RuntimeException(
-                                        new RestClientException(
-                                                "errrr", HttpResponseStatus.valueOf(statusCode)))));
-        var flinkService = new TestingService(testingClusterClient);
+        var flinkService = getTestingService("errrr", HttpResponseStatus.valueOf(statusCode));
 
         JobID jobID = JobID.generate();
         var job = TestUtils.buildSessionJob();
@@ -360,10 +353,104 @@ public void cancelErrorHandling(int statusCode) throws Exception {
             assertEquals(RUNNING, jobStatus.getState());
         } else {
             flinkService.cancelSessionJob(job, SuspendMode.STATELESS, new Configuration());
-            assertEquals(CANCELLING, jobStatus.getState());
+            assertEquals(FINISHED, jobStatus.getState());
+            assertNull(jobStatus.getJobId());
         }
     }
 
+    @Test
+    public void cancelErrorHandlingWithTerminalStateMessage() throws Exception {
+        var flinkService =
+                getTestingService(
+                        "Job cancellation failed because the job has already reached another terminal state (FAILED).",
+                        HttpResponseStatus.BAD_REQUEST);
+
+        JobID jobID = JobID.generate();
+        var job = TestUtils.buildSessionJob();
+        var jobStatus = job.getStatus().getJobStatus();
+        jobStatus.setJobId(jobID.toHexString());
+        jobStatus.setState(RUNNING);
+        ReconciliationUtils.updateStatusForDeployedSpec(job, new Configuration());
+
+        flinkService.cancelSessionJob(job, SuspendMode.STATELESS, new Configuration());
+        assertEquals(FINISHED, jobStatus.getState());
+        assertNull(jobStatus.getJobId());
+    }
+
+    /**
+     * Reproduces the operator-upgrade scenario for Session Mode with CANCEL upgrade mode: when a
+     * running session job's JobManager has already moved the job into a terminal state (e.g.
+     * FAILED) and the operator (after a restart/upgrade) tries to cancel it, the cancellation
+     * request comes back with "already reached another terminal state". Previously this caused the
+     * finalizer to never be removed, leaving the CR stuck in Terminating.
+     */
+    @Test
+    public void cancelSessionJobWithCancelModeAndTerminalStateMessage() throws Exception {
+        var flinkService =
+                getTestingService(
+                        "Job cancellation failed because the job has already reached another terminal state (FAILED).",
+                        HttpResponseStatus.BAD_REQUEST);
+
+        JobID jobID = JobID.generate();
+        var job = TestUtils.buildSessionJob();
+        var jobStatus = job.getStatus().getJobStatus();
+        jobStatus.setJobId(jobID.toHexString());
+        jobStatus.setState(RUNNING);
+        ReconciliationUtils.updateStatusForDeployedSpec(job, new Configuration());
+
+        var result = flinkService.cancelSessionJob(job, SuspendMode.CANCEL, new Configuration());
+        // Must NOT be pending — the CR would otherwise be stuck in Terminating indefinitely
+        assertFalse(result.isPending());
+        assertEquals(FINISHED, jobStatus.getState());
+        assertNull(jobStatus.getJobId());
+    }
+
+    @NotNull
+    private TestingService getTestingService(String message, HttpResponseStatus badRequest)
+            throws Exception {
+        final var testingClusterClient =
+                new TestingClusterClient<>(configuration, TestUtils.TEST_DEPLOYMENT_NAME);
+        testingClusterClient.setCancelFunction(
+                jobID ->
+                        CompletableFuture.failedFuture(
+                                new RuntimeException(
+                                        new RestClientException(message, badRequest))));
+        return new TestingService(testingClusterClient);
+    }
+
+    /**
+     * Reproduces FLINK-37766 for Application Mode: when a running application job's JobManager has
+     * moved the job to a terminal state (e.g. FAILED due to HA desync) and the operator tries to
+     * cancel the job with CANCEL suspend mode (used for last-state upgrades), the "already reached
+     * another terminal state" response previously caused the operator to always return
+     * CancelResult.pending(), looping forever without completing the upgrade/deletion.
+     */
+    @Test
+    public void cancelApplicationJobWithCancelModeAndTerminalStateMessage() throws Exception {
+        var flinkService =
+                getTestingService(
+                        "Job cancellation failed because the job has already reached another terminal state (FAILED).",
+                        HttpResponseStatus.BAD_REQUEST);
+
+        JobID jobID = JobID.generate();
+        FlinkDeployment deployment = TestUtils.buildApplicationCluster();
+        deployment.getStatus().setJobManagerDeploymentStatus(JobManagerDeploymentStatus.READY);
+        JobStatus jobStatus = deployment.getStatus().getJobStatus();
+        jobStatus.setJobId(jobID.toHexString());
+        jobStatus.setState(RUNNING);
+        ReconciliationUtils.updateStatusForDeployedSpec(deployment, new Configuration());
+
+        var result =
+                flinkService.cancelJob(
+                        deployment,
+                        SuspendMode.CANCEL,
+                        configManager.getObserveConfig(deployment),
+                        false);
+        // Must NOT be pending — the operator would otherwise loop forever on the upgrade
+        assertFalse(result.isPending());
+        assertEquals(FINISHED, jobStatus.getState());
+    }
+
     @ParameterizedTest
     @ValueSource(booleans = {true, false})
     public void cancelJobWithSavepointUpgradeModeTest(boolean deleteAfterSavepoint)