gyfora commented on code in PR #438:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/438#discussion_r1022790254
##########
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:
any reason why the logging and exception is not present for the STATELESS
branch?
--
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]
