class Dispatcher extends FencedRpcEndpoint<DispatcherId> implements
	DispatcherGateway, LeaderContender, SubmittedJobGraphStore.SubmittedJobGraphListener {

	@Override
	public void onAddedJobGraph(final JobID jobId) {
		runAsync(
			() -> {
				if (!jobManagerRunnerFutures.containsKey(jobId)) {
					// IMPORTANT: onAddedJobGraph can generate false positives and, thus, we must expect that
					// the specified job is already removed from the SubmittedJobGraphStore. In this case,
					// SubmittedJobGraphStore.recoverJob returns null.
					final CompletableFuture<Optional<JobGraph>> recoveredJob = recoveryOperation.thenApplyAsync(
						FunctionUtils.uncheckedFunction(ignored -> Optional.ofNullable(recoverJob(jobId))),
						getRpcService().getExecutor());

					final DispatcherId dispatcherId = getFencingToken();
					final CompletableFuture<Void> submissionFuture = recoveredJob.thenComposeAsync(
						(Optional<JobGraph> jobGraphOptional) -> jobGraphOptional.map(
							FunctionUtils.uncheckedFunction(jobGraph -> tryRunRecoveredJobGraph(jobGraph, dispatcherId).thenAcceptAsync(
								FunctionUtils.uncheckedConsumer((Boolean isRecoveredJobRunning) -> {
										if (!isRecoveredJobRunning) {
											submittedJobGraphStore.releaseJobGraph(jobId);
										}
									}),
									getRpcService().getExecutor())))
							.orElse(CompletableFuture.completedFuture(null)),
						getUnfencedMainThreadExecutor());

					submissionFuture.whenComplete(
						(Void ignored, Throwable throwable) -> {
							if (throwable != null) {
								onFatalError(
									new DispatcherException(
										String.format("Could not start the added job %s", jobId),
										ExceptionUtils.stripCompletionException(throwable)));
							}
						});

					recoveryOperation = submissionFuture;
				}
			});
	}

	public  Dispatcher(
			RpcService rpcService,
			String endpointId,
			Configuration configuration,
			HighAvailabilityServices highAvailabilityServices,
			SubmittedJobGraphStore submittedJobGraphStore,
			GatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever,
			BlobServer blobServer,
			HeartbeatServices heartbeatServices,
			JobManagerMetricGroup jobManagerMetricGroup,
			@Nullable String metricServiceQueryPath,
			ArchivedExecutionGraphStore archivedExecutionGraphStore,
			JobManagerRunnerFactory jobManagerRunnerFactory,
			FatalErrorHandler fatalErrorHandler,
			HistoryServerArchivist historyServerArchivist);

}