@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 testFlinkJobNotFoundException() throws Exception {
        GatewayRetriever<RestfulGateway> leaderRetriever =
                () -> CompletableFuture.completedFuture(null);
        CheckpointStatsSnapshot checkpointStatsSnapshot1 = getTestCheckpointStatsSnapshot();
        CompletableFuture<CheckpointStatsSnapshot> failedFuture = new CompletableFuture<>();
        failedFuture.completeExceptionally(new FlinkJobNotFoundException(JOB_ID));

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

            assertThat(checkpointStatsHandler.handleRequest(request, restfulGateway))
                    .isCompletedExceptionally()
                    .failsWithin(Duration.ofSeconds(1))
                    .withThrowableOfType(ExecutionException.class)
                    .withStackTraceContaining("Job %s not found", JOB_ID);
        }
    }