class ExecutionGraphCache implements Closeable {

	public CompletableFuture<AccessExecutionGraph> getExecutionGraph(JobID jobId, RestfulGateway restfulGateway) {

		Preconditions.checkState(running, "ExecutionGraphCache is no longer running");

		while (true) {
			final ExecutionGraphEntry oldEntry = cachedExecutionGraphs.get(jobId);

			final long currentTime = System.currentTimeMillis();

			if (oldEntry != null) {
				if (currentTime < oldEntry.getTTL()) {
					final CompletableFuture<AccessExecutionGraph> executionGraphFuture = oldEntry.getExecutionGraphFuture();
					if (executionGraphFuture.isDone() && !executionGraphFuture.isCompletedExceptionally()) {

						// TODO: Remove once we no longer request the actual ExecutionGraph from the JobManager but only the ArchivedExecutionGraph
						try {
							final AccessExecutionGraph executionGraph = executionGraphFuture.get();
							if (executionGraph.getState() != JobStatus.SUSPENDING &&
								executionGraph.getState() != JobStatus.SUSPENDED) {
								return executionGraphFuture;
							}
							// send a new request to get the ExecutionGraph from the new leader
						} catch (InterruptedException | ExecutionException e) {
							throw new RuntimeException("Could not retrieve ExecutionGraph from the orderly completed future. This should never happen.", e);
						}
					} else if (!executionGraphFuture.isDone()) {
						return executionGraphFuture;
					}
					// otherwise it must be completed exceptionally
				}
			}

			final ExecutionGraphEntry newEntry = new ExecutionGraphEntry(currentTime + timeToLive.toMilliseconds());

			final boolean successfulUpdate;

			if (oldEntry == null) {
				successfulUpdate = cachedExecutionGraphs.putIfAbsent(jobId, newEntry) == null;
			} else {
				successfulUpdate = cachedExecutionGraphs.replace(jobId, oldEntry, newEntry);
				// cancel potentially outstanding futures
				oldEntry.getExecutionGraphFuture().cancel(false);
			}

			if (successfulUpdate) {
				final CompletableFuture<? extends AccessExecutionGraph> executionGraphFuture = restfulGateway.requestJob(jobId, timeout);

				executionGraphFuture.whenComplete(
					(AccessExecutionGraph executionGraph, Throwable throwable) -> {
						if (throwable != null) {
							newEntry.getExecutionGraphFuture().completeExceptionally(throwable);

							// remove exceptionally completed entry because it doesn't help
							cachedExecutionGraphs.remove(jobId, newEntry);
						} else {
							newEntry.getExecutionGraphFuture().complete(executionGraph);

							// TODO: Remove once we no longer request the actual ExecutionGraph from the JobManager but only the ArchivedExecutionGraph
							if (executionGraph.getState() == JobStatus.SUSPENDING ||
								executionGraph.getState() == JobStatus.SUSPENDED) {
								// remove the entry in case of suspension --> triggers new request when accessed next time
								cachedExecutionGraphs.remove(jobId, newEntry);
							}
						}
					});

				if (!running) {
					// delete newly added entry in case of a concurrent stopping operation
					cachedExecutionGraphs.remove(jobId, newEntry);
				}

				return newEntry.getExecutionGraphFuture();
			}
		}
	}

	public  ExecutionGraphCache(
			Time timeout,
			Time timeToLive);

}