Skip to content

[FLINK-32098][runtime] Dispatcher#submitJob calls Dispatcher#isInGloballyTerminalState up to three times which might be expensive due to IO #22585

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Aug 1, 2023
Merged

[FLINK-32098][runtime] Dispatcher#submitJob calls Dispatcher#isInGloballyTerminalState up to three times which might be expensive due to IO #22585

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
53 changes: 19 additions & 34 deletions flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -511,32 +511,30 @@ private void stopDispatcherServices() throws Exception {

@Override
public CompletableFuture<Acknowledge> submitJob(JobGraph jobGraph, Time timeout) {
log.info(
"Received JobGraph submission '{}' ({}).", jobGraph.getName(), jobGraph.getJobID());
final JobID jobID = jobGraph.getJobID();
log.info("Received JobGraph submission '{}' ({}).", jobGraph.getName(), jobID);

try {
if (isDuplicateJob(jobGraph.getJobID())) {
if (isInGloballyTerminalState(jobGraph.getJobID())) {
log.warn(
"Ignoring JobGraph submission '{}' ({}) because the job already reached a globally-terminal state (i.e. {}) in a previous execution.",
jobGraph.getName(),
jobGraph.getJobID(),
Arrays.stream(JobStatus.values())
.filter(JobStatus::isGloballyTerminalState)
.map(JobStatus::name)
.collect(Collectors.joining(", ")));
}

final DuplicateJobSubmissionException exception =
isInGloballyTerminalState(jobGraph.getJobID())
? DuplicateJobSubmissionException.ofGloballyTerminated(
jobGraph.getJobID())
: DuplicateJobSubmissionException.of(jobGraph.getJobID());
return FutureUtils.completedExceptionally(exception);
if (isInGloballyTerminalState(jobID)) {
log.warn(
"Ignoring JobGraph submission '{}' ({}) because the job already reached a globally-terminal state (i.e. {}) in a previous execution.",
jobGraph.getName(),
jobID,
Arrays.stream(JobStatus.values())
.filter(JobStatus::isGloballyTerminalState)
.map(JobStatus::name)
.collect(Collectors.joining(", ")));
return FutureUtils.completedExceptionally(
DuplicateJobSubmissionException.ofGloballyTerminated(jobID));
} else if (jobManagerRunnerRegistry.isRegistered(jobID)
|| submittedAndWaitingTerminationJobIDs.contains(jobID)) {
// job with the given jobID is not terminated, yet
return FutureUtils.completedExceptionally(
DuplicateJobSubmissionException.of(jobID));
} else if (isPartialResourceConfigured(jobGraph)) {
return FutureUtils.completedExceptionally(
new JobSubmissionException(
jobGraph.getJobID(),
jobID,
"Currently jobs is not supported if parts of the vertices have "
+ "resources configured. The limitation will be removed in future versions."));
} else {
Expand All @@ -563,19 +561,6 @@ public CompletableFuture<Acknowledge> submitFailedJob(
return archiveExecutionGraphToHistoryServer(executionGraphInfo);
}

/**
* Checks whether the given job has already been submitted, executed, or awaiting termination.
*
* @param jobId identifying the submitted job
* @return true if the job has already been submitted (is running) or has been executed
* @throws FlinkException if the job scheduling status cannot be retrieved
*/
private boolean isDuplicateJob(JobID jobId) throws FlinkException {
return isInGloballyTerminalState(jobId)
|| jobManagerRunnerRegistry.isRegistered(jobId)
|| submittedAndWaitingTerminationJobIDs.contains(jobId);
}

/**
* Checks whether the given job has already been executed.
*
Expand Down