You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by mx...@apache.org on 2023/11/13 10:58:37 UTC
(flink-kubernetes-operator) branch main updated: [FLINK-33522] Use created savepoint even if job cancellation fails (#706)
This is an automated email from the ASF dual-hosted git repository.
mxm 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 2ce7486c [FLINK-33522] Use created savepoint even if job cancellation fails (#706)
2ce7486c is described below
commit 2ce7486c7ee7ea6cb1a4160585fb5bc696025378
Author: Maximilian Michels <mx...@apache.org>
AuthorDate: Mon Nov 13 02:58:32 2023 -0800
[FLINK-33522] Use created savepoint even if job cancellation fails (#706)
---
.../operator/service/AbstractFlinkService.java | 18 +++++++++++
.../operator/service/AbstractFlinkServiceTest.java | 37 +++++++++++++++++++---
2 files changed, 50 insertions(+), 5 deletions(-)
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 399e692d..0821b439 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
@@ -80,6 +80,7 @@ import org.apache.flink.runtime.rest.messages.queue.QueueStatus;
import org.apache.flink.runtime.rest.messages.taskmanager.TaskManagersHeaders;
import org.apache.flink.runtime.rest.messages.taskmanager.TaskManagersInfo;
import org.apache.flink.runtime.rest.util.RestConstants;
+import org.apache.flink.runtime.scheduler.stopwithsavepoint.StopWithSavepointStoppingException;
import org.apache.flink.runtime.state.memory.NonPersistentMetadataCheckpointStorageLocation;
import org.apache.flink.runtime.webmonitor.handlers.JarDeleteHeaders;
import org.apache.flink.runtime.webmonitor.handlers.JarDeleteMessageParameters;
@@ -90,6 +91,7 @@ import org.apache.flink.runtime.webmonitor.handlers.JarUploadHeaders;
import org.apache.flink.runtime.webmonitor.handlers.JarUploadResponseBody;
import org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
import org.apache.flink.util.ConfigurationException;
+import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.FileUtils;
import org.apache.flink.util.FlinkException;
import org.apache.flink.util.FlinkRuntimeException;
@@ -327,6 +329,22 @@ public abstract class AbstractFlinkService implements FlinkService {
ExecutionCheckpointingOptions.CHECKPOINTING_TIMEOUT
.key()),
exception);
+ } catch (Exception e) {
+ var stopWithSavepointException =
+ ExceptionUtils.findThrowable(
+ e, StopWithSavepointStoppingException.class);
+ if (stopWithSavepointException.isPresent()) {
+ // Handle edge case where the savepoint completes but the job fails
+ // right afterward.
+ savepointOpt =
+ Optional.of(
+ stopWithSavepointException
+ .get()
+ .getSavepointPath());
+ } else {
+ // Rethrow if savepoint was not completed successfully.
+ throw e;
+ }
}
} else if (ReconciliationUtils.isJobInTerminalState(deploymentStatus)) {
LOG.info(
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkServiceTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkServiceTest.java
index 2e0b17e9..c12caf21 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkServiceTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkServiceTest.java
@@ -80,6 +80,7 @@ import org.apache.flink.runtime.rest.messages.taskmanager.TaskManagerInfo;
import org.apache.flink.runtime.rest.messages.taskmanager.TaskManagersHeaders;
import org.apache.flink.runtime.rest.messages.taskmanager.TaskManagersInfo;
import org.apache.flink.runtime.rest.util.RestMapperUtils;
+import org.apache.flink.runtime.scheduler.stopwithsavepoint.StopWithSavepointStoppingException;
import org.apache.flink.runtime.taskexecutor.TaskExecutorMemoryConfiguration;
import org.apache.flink.runtime.webmonitor.handlers.JarDeleteHeaders;
import org.apache.flink.runtime.webmonitor.handlers.JarRunRequestBody;
@@ -116,11 +117,13 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
+import static org.apache.flink.kubernetes.operator.api.status.SavepointFormatType.NATIVE;
import static org.apache.flink.kubernetes.operator.config.FlinkConfigBuilder.FLINK_VERSION;
import static org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.OPERATOR_SAVEPOINT_FORMAT_TYPE;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -611,8 +614,19 @@ public class AbstractFlinkServiceTest {
@Test
public void nativeSavepointFormatTest() throws Exception {
+ runNativeSavepointFormatTest(false);
+ }
+
+ @Test
+ public void testSavepointCompletesButJobFailsAfterwards() throws Exception {
+ runNativeSavepointFormatTest(true);
+ }
+
+ private void runNativeSavepointFormatTest(boolean failAfterSavepointCompletes)
+ throws Exception {
final TestingClusterClient<String> testingClusterClient =
new TestingClusterClient<>(configuration, TestUtils.TEST_DEPLOYMENT_NAME);
+ final JobID jobID = JobID.generate();
final String savepointPath = "file:///path/of/svp";
final CompletableFuture<Tuple4<JobID, String, Boolean, SavepointFormatType>>
triggerSavepointFuture = new CompletableFuture<>();
@@ -634,13 +648,20 @@ public class AbstractFlinkServiceTest {
stopWithSavepointFuture = new CompletableFuture<>();
testingClusterClient.setStopWithSavepointFormat(
(id, formatType, savepointDir) -> {
- stopWithSavepointFuture.complete(new Tuple3<>(id, formatType, savepointDir));
+ if (failAfterSavepointCompletes) {
+ stopWithSavepointFuture.completeExceptionally(
+ new CompletionException(
+ new StopWithSavepointStoppingException(
+ savepointPath, jobID)));
+ } else {
+ stopWithSavepointFuture.complete(
+ new Tuple3<>(id, formatType, savepointDir));
+ }
return CompletableFuture.completedFuture(savepointPath);
});
var flinkService = new TestingService(testingClusterClient);
- final JobID jobID = JobID.generate();
final FlinkDeployment deployment = TestUtils.buildApplicationCluster();
deployment
.getSpec()
@@ -672,10 +693,16 @@ public class AbstractFlinkServiceTest {
new Configuration(configManager.getObserveConfig(deployment))
.set(OPERATOR_SAVEPOINT_FORMAT_TYPE, SavepointFormatType.NATIVE),
false);
+
assertTrue(stopWithSavepointFuture.isDone());
- assertEquals(jobID, stopWithSavepointFuture.get().f0);
- assertEquals(SavepointFormatType.NATIVE, stopWithSavepointFuture.get().f1);
- assertEquals(savepointPath, stopWithSavepointFuture.get().f2);
+ assertEquals(
+ failAfterSavepointCompletes, stopWithSavepointFuture.isCompletedExceptionally());
+
+ var lastSavepoint =
+ deployment.getStatus().getJobStatus().getSavepointInfo().getLastSavepoint();
+ assertEquals(NATIVE, lastSavepoint.getFormatType());
+ assertEquals(savepointPath, lastSavepoint.getLocation());
+ assertEquals(jobID.toHexString(), deployment.getStatus().getJobStatus().getJobId());
}
@Test