@Override
    public final void invoke() throws Exception {
        // Allow invoking method 'invoke' without having to call 'restore' before it.
        if (!isRunning) {
            LOG.debug("Restoring during invoke will be called.");
            restoreInternal();
        }

        // final check to exit early before starting to run
        ensureNotCanceled();

        scheduleBufferDebloater();

        // let the task do its work
        getEnvironment().getMetricGroup().getIOMetricGroup().markTaskStart();
        runMailboxLoop();

        // if this left the run() method cleanly despite the fact that this was canceled,
        // make sure the "clean shutdown" is not attempted
        ensureNotCanceled();

        afterInvoke();
    }
@Test
    void testProcessWithRaceInDataAvailability() throws Exception {
        try (final MockEnvironment environment =
                MockEnvironment.builder()
                        .setTaskStateManager(
                                TestTaskStateManager.builder()
                                        // replicate NPE of FLINK-29397
                                        .setStateChangelogStorage(null)
                                        .build())
                        .build()) {
            environment.addOutputs(
                    Collections.singletonList(new AvailabilityTestResultPartitionWriter(true)));

            final StreamInputProcessor inputProcessor = new RacyTestInputProcessor();
            final StreamTask<?, ?> task =
                    new MockStreamTaskBuilder(environment)
                            .setStreamInputProcessor(inputProcessor)
                            .build();

            task.invoke();
        }
    }