@Override
    public void cancel() {
        operationFailureCause = new FlinkException("The job was cancelled.");
        context.goToCanceling(
                getExecutionGraph(),
                getExecutionGraphHandler(),
                getOperatorCoordinatorHandler(),
                getFailures());
    }
@Test
    void testTransitionToCancellingOnCancel() throws Exception {
        try (MockStopWithSavepointContext ctx = new MockStopWithSavepointContext()) {
            StopWithSavepoint sws = createStopWithSavepoint(ctx);
            ctx.setStopWithSavepoint(sws);
            ctx.setExpectCancelling(assertNonNull());

            sws.cancel();
        }
    }