@Override
    public CompletableFuture<Boolean> triggerCheckpointAsync(
            CheckpointMetaData checkpointMetaData, CheckpointOptions checkpointOptions) {
        checkForcedFullSnapshotSupport(checkpointOptions);

        CompletableFuture<Boolean> result = new CompletableFuture<>();
        mainMailboxExecutor.execute(
                () -> {
                    try {
                        boolean noUnfinishedInputGates =
                                Arrays.stream(getEnvironment().getAllInputGates())
                                        .allMatch(InputGate::isFinished);

                        if (noUnfinishedInputGates) {
                            result.complete(
                                    triggerCheckpointAsyncInMailbox(
                                            checkpointMetaData, checkpointOptions));
                        } else {
                            result.complete(
                                    triggerUnfinishedChannelsCheckpoint(
                                            checkpointMetaData, checkpointOptions));
                        }
                    } catch (Exception ex) {
                        // Report the failure both via the Future result but also to the mailbox
                        result.completeExceptionally(ex);
                        throw ex;
                    }
                },
                "checkpoint %s with %s",
                checkpointMetaData,
                checkpointOptions);
        return result;
    }
@Test
    void testUncaughtExceptionInAsynchronousCheckpointingOperation() throws Exception {
        final RuntimeException failingCause = new RuntimeException("Test exception");
        FailingDummyEnvironment failingDummyEnvironment = new FailingDummyEnvironment(failingCause);

        // mock the returned snapshots
        OperatorSnapshotFutures operatorSnapshotResult =
                new OperatorSnapshotFutures(
                        ExceptionallyDoneFuture.of(failingCause),
                        DoneFuture.of(SnapshotResult.empty()),
                        DoneFuture.of(SnapshotResult.empty()),
                        DoneFuture.of(SnapshotResult.empty()),
                        DoneFuture.of(SnapshotResult.empty()),
                        DoneFuture.of(SnapshotResult.empty()));

        final TestingUncaughtExceptionHandler uncaughtExceptionHandler =
                new TestingUncaughtExceptionHandler();

        RunningTask<MockStreamTask> task =
                runTask(
                        () ->
                                new MockStreamTask(
                                        failingDummyEnvironment,
                                        operatorChain(
                                                streamOperatorWithSnapshot(operatorSnapshotResult)),
                                        uncaughtExceptionHandler));
        MockStreamTask streamTask = task.streamTask;

        waitTaskIsRunning(streamTask, task.invocationFuture);

        streamTask.triggerCheckpointAsync(
                new CheckpointMetaData(42L, 1L),
                CheckpointOptions.forCheckpointWithDefaultLocation());

        final Throwable uncaughtException = uncaughtExceptionHandler.waitForUncaughtException();
        assertThat(uncaughtException).isSameAs(failingCause);

        streamTask.finishInput();
        task.waitForTaskCompletion(false);
    }