@Override
    protected CompletableFuture<R> handleRequest(
            @Nonnull HandlerRequest<EmptyRequestBody> request, @Nonnull RestfulGateway gateway)
            throws RestHandlerException {
        JobID jobId = request.getPathParameter(JobIDPathParameter.class);

        try {
            return checkpointStatsSnapshotCache
                    .get(jobId, () -> gateway.requestCheckpointStats(jobId, timeout))
                    .thenApplyAsync(
                            checkpointStatsSnapshot -> {
                                try {
                                    return handleCheckpointStatsRequest(
                                            request, checkpointStatsSnapshot);
                                } catch (RestHandlerException e) {
                                    throw new CompletionException(e);
                                }
                            },
                            executor)
                    .exceptionally(
                            throwable -> {
                                throwable = ExceptionUtils.stripCompletionException(throwable);
                                if (throwable instanceof FlinkJobNotFoundException) {
                                    throw new CompletionException(
                                            new NotFoundException(
                                                    String.format("Job %s not found", jobId),
                                                    throwable));
                                } else {
                                    throw new CompletionException(throwable);
                                }
                            });
        } catch (ExecutionException e) {
            CompletableFuture<R> future = new CompletableFuture<>();
            future.completeExceptionally(e);
            return future;
        }
    }
@Test
    void testRestExceptionPassedThrough() throws Exception {
        GatewayRetriever<RestfulGateway> leaderRetriever =
                () -> CompletableFuture.completedFuture(null);
        CheckpointStatsSnapshot checkpointStatsSnapshot1 = getTestCheckpointStatsSnapshot();
        RestHandlerException restHandlerException =
                new RestHandlerException(
                        "some exception thrown", HttpResponseStatus.INTERNAL_SERVER_ERROR);

        try (ThrowingCheckpointStatsHandler checkpointStatsHandler =
                new ThrowingCheckpointStatsHandler(
                        leaderRetriever,
                        TIMEOUT,
                        Collections.emptyMap(),
                        CheckpointingStatisticsHeaders.getInstance(),
                        CacheBuilder.newBuilder().build(),
                        Executors.directExecutor(),
                        restHandlerException)) {
            RestfulGateway restfulGateway =
                    new TestingRestfulGateway.Builder()
                            .setRequestCheckpointStatsSnapshotFunction(
                                    jobID ->
                                            CompletableFuture.completedFuture(
                                                    checkpointStatsSnapshot1))
                            .build();
            HandlerRequest<EmptyRequestBody> request =
                    HandlerRequest.resolveParametersAndCreate(
                            EmptyRequestBody.getInstance(),
                            new JobMessageParameters(),
                            Collections.singletonMap(JobIDPathParameter.KEY, JOB_ID.toString()),
                            Collections.emptyMap(),
                            Collections.emptyList());

            assertThatExceptionOfType(ExecutionException.class)
                    .isThrownBy(
                            () ->
                                    checkpointStatsHandler
                                            .handleRequest(request, restfulGateway)
                                            .get())
                    .withCause(restHandlerException);
        }
    }