@Override
    public CompletableFuture<String> triggerSavepoint(
            @Nullable String targetDirectory, boolean cancelJob, SavepointFormatType formatType) {
        return state.tryCall(
                        StateWithExecutionGraph.class,
                        stateWithExecutionGraph -> {
                            if (isAnyOutputBlocking(stateWithExecutionGraph.getExecutionGraph())) {
                                return FutureUtils.<String>completedExceptionally(
                                        new CheckpointException(
                                                CheckpointFailureReason.BLOCKING_OUTPUT_EXIST));
                            }
                            return stateWithExecutionGraph.triggerSavepoint(
                                    targetDirectory, cancelJob, formatType);
                        },
                        "triggerSavepoint")
                .orElse(
                        FutureUtils.completedExceptionally(
                                new CheckpointException(
                                        "The Flink job is currently not executing.",
                                        CheckpointFailureReason.TRIGGER_CHECKPOINT_FAILURE)));
    }
@Test
    void testTriggerSavepointFailsInIllegalState() throws Exception {
        final AdaptiveScheduler scheduler =
                new AdaptiveSchedulerBuilder(
                                createJobGraph(),
                                mainThreadExecutor,
                                EXECUTOR_RESOURCE.getExecutor())
                        .build();

        assertThatFuture(
                        scheduler.triggerSavepoint(
                                "some directory", false, SavepointFormatType.CANONICAL))
                .eventuallyFailsWith(ExecutionException.class)
                .withCauseInstanceOf(CheckpointException.class);
    }