gyfora commented on code in PR #438:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/438#discussion_r1022790812
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java:
##########
@@ -391,22 +395,30 @@ public void cancelSessionJob(
conf.get(ExecutionCheckpointingOptions.CHECKPOINTING_TIMEOUT)
.getSeconds();
try {
- String savepoint =
- clusterClient
- .stopWithSavepoint(
- jobId,
- false,
- savepointDirectory,
- conf.get(FLINK_VERSION)
-
.isNewerVersionThan(
-
FlinkVersion.v1_14)
- ? conf.get(
-
KubernetesOperatorConfigOptions
-
.OPERATOR_SAVEPOINT_FORMAT_TYPE)
- : null)
- .get(timeout, TimeUnit.SECONDS);
- savepointOpt = Optional.of(savepoint);
- LOG.info("Job successfully suspended with savepoint
{}.", savepoint);
+ if
(ReconciliationUtils.isJobRunning(sessionJobStatus)) {
+ String savepoint =
+ clusterClient
+ .stopWithSavepoint(
+ jobId,
+ false,
+ savepointDirectory,
+ conf.get(FLINK_VERSION)
+
.isNewerVersionThan(
+
FlinkVersion.v1_14)
+ ? conf.get(
+
KubernetesOperatorConfigOptions
+
.OPERATOR_SAVEPOINT_FORMAT_TYPE)
+ : null)
+ .get(timeout, TimeUnit.SECONDS);
+ savepointOpt = Optional.of(savepoint);
+ LOG.info("Job successfully suspended with
savepoint {}.", savepoint);
+ } else if
(ReconciliationUtils.isJobInTerminalState(sessionJobStatus)) {
+ LOG.info(
+ "Job is already in terminal state skipping
cancel-with-savepoint operation.");
+ } else {
+ throw new RuntimeException(
+ "Unexpected non-terminal status: " +
sessionJobStatus);
+ }
Review Comment:
I think it could be moved simply before the switch statement to apply
uniformly
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
