You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ch...@apache.org on 2021/12/14 10:25:07 UTC
[flink] branch master updated: [FLINK-24232][coordination] Skip history server archiving for suspended jobs
This is an automated email from the ASF dual-hosted git repository.
chesnay pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new efa3362 [FLINK-24232][coordination] Skip history server archiving for suspended jobs
efa3362 is described below
commit efa336294f4280401d9933158dd499f377a63c2a
Author: Chesnay Schepler <ch...@apache.org>
AuthorDate: Thu Dec 9 12:27:20 2021 +0100
[FLINK-24232][coordination] Skip history server archiving for suspended jobs
Do not create an archive for suspended jobs, as this would eventually lead to multiple archive attempts which we currently do not support.
---
.../flink/runtime/dispatcher/Dispatcher.java | 30 +++++++++-------
.../runtime/dispatcher/AbstractDispatcherTest.java | 10 +++++-
.../flink/runtime/dispatcher/DispatcherTest.java | 40 ++++++++++++++++++++++
3 files changed, 66 insertions(+), 14 deletions(-)
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 89d64f2..7b21b6e 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
@@ -943,19 +943,23 @@ public abstract class Dispatcher extends PermanentlyFencedRpcEndpoint<Dispatcher
e);
}
- final CompletableFuture<Acknowledge> executionGraphFuture =
- historyServerArchivist.archiveExecutionGraph(executionGraphInfo);
-
- executionGraphFuture.whenComplete(
- (Acknowledge ignored, Throwable throwable) -> {
- if (throwable != null) {
- log.info(
- "Could not archive completed job {}({}) to the history server.",
- executionGraphInfo.getArchivedExecutionGraph().getJobName(),
- executionGraphInfo.getArchivedExecutionGraph().getJobID(),
- throwable);
- }
- });
+ // do not create an archive for suspended jobs, as this would eventually lead to multiple
+ // archive attempts which we currently do not support
+ if (executionGraphInfo.getArchivedExecutionGraph().getState().isGloballyTerminalState()) {
+ final CompletableFuture<Acknowledge> executionGraphFuture =
+ historyServerArchivist.archiveExecutionGraph(executionGraphInfo);
+
+ executionGraphFuture.whenComplete(
+ (Acknowledge ignored, Throwable throwable) -> {
+ if (throwable != null) {
+ log.info(
+ "Could not archive completed job {}({}) to the history server.",
+ executionGraphInfo.getArchivedExecutionGraph().getJobName(),
+ executionGraphInfo.getArchivedExecutionGraph().getJobID(),
+ throwable);
+ }
+ });
+ }
}
private void jobMasterFailed(JobID jobId, Throwable cause) {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/AbstractDispatcherTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/AbstractDispatcherTest.java
index 5ee43ae..f10e55f 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/AbstractDispatcherTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/AbstractDispatcherTest.java
@@ -145,6 +145,8 @@ public class AbstractDispatcherTest extends TestLogger {
private FatalErrorHandler fatalErrorHandler =
testingFatalErrorHandlerResource.getFatalErrorHandler();
+ private HistoryServerArchivist historyServerArchivist = VoidHistoryServerArchivist.INSTANCE;
+
TestingDispatcherBuilder setHeartbeatServices(HeartbeatServices heartbeatServices) {
this.heartbeatServices = heartbeatServices;
return this;
@@ -176,6 +178,12 @@ public class AbstractDispatcherTest extends TestLogger {
return this;
}
+ public TestingDispatcherBuilder setHistoryServerArchivist(
+ HistoryServerArchivist historyServerArchivist) {
+ this.historyServerArchivist = historyServerArchivist;
+ return this;
+ }
+
TestingDispatcher build() throws Exception {
TestingResourceManagerGateway resourceManagerGateway =
new TestingResourceManagerGateway();
@@ -196,7 +204,7 @@ public class AbstractDispatcherTest extends TestLogger {
heartbeatServices,
executionGraphInfoStore,
fatalErrorHandler,
- VoidHistoryServerArchivist.INSTANCE,
+ historyServerArchivist,
null,
new DispatcherOperationCaches(),
UnregisteredMetricGroups.createUnregisteredJobManagerMetricGroup(),
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java
index e209694..6350ab0 100755
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java
@@ -452,6 +452,46 @@ public class DispatcherTest extends AbstractDispatcherTest {
}
@Test
+ public void testNoHistoryServerArchiveCreatedForSuspendedJob() throws Exception {
+ final CompletableFuture<Void> archiveAttemptFuture = new CompletableFuture<>();
+ final CompletableFuture<JobManagerRunnerResult> jobTerminationFuture =
+ new CompletableFuture<>();
+ dispatcher =
+ new TestingDispatcherBuilder()
+ .setJobManagerRunnerFactory(
+ new FinishingJobManagerRunnerFactory(
+ jobTerminationFuture, () -> {}))
+ .setHistoryServerArchivist(
+ executionGraphInfo -> {
+ archiveAttemptFuture.complete(null);
+ return CompletableFuture.completedFuture(null);
+ })
+ .build();
+ dispatcher.start();
+ jobMasterLeaderElectionService.isLeader(UUID.randomUUID());
+ DispatcherGateway dispatcherGateway = dispatcher.getSelfGateway(DispatcherGateway.class);
+
+ JobID jobId = jobGraph.getJobID();
+
+ dispatcherGateway.submitJob(jobGraph, TIMEOUT).get();
+ jobTerminationFuture.complete(
+ JobManagerRunnerResult.forSuccess(
+ new ExecutionGraphInfo(
+ new ArchivedExecutionGraphBuilder()
+ .setJobID(jobId)
+ .setState(JobStatus.SUSPENDED)
+ .build())));
+
+ // wait for job to finish
+ dispatcherGateway.requestJobResult(jobId, TIMEOUT).get();
+ // sanity check
+ assertThat(
+ dispatcherGateway.requestJobStatus(jobId, TIMEOUT).get(), is(JobStatus.SUSPENDED));
+
+ assertThat(archiveAttemptFuture.isDone(), is(false));
+ }
+
+ @Test
public void testJobManagerRunnerInitializationFailureFailsJob() throws Exception {
final TestingJobManagerRunnerFactory testingJobManagerRunnerFactory =
new TestingJobManagerRunnerFactory();