@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 testProcessWithAvailableOutput() throws Exception {
        try (final MockEnvironment environment = setupEnvironment(true, true)) {
            final int numberOfProcessCalls = 10;
            final AvailabilityTestInputProcessor inputProcessor =
                    new AvailabilityTestInputProcessor(numberOfProcessCalls);
            final StreamTask task =
                    new MockStreamTaskBuilder(environment)
                            .setStreamInputProcessor(inputProcessor)
                            .build();

            task.invoke();
            assertThat(inputProcessor.currentNumProcessCalls).isEqualTo(numberOfProcessCalls);
        }
    }