[GitHub] [flink-kubernetes-operator] gyfora commented on a diff in pull request #438: [FLINK-29974] Not allowing the cancelling the which are already in the completed state.

[GitBox <gi...@apache.org>]

gyfora commented on code in PR #438:
URL: https://github.com/apache/flink-kubernetes-operator/pull/438#discussion_r1022678536


##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java:
##########
@@ -362,11 +362,27 @@ public void cancelSessionJob(
             FlinkSessionJob sessionJob, UpgradeMode upgradeMode, Configuration conf)
             throws Exception {
 
+        // Not allowing the jobs which are already in the completed state.
+        String[] jobStateCannotBeCancelled =
+                new String[] {
+                    JobStatus.CANCELLING.name(),
+                    JobStatus.CANCELED.name(),
+                    JobStatus.FAILING.name(),
+                    JobStatus.FAILED.name(),
+                    JobStatus.FINISHED.name()
+                };
+
         var jobStatus = sessionJob.getStatus().getJobStatus();
+
+        if (Arrays.stream(jobStateCannotBeCancelled).anyMatch(jobStatus.getState()::contains)) {
+            throw new RuntimeException("Job is Already in " + jobStatus.getState() + " state");
+        }
+

Review Comment:
   This logic is not correct. We should not throw an error, here that would only make the current situation worse.
   
   please look at the logic for application clusters:
   ```
   if (ReconciliationUtils.isJobRunning(deploymentStatus)) {
                               LOG.info("Suspending job with savepoint.");
                               String savepoint =
                                       clusterClient
                                               .stopWithSavepoint(
                                                       Preconditions.checkNotNull(jobId),
                                                       false,
                                                       savepointDirectory,
                                                       conf.get(FLINK_VERSION)
                                                                       .isNewerVersionThan(
                                                                               FlinkVersion.v1_14)
                                                               ? savepointFormatType
                                                               : null)
                                               .get(timeout, TimeUnit.SECONDS);
                               savepointOpt = Optional.of(savepoint);
                               LOG.info("Job successfully suspended with savepoint {}.", savepoint);
                           } else if (ReconciliationUtils.isJobInTerminalState(deploymentStatus)) {
                               LOG.info(
                                       "Job is already in terminal state skipping cancel-with-savepoint operation.");
                           } else {
                               throw new RuntimeException(
                                       "Unexpected non-terminal status: " + deploymentStatus);
                           }
   ```
   if the job is already in terminal state we **do not** throw error, because the job is already in the state where we want to get to.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org