eemario commented on code in PR #27741:
URL: https://github.com/apache/flink/pull/27741#discussion_r2973212818
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/runner/SessionDispatcherLeaderProcess.java:
##########
@@ -251,28 +446,40 @@ private Optional<CompletableFuture<Void>>
submitAddedJobIfRunning(ExecutionPlan
private CompletableFuture<Void> submitAddedJob(ExecutionPlan
executionPlan) {
final DispatcherGateway dispatcherGateway =
getDispatcherGatewayInternal();
+ final JobID jobId = executionPlan.getJobID();
+ final Duration timeout =
+
executionPlan.getJobConfiguration().get(RpcOptions.ASK_TIMEOUT_DURATION);
+ // Skip job submission if its associated application exists, as the
application will handle
+ // the job recovery
+ ApplicationID applicationId =
executionPlan.getApplicationId().orElse(null);
return dispatcherGateway
- .submitJob(executionPlan, RpcUtils.INF_TIMEOUT)
- .thenApply(FunctionUtils.nullFn())
- .exceptionally(this::filterOutDuplicateJobSubmissionException);
- }
-
- private Void filterOutDuplicateJobSubmissionException(Throwable throwable)
{
- final Throwable strippedException =
ExceptionUtils.stripCompletionException(throwable);
- if (strippedException instanceof DuplicateJobSubmissionException) {
- final DuplicateJobSubmissionException
duplicateJobSubmissionException =
- (DuplicateJobSubmissionException) strippedException;
-
- log.debug(
- "Ignore recovered job {} because the job is currently
being executed.",
- duplicateJobSubmissionException.getJobID(),
- duplicateJobSubmissionException);
-
- return null;
- } else {
- throw new CompletionException(throwable);
- }
+ .requestApplication(applicationId, timeout)
Review Comment:
Updated: throws early if applicationId is null, as this indicates an
abnormal state that should trigger a fatal error.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
