You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ma...@apache.org on 2022/04/28 12:01:43 UTC
[flink] branch release-1.14 updated (b5d77c0cdb1 -> f9d2c1c0d3a)
This is an automated email from the ASF dual-hosted git repository.
mapohl pushed a change to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git
from b5d77c0cdb1 [FLINK-27409][runtime] Cleanup stale slot allocation record when the resource requirement of a job is empty
new 1c492ed97fe [hotfix][tests] Allow retrieval of termination future for running jobs
new e0fb11741a0 [hotfix][tests] Wait for JobManagerRunner termination
new f9d2c1c0d3a [FLINK-24491][runtime] Make the job termination wait until the archiving of ExecutionGraphInfo finishes
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../flink/runtime/dispatcher/Dispatcher.java | 82 ++++++++++++----------
.../flink/runtime/dispatcher/MiniDispatcher.java | 43 +++++++-----
.../dispatcher/DispatcherResourceCleanupTest.java | 75 +++++++++++++++++++-
.../flink/runtime/dispatcher/DispatcherTest.java | 8 ++-
4 files changed, 149 insertions(+), 59 deletions(-)
[flink] 03/03: [FLINK-24491][runtime] Make the job termination wait until the archiving of ExecutionGraphInfo finishes
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mapohl pushed a commit to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git
commit f9d2c1c0d3aa7e72e95f1c75bcf7c77c1fceac22
Author: Thesharing <cy...@outlook.com>
AuthorDate: Wed Mar 30 11:18:03 2022 +0800
[FLINK-24491][runtime] Make the job termination wait until the archiving of ExecutionGraphInfo finishes
---
.../flink/runtime/dispatcher/Dispatcher.java | 70 +++++++++++---------
.../flink/runtime/dispatcher/MiniDispatcher.java | 43 +++++++------
.../dispatcher/DispatcherResourceCleanupTest.java | 75 +++++++++++++++++++++-
3 files changed, 140 insertions(+), 48 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 c09ba52ca9f..ff73ddaaaa5 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
@@ -428,10 +428,12 @@ public abstract class Dispatcher extends PermanentlyFencedRpcEndpoint<Dispatcher
return handleJobManagerRunnerResult(
jobManagerRunnerResult, executionType);
} else {
- return jobManagerRunnerFailed(jobId, throwable);
+ return CompletableFuture.completedFuture(
+ jobManagerRunnerFailed(jobId, throwable));
}
},
- getMainThreadExecutor());
+ getMainThreadExecutor())
+ .thenCompose(Function.identity());
final CompletableFuture<Void> jobTerminationFuture =
cleanupJobStateFuture
@@ -444,13 +446,14 @@ public abstract class Dispatcher extends PermanentlyFencedRpcEndpoint<Dispatcher
registerJobManagerRunnerTerminationFuture(jobId, jobTerminationFuture);
}
- private CleanupJobState handleJobManagerRunnerResult(
+ private CompletableFuture<CleanupJobState> handleJobManagerRunnerResult(
JobManagerRunnerResult jobManagerRunnerResult, ExecutionType executionType) {
if (jobManagerRunnerResult.isInitializationFailure()) {
if (executionType == ExecutionType.RECOVERY) {
- return jobManagerRunnerFailed(
- jobManagerRunnerResult.getExecutionGraphInfo().getJobId(),
- jobManagerRunnerResult.getInitializationFailure());
+ return CompletableFuture.completedFuture(
+ jobManagerRunnerFailed(
+ jobManagerRunnerResult.getExecutionGraphInfo().getJobId(),
+ jobManagerRunnerResult.getInitializationFailure()));
} else {
return jobReachedTerminalState(jobManagerRunnerResult.getExecutionGraphInfo());
}
@@ -840,7 +843,8 @@ public abstract class Dispatcher extends PermanentlyFencedRpcEndpoint<Dispatcher
fatalErrorHandler.onFatalError(throwable);
}
- protected CleanupJobState jobReachedTerminalState(ExecutionGraphInfo executionGraphInfo) {
+ protected CompletableFuture<CleanupJobState> jobReachedTerminalState(
+ ExecutionGraphInfo executionGraphInfo) {
final ArchivedExecutionGraph archivedExecutionGraph =
executionGraphInfo.getArchivedExecutionGraph();
final JobStatus terminalJobStatus = archivedExecutionGraph.getState();
@@ -870,14 +874,21 @@ public abstract class Dispatcher extends PermanentlyFencedRpcEndpoint<Dispatcher
terminalJobStatus);
}
- archiveExecutionGraph(executionGraphInfo);
+ writeToExecutionGraphInfoStore(executionGraphInfo);
+
+ if (!terminalJobStatus.isGloballyTerminalState()) {
+ return CompletableFuture.completedFuture(CleanupJobState.LOCAL);
+ }
+
+ // do not create an archive for suspended jobs, as this would eventually lead to
+ // multiple archive attempts which we currently do not support
+ CompletableFuture<Acknowledge> archiveToHistoryServerFuture =
+ archiveExecutionGraphToHistoryServer(executionGraphInfo);
- return terminalJobStatus.isGloballyTerminalState()
- ? CleanupJobState.GLOBAL
- : CleanupJobState.LOCAL;
+ return archiveToHistoryServerFuture.thenApply(ignored -> CleanupJobState.GLOBAL);
}
- private void archiveExecutionGraph(ExecutionGraphInfo executionGraphInfo) {
+ private void writeToExecutionGraphInfoStore(ExecutionGraphInfo executionGraphInfo) {
try {
executionGraphInfoStore.put(executionGraphInfo);
} catch (IOException e) {
@@ -887,24 +898,25 @@ public abstract class Dispatcher extends PermanentlyFencedRpcEndpoint<Dispatcher
executionGraphInfo.getArchivedExecutionGraph().getJobID(),
e);
}
+ }
- // 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 CompletableFuture<Acknowledge> archiveExecutionGraphToHistoryServer(
+ ExecutionGraphInfo executionGraphInfo) {
+
+ return historyServerArchivist
+ .archiveExecutionGraph(executionGraphInfo)
+ .handleAsync(
+ (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);
+ }
+ return Acknowledge.get();
+ },
+ getMainThreadExecutor());
}
private void jobMasterFailed(JobID jobId, Throwable cause) {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/MiniDispatcher.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/MiniDispatcher.java
index 99bb7167757..85c8d9b2755 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/MiniDispatcher.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/MiniDispatcher.java
@@ -125,26 +125,33 @@ public class MiniDispatcher extends Dispatcher {
}
@Override
- protected CleanupJobState jobReachedTerminalState(ExecutionGraphInfo executionGraphInfo) {
+ protected CompletableFuture<CleanupJobState> jobReachedTerminalState(
+ ExecutionGraphInfo executionGraphInfo) {
final ArchivedExecutionGraph archivedExecutionGraph =
executionGraphInfo.getArchivedExecutionGraph();
- final CleanupJobState cleanupHAState = super.jobReachedTerminalState(executionGraphInfo);
-
- JobStatus jobStatus =
- Objects.requireNonNull(
- archivedExecutionGraph.getState(), "JobStatus should not be null here.");
- if (jobStatus.isGloballyTerminalState()
- && (jobCancelled || executionMode == ClusterEntrypoint.ExecutionMode.DETACHED)) {
- // shut down if job is cancelled or we don't have to wait for the execution result
- // retrieval
- log.info(
- "Shutting down cluster with state {}, jobCancelled: {}, executionMode: {}",
- jobStatus,
- jobCancelled,
- executionMode);
- shutDownFuture.complete(ApplicationStatus.fromJobStatus(jobStatus));
- }
+ final CompletableFuture<CleanupJobState> cleanupHAState =
+ super.jobReachedTerminalState(executionGraphInfo);
+
+ return cleanupHAState.thenApply(
+ cleanupJobState -> {
+ JobStatus jobStatus =
+ Objects.requireNonNull(
+ archivedExecutionGraph.getState(),
+ "JobStatus should not be null here.");
+ if (jobStatus.isGloballyTerminalState()
+ && (jobCancelled
+ || executionMode == ClusterEntrypoint.ExecutionMode.DETACHED)) {
+ // shut down if job is cancelled or we don't have to wait for the execution
+ // result retrieval
+ log.info(
+ "Shutting down cluster with state {}, jobCancelled: {}, executionMode: {}",
+ jobStatus,
+ jobCancelled,
+ executionMode);
+ shutDownFuture.complete(ApplicationStatus.fromJobStatus(jobStatus));
+ }
- return cleanupHAState;
+ return cleanupJobState;
+ });
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherResourceCleanupTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherResourceCleanupTest.java
index bef8ec2df4e..81d8d135185 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherResourceCleanupTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherResourceCleanupTest.java
@@ -84,9 +84,12 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.is;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.fail;
@@ -132,6 +135,7 @@ public class DispatcherResourceCleanupTest extends TestLogger {
private CompletableFuture<JobID> cleanupJobFuture;
private CompletableFuture<JobID> cleanupJobHADataFuture;
private JobGraphWriter jobGraphWriter = NoOpJobGraphWriter.INSTANCE;
+ private HistoryServerArchivist historyServerArchivist = VoidHistoryServerArchivist.INSTANCE;
@BeforeClass
public static void setupClass() {
@@ -213,7 +217,7 @@ public class DispatcherResourceCleanupTest extends TestLogger {
heartbeatServices,
archivedExecutionGraphStore,
testingFatalErrorHandlerResource.getFatalErrorHandler(),
- VoidHistoryServerArchivist.INSTANCE,
+ historyServerArchivist,
null,
UnregisteredMetricGroups.createUnregisteredJobManagerMetricGroup(),
jobGraphWriter,
@@ -637,6 +641,75 @@ public class DispatcherResourceCleanupTest extends TestLogger {
assertThat(deleteAllHABlobsFuture.get(), equalTo(jobId));
}
+ @Test
+ public void testArchivingFinishedJobToHistoryServer() throws Exception {
+
+ final OneShotLatch archivingLatch = new OneShotLatch();
+ final CompletableFuture<Acknowledge> archiveFuture = new CompletableFuture<>();
+
+ historyServerArchivist =
+ executionGraphInfo -> {
+ archivingLatch.trigger();
+ return archiveFuture;
+ };
+ final TestingJobManagerRunnerFactory jobManagerRunnerFactory =
+ startDispatcherAndSubmitJob(0);
+
+ finishJob(jobManagerRunnerFactory.takeCreatedJobManagerRunner());
+
+ archivingLatch.await();
+
+ // The cleanup is not triggered before archiving is done
+ assertThatNoCleanupWasTriggered();
+
+ archiveFuture.complete(Acknowledge.get());
+
+ assertGlobalCleanupTriggered(jobId);
+ }
+
+ @Test
+ public void testNotArchivingSuspendedJobToHistoryServer() throws Exception {
+
+ final AtomicBoolean isArchived = new AtomicBoolean(false);
+
+ historyServerArchivist =
+ executionGraphInfo -> {
+ isArchived.set(true);
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ };
+ final TestingJobManagerRunnerFactory jobManagerRunnerFactory =
+ startDispatcherAndSubmitJob(0);
+
+ suspendJob(jobManagerRunnerFactory.takeCreatedJobManagerRunner());
+
+ assertLocalCleanupTriggered(jobId);
+ dispatcher.getJobTerminationFuture(jobId, Time.hours(1)).join();
+
+ assertFalse(
+ "Archiving should not be triggered for a non-globally terminal job.",
+ isArchived.get());
+ }
+
+ private void assertThatNoCleanupWasTriggered() {
+ assertFalse(cleanupJobFuture.isDone());
+ assertFalse(deleteAllHABlobsFuture.isDone());
+ assertFalse(cleanupJobHADataFuture.isDone());
+ }
+
+ private void assertLocalCleanupTriggered(JobID jobId)
+ throws ExecutionException, InterruptedException {
+ assertEquals(cleanupJobFuture.get(), jobId);
+ assertFalse(deleteAllHABlobsFuture.isDone());
+ assertFalse(cleanupJobHADataFuture.isDone());
+ }
+
+ private void assertGlobalCleanupTriggered(JobID jobId)
+ throws ExecutionException, InterruptedException {
+ assertEquals(cleanupJobFuture.get(), jobId);
+ assertEquals(deleteAllHABlobsFuture.get(), jobId);
+ assertEquals(cleanupJobHADataFuture.get(), jobId);
+ }
+
private static final class TestingBlobServer extends BlobServer {
private final CompletableFuture<JobID> cleanupJobFuture;
[flink] 01/03: [hotfix][tests] Allow retrieval of termination future for running jobs
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mapohl pushed a commit to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 1c492ed97fe2876041804b944c6cb370430b3519
Author: Chesnay Schepler <ch...@apache.org>
AuthorDate: Mon Apr 11 17:33:51 2022 +0200
[hotfix][tests] Allow retrieval of termination future for running jobs
---
.../java/org/apache/flink/runtime/dispatcher/Dispatcher.java | 12 ++++--------
1 file changed, 4 insertions(+), 8 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 121990e2199..c09ba52ca9f 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
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.dispatcher;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobStatus;
import org.apache.flink.api.common.operators.ResourceSpec;
@@ -996,15 +997,10 @@ public abstract class Dispatcher extends PermanentlyFencedRpcEndpoint<Dispatcher
getMainThreadExecutor());
}
+ @VisibleForTesting
CompletableFuture<Void> getJobTerminationFuture(JobID jobId) {
- if (runningJobs.containsKey(jobId)) {
- return FutureUtils.completedExceptionally(
- new DispatcherException(
- String.format("Job with job id %s is still running.", jobId)));
- } else {
- return jobManagerRunnerTerminationFutures.getOrDefault(
- jobId, CompletableFuture.completedFuture(null));
- }
+ return jobManagerRunnerTerminationFutures.getOrDefault(
+ jobId, CompletableFuture.completedFuture(null));
}
private void registerDispatcherMetrics(MetricGroup jobManagerMetricGroup) {
[flink] 02/03: [hotfix][tests] Wait for JobManagerRunner termination
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mapohl pushed a commit to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git
commit e0fb11741a000478852d51fa9f2823208e6717c6
Author: Chesnay Schepler <ch...@apache.org>
AuthorDate: Mon Apr 11 12:00:41 2022 +0200
[hotfix][tests] Wait for JobManagerRunner termination
Need to make sure the job manager runner is complete, because the test runner does not implement required methods to query job details.
---
.../java/org/apache/flink/runtime/dispatcher/DispatcherTest.java | 8 +++++---
1 file changed, 5 insertions(+), 3 deletions(-)
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 68f818c6b8c..3bf3f31e111 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
@@ -449,7 +449,7 @@ public class DispatcherTest extends AbstractDispatcherTest {
testFailure));
// wait till job has failed
- dispatcherGateway.requestJobResult(jobId, TIMEOUT).get();
+ dispatcher.getJobTerminationFuture(jobId, TIMEOUT).get();
// get failure cause
ArchivedExecutionGraph execGraph =
@@ -870,7 +870,7 @@ public class DispatcherTest extends AbstractDispatcherTest {
jobMasterLeaderElectionService.isLeader(UUID.randomUUID());
dispatcherGateway.submitJob(jobGraph, TIMEOUT).get();
- dispatcherGateway.requestJobResult(jobId, TIMEOUT).get();
+ dispatcher.getJobTerminationFuture(jobId, TIMEOUT).get();
assertOnlyContainsSingleJobWithState(
JobStatus.SUSPENDED, dispatcherGateway.requestMultipleJobDetails(TIMEOUT).get());
@@ -916,6 +916,8 @@ public class DispatcherTest extends AbstractDispatcherTest {
// run second job, which completes with FINISHED
dispatcherGateway.submitJob(jobGraph, TIMEOUT).get();
+ dispatcher.getJobTerminationFuture(jobId, TIMEOUT).get();
+
assertOnlyContainsSingleJobWithState(
JobStatus.FINISHED, dispatcherGateway.requestMultipleJobDetails(TIMEOUT).get());
}
@@ -931,7 +933,7 @@ public class DispatcherTest extends AbstractDispatcherTest {
jobMasterLeaderElectionService.isLeader(UUID.randomUUID());
dispatcherGateway.submitJob(jobGraph, TIMEOUT).get();
- dispatcherGateway.requestJobResult(jobId, TIMEOUT).get();
+ dispatcher.getJobTerminationFuture(jobId, TIMEOUT).get();
final MultipleJobsDetails multipleJobsDetails =
dispatcherGateway.requestMultipleJobDetails(TIMEOUT).get();