@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 testForceFullSnapshotOnIncompatibleStateBackend() throws Exception {
        try (StreamTaskMailboxTestHarness<Integer> harness =
                new StreamTaskMailboxTestHarnessBuilder<>(
                                OneInputStreamTask::new, BasicTypeInfo.INT_TYPE_INFO)
                        .modifyStreamConfig(
                                config -> config.setStateBackend(new OnlyIncrementalStateBackend()))
                        .addInput(BasicTypeInfo.INT_TYPE_INFO)
                        .setupOutputForSingletonOperatorChain(new StreamMap<>(value -> null))
                        .build()) {
            assertThatThrownBy(
                            () ->
                                    harness.streamTask.triggerCheckpointAsync(
                                            new CheckpointMetaData(42L, 1L),
                                            CheckpointOptions.forConfig(
                                                    CheckpointType.FULL_CHECKPOINT,
                                                    getDefault(),
                                                    true,
                                                    false,
                                                    0L)))
                    .isInstanceOf(IllegalStateException.class)
                    .hasMessage(
                            "Configured state backend (OnlyIncrementalStateBackend) does not"
                                    + " support enforcing a full snapshot. If you are restoring in"
                                    + " NO_CLAIM mode, please consider choosing CLAIM mode.");
        }
    }