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
commit 8d006224d209c981c563ef6625bc6f617d40684a Author: Chesnay Schepler <[email protected]> AuthorDate: Fri Apr 8 12:28:13 2022 +0200 [FLINK-27140][coordination] Write job result in ioExecutor --- .../flink/runtime/dispatcher/Dispatcher.java | 89 +++++++++++++--------- .../flink/runtime/dispatcher/MiniDispatcher.java | 43 ++++++----- 2 files changed, 79 insertions(+), 53 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 6d8897e3a65..e6c9e74b443 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 @@ -612,10 +612,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.thenCompose( @@ -644,13 +646,14 @@ public abstract class Dispatcher extends PermanentlyFencedRpcEndpoint<Dispatcher return null; } - private CleanupJobState handleJobManagerRunnerResult( + private CompletableFuture<CleanupJobState> handleJobManagerRunnerResult( JobManagerRunnerResult jobManagerRunnerResult, ExecutionType executionType) { if (jobManagerRunnerResult.isInitializationFailure() && executionType == ExecutionType.RECOVERY) { - return jobManagerRunnerFailed( - jobManagerRunnerResult.getExecutionGraphInfo().getJobId(), - jobManagerRunnerResult.getInitializationFailure()); + return CompletableFuture.completedFuture( + jobManagerRunnerFailed( + jobManagerRunnerResult.getExecutionGraphInfo().getJobId(), + jobManagerRunnerResult.getInitializationFailure())); } return jobReachedTerminalState(jobManagerRunnerResult.getExecutionGraphInfo()); } @@ -985,7 +988,7 @@ public abstract class Dispatcher extends PermanentlyFencedRpcEndpoint<Dispatcher case GLOBAL: return globalResourceCleaner .cleanupAsync(jobId) - .thenRun(() -> markJobAsClean(jobId)); + .thenRunAsync(() -> markJobAsClean(jobId), ioExecutor); default: throw new IllegalStateException("Invalid cleanup state: " + cleanupJobState); } @@ -1030,7 +1033,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(); @@ -1062,35 +1066,50 @@ public abstract class Dispatcher extends PermanentlyFencedRpcEndpoint<Dispatcher archiveExecutionGraph(executionGraphInfo); - if (terminalJobStatus.isGloballyTerminalState()) { - final JobID jobId = executionGraphInfo.getJobId(); - try { - if (jobResultStore.hasCleanJobResultEntry(jobId)) { - log.warn( - "Job {} is already marked as clean but clean up was triggered again.", - jobId); - } else if (!jobResultStore.hasDirtyJobResultEntry(jobId)) { - jobResultStore.createDirtyResult( - new JobResultEntry( - JobResult.createFrom( - executionGraphInfo.getArchivedExecutionGraph()))); - log.info( - "Job {} has been registered for cleanup in the JobResultStore after reaching a terminal state.", - jobId); - } - } catch (IOException e) { - fatalErrorHandler.onFatalError( - new FlinkException( - String.format( - "The job %s couldn't be marked as pre-cleanup finished in JobResultStore.", - jobId), - e)); - } + if (!terminalJobStatus.isGloballyTerminalState()) { + return CompletableFuture.completedFuture(CleanupJobState.LOCAL); } - return terminalJobStatus.isGloballyTerminalState() - ? CleanupJobState.GLOBAL - : CleanupJobState.LOCAL; + final CompletableFuture<Void> writeFuture = new CompletableFuture<>(); + final JobID jobId = executionGraphInfo.getJobId(); + + ioExecutor.execute( + () -> { + try { + if (jobResultStore.hasCleanJobResultEntry(jobId)) { + log.warn( + "Job {} is already marked as clean but clean up was triggered again.", + jobId); + } else if (!jobResultStore.hasDirtyJobResultEntry(jobId)) { + jobResultStore.createDirtyResult( + new JobResultEntry( + JobResult.createFrom( + executionGraphInfo + .getArchivedExecutionGraph()))); + log.info( + "Job {} has been registered for cleanup in the JobResultStore after reaching a terminal state.", + jobId); + } + } catch (IOException e) { + writeFuture.completeExceptionally(e); + return; + } + writeFuture.complete(null); + }); + + return writeFuture.handleAsync( + (ignored, error) -> { + if (error != null) { + fatalErrorHandler.onFatalError( + new FlinkException( + String.format( + "The job %s couldn't be marked as pre-cleanup finished in JobResultStore.", + executionGraphInfo.getJobId()), + error)); + } + return CleanupJobState.GLOBAL; + }, + getMainThreadExecutor()); } private void archiveExecutionGraph(ExecutionGraphInfo executionGraphInfo) { 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 76afe73bf9d..85a08332e66 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 @@ -129,26 +129,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; + }); } }
