protected void afterInvoke() throws Exception {
        LOG.debug("Finished task {}", getName());
        getCompletionFuture().exceptionally(unused -> null).join();

        Set<CompletableFuture<Void>> terminationConditions = new HashSet<>();
        // If checkpoints are enabled, waits for all the records get processed by the downstream
        // tasks. During this process, this task could coordinate with its downstream tasks to
        // continue perform checkpoints.
        if (endOfDataReceived && areCheckpointsWithFinishedTasksEnabled()) {
            LOG.debug("Waiting for all the records processed by the downstream tasks.");

            for (ResultPartitionWriter partitionWriter : getEnvironment().getAllWriters()) {
                terminationConditions.add(partitionWriter.getAllDataProcessedFuture());
            }

            terminationConditions.add(finalCheckpointCompleted);
        }

        if (syncSavepoint != null) {
            terminationConditions.add(finalCheckpointCompleted);
        }

        FutureUtils.waitForAll(terminationConditions)
                .thenRun(mailboxProcessor::allActionsCompleted);

        // Resumes the mailbox processor. The mailbox processor would be completed
        // after all records are processed by the downstream tasks.
        mailboxProcessor.runMailboxLoop();

        // make sure no further checkpoint and notification actions happen.
        // at the same time, this makes sure that during any "regular" exit where still
        actionExecutor.runThrowing(
                () -> {
                    // make sure no new timers can come
                    timerService.quiesce().get();
                    systemTimerService.quiesce().get();

                    // let mailbox execution reject all new letters from this point
                    mailboxProcessor.prepareClose();
                });

        // processes the remaining mails; no new mails can be enqueued
        mailboxProcessor.drain();

        // Set isRunning to false after all the mails are drained so that
        // the queued checkpoint requirements could be triggered normally.
        actionExecutor.runThrowing(
                () -> {
                    // only set the StreamTask to not running after all operators have been
                    // finished!
                    // See FLINK-7430
                    isRunning = false;
                });

        LOG.debug("Finished operators for task {}", getName());

        // make sure all buffered data is flushed
        operatorChain.flushOutputs();

        if (areCheckpointsWithFinishedTasksEnabled()) {
            // No new checkpoints could be triggered since mailbox has been drained.
            subtaskCheckpointCoordinator.waitForPendingCheckpoints();
            LOG.debug("All pending checkpoints are finished");
        }

        disableInterruptOnCancel();

        // make an attempt to dispose the operators such that failures in the dispose call
        // still let the computation fail
        closeAllOperators();
    }
@Test
    void testQuiesceOfMailboxRightBeforeSubmittingActionViaTimerService() throws Exception {
        // given: the stream task with configured handle async exception.
        AtomicBoolean submitThroughputFail = new AtomicBoolean();
        MockEnvironment mockEnvironment = new MockEnvironmentBuilder().build();

        final UnAvailableTestInputProcessor inputProcessor = new UnAvailableTestInputProcessor();
        RunningTask<StreamTask<?, ?>> task =
                runTask(
                        () ->
                                new MockStreamTaskBuilder(mockEnvironment)
                                        .setHandleAsyncException(
                                                (str, t) -> submitThroughputFail.set(true))
                                        .setStreamInputProcessor(inputProcessor)
                                        .build());

        waitTaskIsRunning(task.streamTask, task.invocationFuture);

        TimerService timerService = task.streamTask.systemTimerService;
        MailboxExecutor mainMailboxExecutor =
                task.streamTask.mailboxProcessor.getMainMailboxExecutor();

        CountDownLatch stoppingMailboxLatch = new CountDownLatch(1);
        timerService.registerTimer(
                timerService.getCurrentProcessingTime(),
                (time) -> {
                    stoppingMailboxLatch.await();
                    // The time to the start 'afterInvoke' inside of mailbox.
                    // 'afterInvoke' won't finish until this execution won't finish so it is
                    // impossible to wait on latch or something else.
                    Thread.sleep(5);
                    mainMailboxExecutor.execute(() -> {}, "test");
                });

        // when: Calling the quiesce for mailbox and finishing the timer service.
        mainMailboxExecutor
                .submit(
                        () -> {
                            stoppingMailboxLatch.countDown();
                            task.streamTask.afterInvoke();
                        },
                        "test")
                .get();

        // then: the exception handle wasn't invoked because the such situation is expected.
        assertThat(submitThroughputFail).isFalse();

        // Correctly shutdown the stream task to avoid hanging.
        inputProcessor.availabilityProvider.getUnavailableToResetAvailable().complete(null);
    }