class Dispatcher extends PermanentlyFencedRpcEndpoint<DispatcherId> implements DispatcherGateway {

	@Override
	public CompletableFuture<ArchivedExecutionGraph> requestJob(JobID jobId, Time timeout) {
		Function<Throwable, ArchivedExecutionGraph> checkExecutionGraphStoreOnException = throwable ->  {
			// check whether it is a completed job
			final ArchivedExecutionGraph archivedExecutionGraph = archivedExecutionGraphStore.get(jobId);
			if (archivedExecutionGraph == null) {
				throw new CompletionException(ExceptionUtils.stripCompletionException(throwable));
			} else {
				return archivedExecutionGraph;
			}
		};
		Optional<DispatcherJob> maybeJob = getDispatcherJob(jobId);
		return maybeJob.map(job -> job.requestJob(timeout))
			.orElse(FutureUtils.completedExceptionally(new FlinkJobNotFoundException(jobId)))
			.exceptionally(checkExecutionGraphStoreOnException);
	}

	public  Dispatcher(
			RpcService rpcService,
			DispatcherId fencingToken,
			DispatcherBootstrap dispatcherBootstrap,
			DispatcherServices dispatcherServices);

}