CompletableFuture<String> stopWithSavepoint(
            @Nullable final String targetDirectory,
            boolean terminate,
            SavepointFormatType formatType) {
        final ExecutionGraph executionGraph = getExecutionGraph();

        StopWithSavepointTerminationManager.checkSavepointActionPreconditions(
                executionGraph.getCheckpointCoordinator(),
                targetDirectory,
                executionGraph.getJobID(),
                getLogger());

        getLogger().info("Triggering stop-with-savepoint for job {}.", executionGraph.getJobID());

        CheckpointScheduling schedulingProvider = new CheckpointSchedulingProvider(executionGraph);

        schedulingProvider.stopCheckpointScheduler();

        final CompletableFuture<String> savepointFuture =
                Objects.requireNonNull(executionGraph.getCheckpointCoordinator())
                        .triggerSynchronousSavepoint(terminate, targetDirectory, formatType)
                        .thenApply(CompletedCheckpoint::getExternalPointer);
        return context.goToStopWithSavepoint(
                executionGraph,
                getExecutionGraphHandler(),
                getOperatorCoordinatorHandler(),
                schedulingProvider,
                savepointFuture,
                getFailures());
    }
@Test
    void testCheckpointSchedulerIsStoppedOnStopWithSavepoint() throws Exception {
        try (MockExecutingContext ctx = new MockExecutingContext()) {
            CheckpointCoordinator coordinator =
                    new CheckpointCoordinatorTestingUtils.CheckpointCoordinatorBuilder()
                            .build(EXECUTOR_EXTENSION.getExecutor());
            StateTrackingMockExecutionGraph mockedExecutionGraphWithCheckpointCoordinator =
                    new StateTrackingMockExecutionGraph() {
                        @Nullable
                        @Override
                        public CheckpointCoordinator getCheckpointCoordinator() {
                            return coordinator;
                        }
                    };
            Executing exec =
                    new ExecutingStateBuilder()
                            .setExecutionGraph(mockedExecutionGraphWithCheckpointCoordinator)
                            .build(ctx);

            coordinator.startCheckpointScheduler();

            // we assume checkpointing to be enabled
            assertThat(coordinator.isPeriodicCheckpointingStarted()).isTrue();

            ctx.setExpectStopWithSavepoint(assertNonNull());
            exec.stopWithSavepoint("file:///tmp/target", true, SavepointFormatType.CANONICAL);

            assertThat(coordinator.isPeriodicCheckpointingStarted()).isFalse();
        }
    }