class CheckpointStatsDetailsHandler extends AbstractExecutionGraphRequestHandler {

	@Override
	public CompletableFuture<String> handleRequest(AccessExecutionGraph graph, Map<String, String> params) {
		return CompletableFuture.supplyAsync(
			() -> {
				long checkpointId = parseCheckpointId(params);
				if (checkpointId == -1) {
					return "{}";
				}

				CheckpointStatsSnapshot snapshot = graph.getCheckpointStatsSnapshot();
				if (snapshot == null) {
					return "{}";
				}

				AbstractCheckpointStats checkpoint = snapshot.getHistory().getCheckpointById(checkpointId);

				if (checkpoint != null) {
					cache.tryAdd(checkpoint);
				} else {
					checkpoint = cache.tryGet(checkpointId);

					if (checkpoint == null) {
						return "{}";
					}
				}

				try {
					return createCheckpointDetailsJson(checkpoint);
				} catch (IOException e) {
					throw new CompletionException(new FlinkException("Could not create checkpoint details json.", e));
				}
			},
			executor);
	}

	public  CheckpointStatsDetailsHandler(ExecutionGraphCache executionGraphHolder, Executor executor, CheckpointStatsCache cache);

}