@Override
    public final void cancel() throws Exception {
        isRunning = false;
        canceled = true;

        FlinkSecurityManager.monitorUserSystemExitForCurrentThread();
        // the "cancel task" call must come first, but the cancelables must be
        // closed no matter what
        try {
            cancelTask();
        } finally {
            FlinkSecurityManager.unmonitorUserSystemExitForCurrentThread();
            getCompletionFuture()
                    .whenComplete(
                            (unusedResult, unusedError) -> {
                                // WARN: the method is called from the task thread but the callback
                                // can be invoked from a different thread
                                mailboxProcessor.allActionsCompleted();
                                try {
                                    subtaskCheckpointCoordinator.cancel();
                                    cancelables.close();
                                } catch (IOException e) {
                                    throw new CompletionException(e);
                                }
                            });
        }
    }
@Test
    void testRestorePerformedFromInvoke() throws Exception {
        // given: the operator with empty snapshot result (all state handles are null)
        OneInputStreamOperator<String, String> statelessOperator =
                streamOperatorWithSnapshot(new OperatorSnapshotFutures());
        DummyEnvironment dummyEnvironment = new DummyEnvironment();

        // when: Launch the task.
        RunningTask<MockStreamTask> task =
                runTask(
                        () ->
                                createMockStreamTask(
                                        dummyEnvironment, operatorChain(statelessOperator)));

        waitTaskIsRunning(task.streamTask, task.invocationFuture);

        task.streamTask.cancel();

        // then: 'restore' was called even without explicit 'restore' invocation.
        assertThat(task.streamTask.restoreInvocationCount).isOne();
    }