@Override
    public void suspend(Throwable cause) {
        suspend(cause, null);
    }
@Test
    void testSuspendCanBeCalledWhenExecutionGraphHasReachedGloballyTerminalState()
            throws Exception {
        try (MockStateWithExecutionGraphContext context =
                new MockStateWithExecutionGraphContext()) {
            final StateTrackingMockExecutionGraph testingExecutionGraph =
                    new StateTrackingMockExecutionGraph();
            testingExecutionGraph.transitionToRunning();

            final TestingStateWithExecutionGraph stateWithExecutionGraph =
                    createStateWithExecutionGraph(context, testingExecutionGraph);

            context.setExpectFinished(
                    archivedExecutionGraph ->
                            assertThat(archivedExecutionGraph.getState())
                                    .isEqualTo(JobStatus.FAILED));

            // transition to FAILED
            testingExecutionGraph.failJob(
                    new FlinkException("Transition job to FAILED state"),
                    System.currentTimeMillis());
            testingExecutionGraph.completeTerminationFuture(JobStatus.FAILED);

            assertThat(testingExecutionGraph.getState()).isEqualTo(JobStatus.FAILED);

            // As long as we don't execute StateWithExecutionGraph#onGloballyTerminalState
            // immediately when reaching a globally terminal state or if don't immediately leave
            // this state when reaching a globally terminal state, this test is still valid because
            // the suspend call can happen asynchronously.
            assertThatFuture(stateWithExecutionGraph.getGloballyTerminalStateFuture()).isNotDone();
            stateWithExecutionGraph.suspend(new FlinkException("Test exception"));
        }
    }