You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2018/03/03 07:27:58 UTC
[6/6] flink git commit: [FLINK-8459][flip6] Implement
RestClusterClient.cancelWithSavepoint
[FLINK-8459][flip6] Implement RestClusterClient.cancelWithSavepoint
Introduce cancelJob flag to existing triggerSavepoint methods in Dispatcher and
JobMaster. Stop checkpoint scheduler before taking savepoint to make sure that
the savepoint created by this command is the last one.
This closes #5622.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/11c30c8c
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/11c30c8c
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/11c30c8c
Branch: refs/heads/release-1.5
Commit: 11c30c8cc9e1628032879731a6e851953754b8db
Parents: 1916e26
Author: gyao <ga...@data-artisans.com>
Authored: Fri Mar 2 15:11:36 2018 +0100
Committer: Till Rohrmann <tr...@apache.org>
Committed: Sat Mar 3 00:23:53 2018 +0100
----------------------------------------------------------------------
.../flink/client/program/MiniClusterClient.java | 2 +-
.../client/program/rest/RestClusterClient.java | 15 +-
.../flink/runtime/dispatcher/Dispatcher.java | 3 +-
.../runtime/executiongraph/ExecutionGraph.java | 1 +
.../flink/runtime/jobmaster/JobMaster.java | 45 +++-
.../runtime/jobmaster/JobMasterGateway.java | 1 +
.../flink/runtime/minicluster/MiniCluster.java | 11 +
.../job/savepoints/SavepointHandlers.java | 3 +-
.../savepoints/SavepointTriggerRequestBody.java | 12 +-
.../runtime/webmonitor/RestfulGateway.java | 1 +
.../utils/TestingJobMasterGateway.java | 4 +-
...stractAsynchronousOperationHandlersTest.java | 2 +-
.../job/savepoints/SavepointHandlersTest.java | 2 +-
.../SavepointTriggerRequestBodyTest.java | 25 ++-
.../webmonitor/TestingRestfulGateway.java | 2 +-
.../jobmaster/JobMasterTriggerSavepointIT.java | 219 +++++++++++++++++++
16 files changed, 323 insertions(+), 25 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/11c30c8c/flink-clients/src/main/java/org/apache/flink/client/program/MiniClusterClient.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/program/MiniClusterClient.java b/flink-clients/src/main/java/org/apache/flink/client/program/MiniClusterClient.java
index aca75e0..dd99f0d 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/program/MiniClusterClient.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/MiniClusterClient.java
@@ -87,7 +87,7 @@ public class MiniClusterClient extends ClusterClient<MiniClusterClient.MiniClust
@Override
public String cancelWithSavepoint(JobID jobId, @Nullable String savepointDirectory) throws Exception {
- throw new UnsupportedOperationException("MiniClusterClient does not yet support this operation.");
+ return miniCluster.triggerSavepoint(jobId, savepointDirectory, true).get();
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/11c30c8c/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java b/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java
index 5a0936d..560a10e 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java
@@ -354,24 +354,29 @@ public class RestClusterClient<T> extends ClusterClient<T> {
@Override
public String cancelWithSavepoint(JobID jobId, @Nullable String savepointDirectory) throws Exception {
- throw new UnsupportedOperationException("Not implemented yet.");
+ return triggerSavepoint(jobId, savepointDirectory, true).get();
}
@Override
public CompletableFuture<String> triggerSavepoint(
final JobID jobId,
final @Nullable String savepointDirectory) throws FlinkException {
+ return triggerSavepoint(jobId, savepointDirectory, false);
+ }
+
+ private CompletableFuture<String> triggerSavepoint(
+ final JobID jobId,
+ final @Nullable String savepointDirectory,
+ final boolean cancelJob) {
final SavepointTriggerHeaders savepointTriggerHeaders = SavepointTriggerHeaders.getInstance();
final SavepointTriggerMessageParameters savepointTriggerMessageParameters =
savepointTriggerHeaders.getUnresolvedMessageParameters();
savepointTriggerMessageParameters.jobID.resolve(jobId);
- final CompletableFuture<TriggerResponse> responseFuture;
-
- responseFuture = sendRequest(
+ final CompletableFuture<TriggerResponse> responseFuture = sendRequest(
savepointTriggerHeaders,
savepointTriggerMessageParameters,
- new SavepointTriggerRequestBody(savepointDirectory));
+ new SavepointTriggerRequestBody(savepointDirectory, cancelJob));
return responseFuture.thenCompose(savepointTriggerResponseBody -> {
final TriggerId savepointTriggerId = savepointTriggerResponseBody.getTriggerId();
http://git-wip-us.apache.org/repos/asf/flink/blob/11c30c8c/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
index 7a11cf0..9b2411c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
@@ -487,11 +487,12 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
public CompletableFuture<String> triggerSavepoint(
final JobID jobId,
final String targetDirectory,
+ final boolean cancelJob,
final Time timeout) {
if (jobManagerRunners.containsKey(jobId)) {
return jobManagerRunners.get(jobId)
.getJobManagerGateway()
- .triggerSavepoint(targetDirectory, timeout);
+ .triggerSavepoint(targetDirectory, cancelJob, timeout);
} else {
return FutureUtils.completedExceptionally(new FlinkJobNotFoundException(jobId));
}
http://git-wip-us.apache.org/repos/asf/flink/blob/11c30c8c/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
index 4e8b972..ee23884 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
@@ -524,6 +524,7 @@ public class ExecutionGraph implements AccessExecutionGraph {
}
}
+ @Nullable
public CheckpointCoordinator getCheckpointCoordinator() {
return checkpointCoordinator;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/11c30c8c/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
index cc4cdba..74f9b65 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
@@ -972,19 +972,43 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
@Override
public CompletableFuture<String> triggerSavepoint(
@Nullable final String targetDirectory,
+ final boolean cancelJob,
final Time timeout) {
+
final CheckpointCoordinator checkpointCoordinator = executionGraph.getCheckpointCoordinator();
+ if (checkpointCoordinator == null) {
+ return FutureUtils.completedExceptionally(new IllegalStateException(
+ String.format("Job %s is not a streaming job.", jobGraph.getJobID())));
+ }
- if (checkpointCoordinator != null) {
- return checkpointCoordinator
- .triggerSavepoint(System.currentTimeMillis(), targetDirectory)
- .thenApply(CompletedCheckpoint::getExternalPointer);
- } else {
- return FutureUtils.completedExceptionally(
- new FlinkException(
- String.format(
- "Cannot trigger a savepoint because the job %s is not a streaming job.",
- jobGraph.getJobID())));
+ if (cancelJob) {
+ checkpointCoordinator.stopCheckpointScheduler();
+ }
+ return checkpointCoordinator
+ .triggerSavepoint(System.currentTimeMillis(), targetDirectory)
+ .thenApply(CompletedCheckpoint::getExternalPointer)
+ .thenApplyAsync(path -> {
+ if (cancelJob) {
+ log.info("Savepoint stored in {}. Now cancelling {}.", path, jobGraph.getJobID());
+ cancel(timeout);
+ }
+ return path;
+ }, getMainThreadExecutor())
+ .exceptionally(throwable -> {
+ if (cancelJob) {
+ startCheckpointScheduler(checkpointCoordinator);
+ }
+ throw new CompletionException(throwable);
+ });
+ }
+
+ private void startCheckpointScheduler(final CheckpointCoordinator checkpointCoordinator) {
+ if (checkpointCoordinator.isPeriodicCheckpointingConfigured()) {
+ try {
+ checkpointCoordinator.startCheckpointScheduler();
+ } catch (IllegalStateException ignored) {
+ // Concurrent shut down of the coordinator
+ }
}
}
@@ -1321,6 +1345,7 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
private CompletableFuture<String> getJobModificationSavepoint(Time timeout) {
return triggerSavepoint(
null,
+ false,
timeout)
.handleAsync(
(String savepointPath, Throwable throwable) -> {
http://git-wip-us.apache.org/repos/asf/flink/blob/11c30c8c/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterGateway.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterGateway.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterGateway.java
index 6173a26..1e1bdda 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterGateway.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterGateway.java
@@ -263,6 +263,7 @@ public interface JobMasterGateway extends
*/
CompletableFuture<String> triggerSavepoint(
@Nullable final String targetDirectory,
+ final boolean cancelJob,
final Time timeout);
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/11c30c8c/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java b/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java
index 6b5f9b5..98c8ca2 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java
@@ -483,6 +483,17 @@ public class MiniCluster implements JobExecutorService, AutoCloseableAsync {
}
}
+ public CompletableFuture<String> triggerSavepoint(JobID jobId, String targetDirectory, boolean cancelJob) {
+ try {
+ return getDispatcherGateway().triggerSavepoint(jobId, targetDirectory, cancelJob, rpcTimeout);
+ } catch (LeaderRetrievalException | InterruptedException e) {
+ return FutureUtils.completedExceptionally(
+ new FlinkException(
+ String.format("Could not trigger savepoint for job %s.", jobId),
+ e));
+ }
+ }
+
// ------------------------------------------------------------------------
// running jobs
// ------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/11c30c8c/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/savepoints/SavepointHandlers.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/savepoints/SavepointHandlers.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/savepoints/SavepointHandlers.java
index cb3ff5b..17e263b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/savepoints/SavepointHandlers.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/savepoints/SavepointHandlers.java
@@ -127,8 +127,9 @@ public class SavepointHandlers extends AbstractAsynchronousOperationHandlers<Asy
HttpResponseStatus.BAD_REQUEST);
}
+ final boolean cancelJob = request.getRequestBody().isCancelJob();
final String targetDirectory = requestedTargetDirectory != null ? requestedTargetDirectory : defaultSavepointDir;
- return gateway.triggerSavepoint(jobId, targetDirectory, RpcUtils.INF_TIMEOUT);
+ return gateway.triggerSavepoint(jobId, targetDirectory, cancelJob, RpcUtils.INF_TIMEOUT);
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/11c30c8c/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/savepoints/SavepointTriggerRequestBody.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/savepoints/SavepointTriggerRequestBody.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/savepoints/SavepointTriggerRequestBody.java
index 16062a2..feaf00e 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/savepoints/SavepointTriggerRequestBody.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/savepoints/SavepointTriggerRequestBody.java
@@ -32,14 +32,21 @@ public class SavepointTriggerRequestBody implements RequestBody {
public static final String FIELD_NAME_TARGET_DIRECTORY = "target-directory";
+ private static final String FIELD_NAME_CANCEL_JOB = "cancel-job";
+
@JsonProperty(FIELD_NAME_TARGET_DIRECTORY)
@Nullable
private final String targetDirectory;
+ @JsonProperty(FIELD_NAME_CANCEL_JOB)
+ private final boolean cancelJob;
+
@JsonCreator
public SavepointTriggerRequestBody(
- @Nullable @JsonProperty(FIELD_NAME_TARGET_DIRECTORY) final String targetDirectory) {
+ @Nullable @JsonProperty(FIELD_NAME_TARGET_DIRECTORY) final String targetDirectory,
+ @JsonProperty(value = FIELD_NAME_CANCEL_JOB, defaultValue = "false") final boolean cancelJob) {
this.targetDirectory = targetDirectory;
+ this.cancelJob = cancelJob;
}
@Nullable
@@ -47,4 +54,7 @@ public class SavepointTriggerRequestBody implements RequestBody {
return targetDirectory;
}
+ public boolean isCancelJob() {
+ return cancelJob;
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/11c30c8c/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/RestfulGateway.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/RestfulGateway.java b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/RestfulGateway.java
index ed90f37..4714206 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/RestfulGateway.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/RestfulGateway.java
@@ -139,6 +139,7 @@ public interface RestfulGateway extends RpcGateway {
default CompletableFuture<String> triggerSavepoint(
JobID jobId,
String targetDirectory,
+ boolean cancelJob,
@RpcTimeout Time timeout) {
throw new UnsupportedOperationException();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/11c30c8c/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/utils/TestingJobMasterGateway.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/utils/TestingJobMasterGateway.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/utils/TestingJobMasterGateway.java
index cac7e90..0d57a56 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/utils/TestingJobMasterGateway.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/utils/TestingJobMasterGateway.java
@@ -48,6 +48,8 @@ import org.apache.flink.runtime.taskexecutor.slot.SlotOffer;
import org.apache.flink.runtime.taskmanager.TaskExecutionState;
import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
+import javax.annotation.Nullable;
+
import java.net.InetSocketAddress;
import java.util.Collection;
import java.util.concurrent.CompletableFuture;
@@ -153,7 +155,7 @@ public class TestingJobMasterGateway implements JobMasterGateway {
}
@Override
- public CompletableFuture<String> triggerSavepoint(String targetDirectory, Time timeout) {
+ public CompletableFuture<String> triggerSavepoint(@Nullable final String targetDirectory, final boolean cancelJob, final Time timeout) {
throw new UnsupportedOperationException();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/11c30c8c/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/async/AbstractAsynchronousOperationHandlersTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/async/AbstractAsynchronousOperationHandlersTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/async/AbstractAsynchronousOperationHandlersTest.java
index 848e253..7ad140e 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/async/AbstractAsynchronousOperationHandlersTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/async/AbstractAsynchronousOperationHandlersTest.java
@@ -305,7 +305,7 @@ public class AbstractAsynchronousOperationHandlersTest extends TestLogger {
@Override
protected CompletableFuture<String> triggerOperation(HandlerRequest<EmptyRequestBody, EmptyMessageParameters> request, RestfulGateway gateway) throws RestHandlerException {
- return gateway.triggerSavepoint(new JobID(), null, timeout);
+ return gateway.triggerSavepoint(new JobID(), null, false, timeout);
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/11c30c8c/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/savepoints/SavepointHandlersTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/savepoints/SavepointHandlersTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/savepoints/SavepointHandlersTest.java
index a8f4b3f..0694452 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/savepoints/SavepointHandlersTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/savepoints/SavepointHandlersTest.java
@@ -200,7 +200,7 @@ public class SavepointHandlersTest extends TestLogger {
final String targetDirectory
) throws HandlerRequestException {
return new HandlerRequest<>(
- new SavepointTriggerRequestBody(targetDirectory),
+ new SavepointTriggerRequestBody(targetDirectory, false),
new SavepointTriggerMessageParameters(),
Collections.singletonMap(JobIDPathParameter.KEY, JOB_ID.toString()),
Collections.emptyMap());
http://git-wip-us.apache.org/repos/asf/flink/blob/11c30c8c/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/job/savepoints/SavepointTriggerRequestBodyTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/job/savepoints/SavepointTriggerRequestBodyTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/job/savepoints/SavepointTriggerRequestBodyTest.java
index f7c3973..f79f8a2 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/job/savepoints/SavepointTriggerRequestBodyTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/job/savepoints/SavepointTriggerRequestBodyTest.java
@@ -20,22 +20,43 @@ package org.apache.flink.runtime.rest.messages.job.savepoints;
import org.apache.flink.runtime.rest.messages.RestRequestMarshallingTestBase;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.util.Arrays;
+import java.util.Collection;
+
import static org.junit.Assert.assertEquals;
/**
* Tests for {@link SavepointTriggerRequestBody}.
*/
+@RunWith(Parameterized.class)
public class SavepointTriggerRequestBodyTest
extends RestRequestMarshallingTestBase<SavepointTriggerRequestBody> {
+ private final SavepointTriggerRequestBody savepointTriggerRequestBody;
+
+ public SavepointTriggerRequestBodyTest(final SavepointTriggerRequestBody savepointTriggerRequestBody) {
+ this.savepointTriggerRequestBody = savepointTriggerRequestBody;
+ }
+
+ @Parameterized.Parameters
+ public static Collection<Object[]> data() {
+ return Arrays.asList(new Object[][]{
+ {new SavepointTriggerRequestBody("/tmp", true)},
+ {new SavepointTriggerRequestBody("/tmp", false)}
+ });
+ }
+
@Override
protected Class<SavepointTriggerRequestBody> getTestRequestClass() {
return SavepointTriggerRequestBody.class;
}
@Override
- protected SavepointTriggerRequestBody getTestRequestInstance() throws Exception {
- return new SavepointTriggerRequestBody("/tmp");
+ protected SavepointTriggerRequestBody getTestRequestInstance() {
+ return savepointTriggerRequestBody;
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/11c30c8c/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/TestingRestfulGateway.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/TestingRestfulGateway.java b/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/TestingRestfulGateway.java
index 5eff5a6..b92ba51 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/TestingRestfulGateway.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/TestingRestfulGateway.java
@@ -190,7 +190,7 @@ public class TestingRestfulGateway implements RestfulGateway {
}
@Override
- public CompletableFuture<String> triggerSavepoint(JobID jobId, String targetDirectory, Time timeout) {
+ public CompletableFuture<String> triggerSavepoint(JobID jobId, String targetDirectory, boolean cancelJob, Time timeout) {
return triggerSavepointFunction.apply(jobId, targetDirectory);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/11c30c8c/flink-tests/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTriggerSavepointIT.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTriggerSavepointIT.java b/flink-tests/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTriggerSavepointIT.java
new file mode 100644
index 0000000..f9edfa6
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTriggerSavepointIT.java
@@ -0,0 +1,219 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.jobmaster;
+
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.client.program.MiniClusterClient;
+import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
+import org.apache.flink.runtime.checkpoint.CheckpointMetrics;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.checkpoint.CheckpointRetentionPolicy;
+import org.apache.flink.runtime.checkpoint.CheckpointTriggerException;
+import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
+import org.apache.flink.runtime.checkpoint.TaskStateSnapshot;
+import org.apache.flink.runtime.execution.Environment;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.jobgraph.JobStatus;
+import org.apache.flink.runtime.jobgraph.JobVertex;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
+import org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration;
+import org.apache.flink.runtime.jobgraph.tasks.JobCheckpointingSettings;
+import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.flink.testutils.category.Flip6;
+import org.apache.flink.util.ExceptionUtils;
+
+import org.junit.Assume;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.hasItem;
+import static org.hamcrest.Matchers.isOneOf;
+
+/**
+ * Tests for {@link org.apache.flink.runtime.jobmaster.JobMaster#triggerSavepoint(String, boolean, Time)}.
+ *
+ * @see org.apache.flink.runtime.jobmaster.JobMaster
+ */
+@Category(Flip6.class)
+public class JobMasterTriggerSavepointIT extends AbstractTestBase {
+
+ private static CountDownLatch invokeLatch;
+
+ private static volatile CountDownLatch triggerCheckpointLatch;
+
+ @Rule
+ public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ private Path savepointDirectory;
+ private MiniClusterClient clusterClient;
+ private JobGraph jobGraph;
+
+ @Before
+ public void setUp() throws Exception {
+ invokeLatch = new CountDownLatch(1);
+ triggerCheckpointLatch = new CountDownLatch(1);
+ savepointDirectory = temporaryFolder.newFolder().toPath();
+
+ Assume.assumeTrue(
+ "ClusterClient is not an instance of MiniClusterClient",
+ miniClusterResource.getClusterClient() instanceof MiniClusterClient);
+
+ clusterClient = (MiniClusterClient) miniClusterResource.getClusterClient();
+ clusterClient.setDetached(true);
+
+ jobGraph = new JobGraph();
+
+ final JobVertex vertex = new JobVertex("testVertex");
+ vertex.setInvokableClass(NoOpBlockingInvokable.class);
+ jobGraph.addVertex(vertex);
+
+ jobGraph.setSnapshotSettings(new JobCheckpointingSettings(
+ Collections.singletonList(vertex.getID()),
+ Collections.singletonList(vertex.getID()),
+ Collections.singletonList(vertex.getID()),
+ new CheckpointCoordinatorConfiguration(
+ 10,
+ 60_000,
+ 10,
+ 1,
+ CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION,
+ true),
+ null
+ ));
+
+ clusterClient.submitJob(jobGraph, ClassLoader.getSystemClassLoader());
+ invokeLatch.await(60, TimeUnit.SECONDS);
+ waitForJob();
+ }
+
+ @Test
+ public void testStopJobAfterSavepoint() throws Exception {
+ final String savepointLocation = cancelWithSavepoint();
+ final JobStatus jobStatus = clusterClient.getJobStatus(jobGraph.getJobID()).get(60, TimeUnit.SECONDS);
+
+ assertThat(jobStatus, isOneOf(JobStatus.CANCELED, JobStatus.CANCELLING));
+
+ final List<Path> savepoints = Files.list(savepointDirectory).map(Path::getFileName).collect(Collectors.toList());
+ assertThat(savepoints, hasItem(Paths.get(savepointLocation).getFileName()));
+ }
+
+ @Test
+ public void testDoNotCancelJobIfSavepointFails() throws Exception {
+ try {
+ Files.setPosixFilePermissions(savepointDirectory, Collections.emptySet());
+ } catch (IOException e) {
+ Assume.assumeNoException(e);
+ }
+
+ try {
+ cancelWithSavepoint();
+ } catch (Exception e) {
+ assertThat(ExceptionUtils.findThrowable(e, CheckpointTriggerException.class).isPresent(), equalTo(true));
+ }
+
+ final JobStatus jobStatus = clusterClient.getJobStatus(jobGraph.getJobID()).get(60, TimeUnit.SECONDS);
+ assertThat(jobStatus, equalTo(JobStatus.RUNNING));
+
+ // assert that checkpoints are continued to be triggered
+ triggerCheckpointLatch = new CountDownLatch(1);
+ assertThat(triggerCheckpointLatch.await(60, TimeUnit.SECONDS), equalTo(true));
+ }
+
+ private void waitForJob() throws Exception {
+ for (int i = 0; i < 60; i++) {
+ try {
+ final JobStatus jobStatus = clusterClient.getJobStatus(jobGraph.getJobID()).get(60, TimeUnit.SECONDS);
+ assertThat(jobStatus.isGloballyTerminalState(), equalTo(false));
+ if (jobStatus == JobStatus.RUNNING) {
+ return;
+ }
+ } catch (ExecutionException ignored) {
+ // JobManagerRunner is not yet registered in Dispatcher
+ }
+ Thread.sleep(1000);
+ }
+ throw new AssertionError("Job did not become running within timeout.");
+ }
+
+ /**
+ * Invokable which calls {@link CountDownLatch#countDown()} on
+ * {@link JobMasterTriggerSavepointIT#invokeLatch}, and then blocks afterwards.
+ */
+ public static class NoOpBlockingInvokable extends AbstractInvokable {
+
+ public NoOpBlockingInvokable(final Environment environment) {
+ super(environment);
+ }
+
+ @Override
+ public void invoke() {
+ invokeLatch.countDown();
+ try {
+ Thread.sleep(Long.MAX_VALUE);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ @Override
+ public boolean triggerCheckpoint(final CheckpointMetaData checkpointMetaData, final CheckpointOptions checkpointOptions) throws Exception {
+ final TaskStateSnapshot checkpointStateHandles = new TaskStateSnapshot();
+ checkpointStateHandles.putSubtaskStateByOperatorID(
+ OperatorID.fromJobVertexID(getEnvironment().getJobVertexId()),
+ new OperatorSubtaskState());
+
+ getEnvironment().acknowledgeCheckpoint(
+ checkpointMetaData.getCheckpointId(),
+ new CheckpointMetrics(),
+ checkpointStateHandles);
+
+ triggerCheckpointLatch.countDown();
+
+ return true;
+ }
+
+ @Override
+ public void notifyCheckpointComplete(final long checkpointId) throws Exception {
+ }
+ }
+
+ private String cancelWithSavepoint() throws Exception {
+ return clusterClient.cancelWithSavepoint(
+ jobGraph.getJobID(),
+ savepointDirectory.toAbsolutePath().toString());
+ }
+
+}