CompletableFuture<String> getOperationFuture() {
        return operationFuture;
    }
@Test
    void testJobFailedAndSavepointOperationFails() throws Exception {
        try (MockStopWithSavepointContext ctx = new MockStopWithSavepointContext()) {
            StateTrackingMockExecutionGraph mockExecutionGraph =
                    new StateTrackingMockExecutionGraph();
            CompletableFuture<String> savepointFuture = new CompletableFuture<>();
            StopWithSavepoint sws =
                    createStopWithSavepoint(ctx, mockExecutionGraph, savepointFuture);
            ctx.setStopWithSavepoint(sws);
            ctx.setHowToHandleFailure(FailureResult::canNotRestart);

            ctx.setExpectFailing(
                    failingArguments -> {
                        assertThat(failingArguments.getExecutionGraph().getState())
                                .isEqualTo(JobStatus.FAILED);
                        assertThat(failingArguments.getFailureCause())
                                .satisfies(FlinkAssertions.anyCauseMatches(FlinkException.class));
                    });

            // fail job:
            mockExecutionGraph.completeTerminationFuture(JobStatus.FAILED);
            savepointFuture.completeExceptionally(new RuntimeException());
            ctx.triggerExecutors();

            assertThat(sws.getOperationFuture()).isCompletedExceptionally();
        }
    }