You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by mb...@apache.org on 2022/12/02 13:08:33 UTC
[flink-kubernetes-operator] branch main updated: [FLINK-29974] Allow session job cancel to be called for each job state
This is an automated email from the ASF dual-hosted git repository.
mbalassi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
The following commit(s) were added to refs/heads/main by this push:
new ea01e294 [FLINK-29974] Allow session job cancel to be called for each job state
ea01e294 is described below
commit ea01e294cf1b68d597244d0a11b3c81822a163e7
Author: Gabor Somogyi <ga...@apple.com>
AuthorDate: Fri Dec 2 14:08:27 2022 +0100
[FLINK-29974] Allow session job cancel to be called for each job state
---
e2e-tests/utils.sh | 6 +-
.../operator/service/AbstractFlinkService.java | 161 +++++++------
.../kubernetes/operator/TestingFlinkService.java | 7 +
.../sessionjob/SessionJobReconcilerTest.java | 257 +++++++++++++++------
4 files changed, 286 insertions(+), 145 deletions(-)
diff --git a/e2e-tests/utils.sh b/e2e-tests/utils.sh
index e38f9631..b092b7a2 100755
--- a/e2e-tests/utils.sh
+++ b/e2e-tests/utils.sh
@@ -159,12 +159,14 @@ function debug_and_show_logs {
kubectl describe all
echo "Operator logs:"
+ operator_pod_namespace=$(get_operator_pod_namespace)
operator_pod_name=$(get_operator_pod_name)
- kubectl logs "${operator_pod_name}"
+ echo "Operator namespace: ${operator_pod_namespace} pod: ${operator_pod_name}"
+ kubectl logs -n "${operator_pod_namespace}" "${operator_pod_name}"
echo "Flink logs:"
kubectl get pods -o jsonpath='{range .items[*]}{.metadata.name}{"\n"}{end}' | while read pod;do
- containers=(`kubectl get pods $pod -o jsonpath='{.spec.containers[*].name}'`)
+ containers=(`kubectl get pods $pod -o jsonpath='{.spec.containers[*].name}'`)
i=0
for container in "${containers[@]}"; do
echo "Current logs for $pod:$container: "
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
index e68168e8..0e33386e 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
@@ -292,8 +292,8 @@ public abstract class AbstractFlinkService implements FlinkService {
final long timeout =
conf.get(ExecutionCheckpointingOptions.CHECKPOINTING_TIMEOUT)
.getSeconds();
- try {
- if (ReconciliationUtils.isJobRunning(deploymentStatus)) {
+ if (ReconciliationUtils.isJobRunning(deploymentStatus)) {
+ try {
LOG.info("Suspending job with savepoint.");
String savepoint =
clusterClient
@@ -309,22 +309,23 @@ public abstract class AbstractFlinkService implements FlinkService {
.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);
+ } catch (TimeoutException exception) {
+ throw new FlinkException(
+ String.format(
+ "Timed out stopping the job %s in Flink cluster %s with savepoint, "
+ + "please configure a larger timeout via '%s'",
+ jobId,
+ clusterId,
+ ExecutionCheckpointingOptions.CHECKPOINTING_TIMEOUT
+ .key()),
+ exception);
}
- } catch (TimeoutException exception) {
- throw new FlinkException(
- String.format(
- "Timed out stopping the job %s in Flink cluster %s with savepoint, "
- + "please configure a larger timeout via '%s'",
- jobId,
- clusterId,
- ExecutionCheckpointingOptions.CHECKPOINTING_TIMEOUT.key()),
- exception);
+ } 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 (deleteClusterAfterSavepoint) {
LOG.info("Cleaning up deployment after stop-with-savepoint");
@@ -364,67 +365,85 @@ public abstract class AbstractFlinkService implements FlinkService {
FlinkSessionJob sessionJob, UpgradeMode upgradeMode, Configuration conf)
throws Exception {
- var jobStatus = sessionJob.getStatus().getJobStatus();
+ var sessionJobStatus = sessionJob.getStatus();
+ var jobStatus = sessionJobStatus.getJobStatus();
var jobIdString = jobStatus.getJobId();
Preconditions.checkNotNull(jobIdString, "The job to be suspend should not be null");
var jobId = JobID.fromHexString(jobIdString);
Optional<String> savepointOpt = Optional.empty();
- try (ClusterClient<String> clusterClient = getClusterClient(conf)) {
- final String clusterId = clusterClient.getClusterId();
- switch (upgradeMode) {
- case STATELESS:
- LOG.info("Cancelling job.");
- clusterClient
- .cancel(jobId)
- .get(
- configManager
- .getOperatorConfiguration()
- .getFlinkCancelJobTimeout()
- .toSeconds(),
- TimeUnit.SECONDS);
- LOG.info("Job successfully cancelled.");
- break;
- case SAVEPOINT:
- LOG.info("Suspending job with savepoint.");
- final String savepointDirectory =
- Preconditions.checkNotNull(
- conf.get(CheckpointingOptions.SAVEPOINT_DIRECTORY));
- final long timeout =
- conf.get(ExecutionCheckpointingOptions.CHECKPOINTING_TIMEOUT)
- .getSeconds();
- try {
- String savepoint =
- clusterClient
- .stopWithSavepoint(
+
+ LOG.debug("Current job state: {}", jobStatus.getState());
+
+ if (!ReconciliationUtils.isJobInTerminalState(sessionJobStatus)) {
+ LOG.debug("Job is not in terminal state, cancelling it");
+
+ try (ClusterClient<String> clusterClient = getClusterClient(conf)) {
+ final String clusterId = clusterClient.getClusterId();
+ switch (upgradeMode) {
+ case STATELESS:
+ LOG.info("Cancelling job.");
+ clusterClient
+ .cancel(jobId)
+ .get(
+ configManager
+ .getOperatorConfiguration()
+ .getFlinkCancelJobTimeout()
+ .toSeconds(),
+ TimeUnit.SECONDS);
+ LOG.info("Job successfully cancelled.");
+ break;
+ case SAVEPOINT:
+ if (ReconciliationUtils.isJobRunning(sessionJobStatus)) {
+ LOG.info("Suspending job with savepoint.");
+ final String savepointDirectory =
+ Preconditions.checkNotNull(
+ conf.get(CheckpointingOptions.SAVEPOINT_DIRECTORY));
+ final long timeout =
+ 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);
+ } catch (TimeoutException exception) {
+ throw new FlinkException(
+ String.format(
+ "Timed out stopping the job %s in Flink cluster %s with savepoint, "
+ + "please configure a larger timeout via '%s'",
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);
- } catch (TimeoutException exception) {
- throw new FlinkException(
- String.format(
- "Timed out stopping the job %s in Flink cluster %s with savepoint, "
- + "please configure a larger timeout via '%s'",
- jobId,
- clusterId,
- ExecutionCheckpointingOptions.CHECKPOINTING_TIMEOUT.key()),
- exception);
- }
- break;
- case LAST_STATE:
- default:
- throw new RuntimeException("Unsupported upgrade mode " + upgradeMode);
+ clusterId,
+ ExecutionCheckpointingOptions.CHECKPOINTING_TIMEOUT
+ .key()),
+ exception);
+ }
+ } else {
+ throw new RuntimeException(
+ "Unexpected non-terminal status: " + jobStatus.getState());
+ }
+ break;
+ case LAST_STATE:
+ default:
+ throw new RuntimeException("Unsupported upgrade mode " + upgradeMode);
+ }
}
+ } else {
+ LOG.debug("Job is in terminal state, skipping cancel");
}
+
jobStatus.setState(JobStatus.FINISHED.name());
savepointOpt.ifPresent(
location -> {
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java
index 5d098806..04a03c54 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java
@@ -102,6 +102,7 @@ public class TestingFlinkService extends AbstractFlinkService {
private final List<String> disposedSavepoints = new ArrayList<>();
private final Map<String, Boolean> savepointTriggers = new HashMap<>();
private int desiredReplicas = 0;
+ private int cancelJobCallCount = 0;
private Map<String, String> metricsValues = new HashMap<>();
@@ -366,6 +367,8 @@ public class TestingFlinkService extends AbstractFlinkService {
private String cancelJob(FlinkVersion flinkVersion, JobID jobID, boolean savepoint)
throws Exception {
+ cancelJobCallCount++;
+
var jobOpt = jobs.stream().filter(js -> js.f1.getJobId().equals(jobID)).findAny();
if (jobOpt.isEmpty()) {
@@ -510,4 +513,8 @@ public class TestingFlinkService extends AbstractFlinkService {
Configuration conf, String jobId, List<String> metricNames) {
return metricsValues;
}
+
+ public int getCancelJobCallCount() {
+ return cancelJobCallCount;
+ }
}
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconcilerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconcilerTest.java
index c3473a43..b9309c9f 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconcilerTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconcilerTest.java
@@ -45,17 +45,33 @@ import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
import javax.annotation.Nullable;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ThreadLocalRandom;
-
+import java.util.stream.Stream;
+
+import static org.apache.flink.api.common.JobStatus.CANCELED;
+import static org.apache.flink.api.common.JobStatus.CANCELLING;
+import static org.apache.flink.api.common.JobStatus.CREATED;
+import static org.apache.flink.api.common.JobStatus.FAILED;
+import static org.apache.flink.api.common.JobStatus.FAILING;
+import static org.apache.flink.api.common.JobStatus.FINISHED;
+import static org.apache.flink.api.common.JobStatus.INITIALIZING;
+import static org.apache.flink.api.common.JobStatus.RECONCILING;
+import static org.apache.flink.api.common.JobStatus.RESTARTING;
+import static org.apache.flink.api.common.JobStatus.RUNNING;
+import static org.apache.flink.api.common.JobStatus.SUSPENDED;
import static org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.OPERATOR_JOB_RESTART_FAILED;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
/** Tests for {@link SessionJobReconciler}. */
@@ -117,16 +133,10 @@ public class SessionJobReconcilerTest {
reconciler.reconcile(sessionJob, TestUtils.createContextWithReadyFlinkDeployment());
assertEquals(1, flinkService.listJobs().size());
verifyAndSetRunningJobsToStatus(
- sessionJob,
- JobState.RUNNING,
- org.apache.flink.api.common.JobStatus.RECONCILING.name(),
- null,
- flinkService.listJobs());
+ sessionJob, JobState.RUNNING, RECONCILING.name(), null, flinkService.listJobs());
// clean up
reconciler.cleanup(sessionJob, TestUtils.createContextWithReadyFlinkDeployment());
- assertEquals(
- org.apache.flink.api.common.JobStatus.FINISHED,
- flinkService.listJobs().get(0).f1.getJobState());
+ assertEquals(FINISHED, flinkService.listJobs().get(0).f1.getJobState());
}
@Test
@@ -137,23 +147,13 @@ public class SessionJobReconcilerTest {
reconciler.reconcile(sessionJob, TestUtils.createContextWithReadyFlinkDeployment());
assertEquals(1, flinkService.listJobs().size());
verifyAndSetRunningJobsToStatus(
- sessionJob,
- JobState.RUNNING,
- org.apache.flink.api.common.JobStatus.RECONCILING.name(),
- null,
- flinkService.listJobs());
+ sessionJob, JobState.RUNNING, RECONCILING.name(), null, flinkService.listJobs());
sessionJob.getSpec().setRestartNonce(2L);
reconciler.reconcile(sessionJob, TestUtils.createContextWithReadyFlinkDeployment());
- assertEquals(
- org.apache.flink.api.common.JobStatus.FINISHED,
- flinkService.listJobs().get(0).f1.getJobState());
+ assertEquals(FINISHED, flinkService.listJobs().get(0).f1.getJobState());
reconciler.reconcile(sessionJob, TestUtils.createContextWithReadyFlinkDeployment());
verifyAndSetRunningJobsToStatus(
- sessionJob,
- JobState.RUNNING,
- org.apache.flink.api.common.JobStatus.RECONCILING.name(),
- null,
- flinkService.listJobs());
+ sessionJob, JobState.RUNNING, RECONCILING.name(), null, flinkService.listJobs());
}
@Test
@@ -165,21 +165,12 @@ public class SessionJobReconcilerTest {
reconciler.reconcile(sessionJob, readyContext);
assertEquals(1, flinkService.listJobs().size());
verifyAndSetRunningJobsToStatus(
- sessionJob,
- JobState.RUNNING,
- org.apache.flink.api.common.JobStatus.RECONCILING.name(),
- null,
- flinkService.listJobs());
+ sessionJob, JobState.RUNNING, RECONCILING.name(), null, flinkService.listJobs());
- sessionJob
- .getStatus()
- .getJobStatus()
- .setState(org.apache.flink.api.common.JobStatus.FAILED.name());
+ sessionJob.getStatus().getJobStatus().setState(FAILED.name());
reconciler.reconcile(sessionJob, readyContext);
assertEquals(2, flinkService.listJobs().size());
- assertEquals(
- org.apache.flink.api.common.JobStatus.RUNNING,
- flinkService.listJobs().get(1).f1.getJobState());
+ assertEquals(RUNNING, flinkService.listJobs().get(1).f1.getJobState());
}
@Test
@@ -192,7 +183,7 @@ public class SessionJobReconcilerTest {
verifyAndSetRunningJobsToStatus(
sessionJob,
JobState.RUNNING,
- org.apache.flink.api.common.JobStatus.RECONCILING.name(),
+ RECONCILING.name(),
initSavepointPath,
flinkService.listJobs());
}
@@ -205,20 +196,14 @@ public class SessionJobReconcilerTest {
reconciler.reconcile(sessionJob, readyContext);
assertEquals(1, flinkService.listJobs().size());
verifyAndSetRunningJobsToStatus(
- sessionJob,
- JobState.RUNNING,
- org.apache.flink.api.common.JobStatus.RECONCILING.name(),
- null,
- flinkService.listJobs());
+ sessionJob, JobState.RUNNING, RECONCILING.name(), null, flinkService.listJobs());
var statelessSessionJob = ReconciliationUtils.clone(sessionJob);
statelessSessionJob.getSpec().getJob().setUpgradeMode(UpgradeMode.STATELESS);
statelessSessionJob.getSpec().getJob().setParallelism(2);
// job suspended first
reconciler.reconcile(statelessSessionJob, readyContext);
- assertEquals(
- org.apache.flink.api.common.JobStatus.FINISHED,
- flinkService.listJobs().get(0).f1.getJobState());
+ assertEquals(FINISHED, flinkService.listJobs().get(0).f1.getJobState());
verifyJobState(statelessSessionJob, JobState.SUSPENDED, "FINISHED");
flinkService.clear();
@@ -227,7 +212,7 @@ public class SessionJobReconcilerTest {
verifyAndSetRunningJobsToStatus(
statelessSessionJob,
JobState.RUNNING,
- org.apache.flink.api.common.JobStatus.RECONCILING.name(),
+ RECONCILING.name(),
null,
flinkService.listJobs());
}
@@ -256,16 +241,14 @@ public class SessionJobReconcilerTest {
verifyAndSetRunningJobsToStatus(
statefulSessionJob,
JobState.RUNNING,
- org.apache.flink.api.common.JobStatus.RECONCILING.name(),
+ RECONCILING.name(),
null,
flinkService.listJobs());
reconciler.reconcile(statefulSessionJob, readyContext);
// job suspended first
- assertEquals(
- org.apache.flink.api.common.JobStatus.FINISHED,
- flinkService.listJobs().get(0).f1.getJobState());
+ assertEquals(FINISHED, flinkService.listJobs().get(0).f1.getJobState());
verifyJobState(statefulSessionJob, JobState.SUSPENDED, "FINISHED");
assertEquals(
1,
@@ -291,7 +274,7 @@ public class SessionJobReconcilerTest {
verifyAndSetRunningJobsToStatus(
statefulSessionJob,
JobState.RUNNING,
- org.apache.flink.api.common.JobStatus.RECONCILING.name(),
+ RECONCILING.name(),
"savepoint_0",
flinkService.listJobs());
}
@@ -304,11 +287,7 @@ public class SessionJobReconcilerTest {
var readyContext = TestUtils.createContextWithReadyFlinkDeployment();
reconciler.reconcile(sessionJob, readyContext);
verifyAndSetRunningJobsToStatus(
- sessionJob,
- JobState.RUNNING,
- org.apache.flink.api.common.JobStatus.RECONCILING.name(),
- null,
- flinkService.listJobs());
+ sessionJob, JobState.RUNNING, RECONCILING.name(), null, flinkService.listJobs());
assertFalse(SavepointUtils.savepointInProgress(sessionJob.getStatus().getJobStatus()));
@@ -320,18 +299,12 @@ public class SessionJobReconcilerTest {
assertFalse(SavepointUtils.savepointInProgress(sp1SessionJob.getStatus().getJobStatus()));
sp1SessionJob.getSpec().getJob().setSavepointTriggerNonce(2L);
- sp1SessionJob
- .getStatus()
- .getJobStatus()
- .setState(org.apache.flink.api.common.JobStatus.CREATED.name());
+ sp1SessionJob.getStatus().getJobStatus().setState(CREATED.name());
reconciler.reconcile(sp1SessionJob, readyContext);
// do not trigger savepoint if job is not running
assertFalse(SavepointUtils.savepointInProgress(sp1SessionJob.getStatus().getJobStatus()));
- sp1SessionJob
- .getStatus()
- .getJobStatus()
- .setState(org.apache.flink.api.common.JobStatus.RUNNING.name());
+ sp1SessionJob.getStatus().getJobStatus().setState(RUNNING.name());
reconciler.reconcile(sp1SessionJob, readyContext);
assertTrue(SavepointUtils.savepointInProgress(sp1SessionJob.getStatus().getJobStatus()));
@@ -398,11 +371,7 @@ public class SessionJobReconcilerTest {
.getJob()
.getParallelism());
verifyAndSetRunningJobsToStatus(
- sp1SessionJob,
- JobState.RUNNING,
- org.apache.flink.api.common.JobStatus.RECONCILING.name(),
- null,
- flinkService.listJobs());
+ sp1SessionJob, JobState.RUNNING, RECONCILING.name(), null, flinkService.listJobs());
sp1SessionJob.getStatus().getJobStatus().getSavepointInfo().resetTrigger();
ReconciliationUtils.updateLastReconciledSavepointTriggerNonce(
@@ -425,6 +394,154 @@ public class SessionJobReconcilerTest {
assertFalse(SavepointUtils.savepointInProgress(sp1SessionJob.getStatus().getJobStatus()));
}
+ private static Stream<Arguments> cancelStatelessSessionJobParams() {
+ return Stream.of(
+ Arguments.of(INITIALIZING, true),
+ Arguments.of(CREATED, true),
+ Arguments.of(RUNNING, true),
+ Arguments.of(FAILING, true),
+ Arguments.of(FAILED, false),
+ Arguments.of(CANCELLING, true),
+ Arguments.of(CANCELED, false),
+ Arguments.of(FINISHED, false),
+ Arguments.of(RESTARTING, true),
+ Arguments.of(SUSPENDED, true),
+ Arguments.of(RECONCILING, true));
+ }
+
+ @Test
+ public void testCancelStatelessSessionJobParams() {
+ assertEquals(
+ org.apache.flink.api.common.JobStatus.values().length,
+ cancelStatelessSessionJobParams().count());
+ }
+
+ @ParameterizedTest
+ @MethodSource("cancelStatelessSessionJobParams")
+ public void testCancelStatelessSessionJob(
+ org.apache.flink.api.common.JobStatus fromJobStatus, boolean shouldCallCancel)
+ throws Exception {
+ FlinkSessionJob sessionJob = TestUtils.buildSessionJob();
+
+ var readyContext = TestUtils.createContextWithReadyFlinkDeployment();
+
+ reconciler.reconcile(sessionJob, readyContext);
+ assertEquals(1, flinkService.listJobs().size());
+ verifyAndSetRunningJobsToStatus(
+ sessionJob, JobState.RUNNING, RECONCILING.name(), null, flinkService.listJobs());
+
+ var job = flinkService.listJobs().get(0);
+ var jobStatusMessage = job.f1;
+ var jobConfig = job.f2;
+
+ sessionJob.getSpec().getJob().setUpgradeMode(UpgradeMode.STATELESS);
+
+ // Set JobStatusMessage which may be overwritten by cancel
+ job.f1 =
+ new JobStatusMessage(
+ jobStatusMessage.getJobId(),
+ jobStatusMessage.getJobName(),
+ fromJobStatus,
+ jobStatusMessage.getStartTime());
+ // Set state which must be overwritten by cancelSessionJob
+ sessionJob.getStatus().getJobStatus().setState(fromJobStatus.name());
+
+ flinkService.cancelSessionJob(sessionJob, UpgradeMode.STATELESS, jobConfig);
+
+ if (!shouldCallCancel) {
+ assertEquals(0, flinkService.getCancelJobCallCount());
+ assertEquals(fromJobStatus, job.f1.getJobState());
+ } else {
+ assertEquals(1, flinkService.getCancelJobCallCount());
+ assertEquals(FINISHED, job.f1.getJobState());
+ }
+ assertEquals(FINISHED.name(), sessionJob.getStatus().getJobStatus().getState());
+ }
+
+ private static Stream<Arguments> cancelSavepointSessionJobParams() {
+ return Stream.of(
+ Arguments.of(INITIALIZING, true, false),
+ Arguments.of(CREATED, true, false),
+ Arguments.of(RUNNING, false, true),
+ Arguments.of(FAILING, true, false),
+ Arguments.of(FAILED, false, false),
+ Arguments.of(CANCELLING, true, false),
+ Arguments.of(CANCELED, false, false),
+ Arguments.of(FINISHED, false, false),
+ Arguments.of(RESTARTING, true, false),
+ Arguments.of(SUSPENDED, true, false),
+ Arguments.of(RECONCILING, true, false));
+ }
+
+ @Test
+ public void testCancelSavepointSessionJobParams() {
+ assertEquals(
+ org.apache.flink.api.common.JobStatus.values().length,
+ cancelSavepointSessionJobParams().count());
+ }
+
+ @ParameterizedTest
+ @MethodSource("cancelSavepointSessionJobParams")
+ public void testCancelSavepointSessionJob(
+ org.apache.flink.api.common.JobStatus fromJobStatus,
+ boolean shouldThrowException,
+ boolean shouldCallCancel)
+ throws Exception {
+ assertTrue(
+ !shouldThrowException || !shouldCallCancel,
+ "Expecting an exception and cancel to be called is and oxymoron");
+
+ FlinkSessionJob sessionJob = TestUtils.buildSessionJob();
+
+ var readyContext = TestUtils.createContextWithReadyFlinkDeployment();
+
+ reconciler.reconcile(sessionJob, readyContext);
+ assertEquals(1, flinkService.listJobs().size());
+ verifyAndSetRunningJobsToStatus(
+ sessionJob, JobState.RUNNING, RECONCILING.name(), null, flinkService.listJobs());
+
+ var job = flinkService.listJobs().get(0);
+ var jobStatusMessage = job.f1;
+ var jobConfig = job.f2;
+
+ sessionJob.getSpec().getJob().setUpgradeMode(UpgradeMode.SAVEPOINT);
+
+ // Set JobStatusMessage which may be overwritten by cancel
+ job.f1 =
+ new JobStatusMessage(
+ jobStatusMessage.getJobId(),
+ jobStatusMessage.getJobName(),
+ fromJobStatus,
+ jobStatusMessage.getStartTime());
+ // Set state which must be overwritten by cancelSessionJob
+ sessionJob.getStatus().getJobStatus().setState(fromJobStatus.name());
+
+ if (!shouldThrowException) {
+ flinkService.cancelSessionJob(sessionJob, UpgradeMode.SAVEPOINT, jobConfig);
+ } else {
+ var e =
+ assertThrows(
+ RuntimeException.class,
+ () ->
+ flinkService.cancelSessionJob(
+ sessionJob, UpgradeMode.SAVEPOINT, jobConfig));
+ Assertions.assertTrue(e.getMessage().contains("Unexpected non-terminal status"));
+ }
+
+ if (!shouldCallCancel) {
+ assertEquals(0, flinkService.getCancelJobCallCount());
+ assertNull(job.f0);
+ assertEquals(fromJobStatus, job.f1.getJobState());
+ } else {
+ assertEquals(1, flinkService.getCancelJobCallCount());
+ assertEquals("savepoint_0", job.f0);
+ assertEquals(FINISHED, job.f1.getJobState());
+ }
+ if (!shouldThrowException) {
+ assertEquals(FINISHED.name(), sessionJob.getStatus().getJobStatus().getState());
+ }
+ }
+
private Tuple3<String, JobStatusMessage, Configuration> verifyAndReturnTheSubmittedJob(
FlinkSessionJob sessionJob,
List<Tuple3<String, JobStatusMessage, Configuration>> jobs) {
@@ -471,11 +588,7 @@ public class SessionJobReconcilerTest {
FlinkSessionJob sessionJob = TestUtils.buildSessionJob();
reconciler.reconcile(sessionJob, readyContext);
verifyAndSetRunningJobsToStatus(
- sessionJob,
- JobState.RUNNING,
- org.apache.flink.api.common.JobStatus.RECONCILING.name(),
- null,
- flinkService.listJobs());
+ sessionJob, JobState.RUNNING, RECONCILING.name(), null, flinkService.listJobs());
FlinkSessionJob spSessionJob = ReconciliationUtils.clone(sessionJob);
spSessionJob