@Override
    public CompletableFuture<Boolean> triggerCheckpointAsync(
            CheckpointMetaData checkpointMetaData, CheckpointOptions checkpointOptions) {
        if (!isExternallyInducedSource()) {
            return triggerCheckpointNowAsync(checkpointMetaData, checkpointOptions);
        }
        CompletableFuture<Boolean> triggerFuture = new CompletableFuture<>();
        // immediately move RPC to mailbox so we don't need to synchronize fields
        mainMailboxExecutor.execute(
                () ->
                        triggerCheckpointOnExternallyInducedSource(
                                checkpointMetaData, checkpointOptions, triggerFuture),
                "SourceOperatorStreamTask#triggerCheckpointAsync(%s, %s)",
                checkpointMetaData,
                checkpointOptions);
        return triggerFuture;
    }
@Test
    void testTriggeringStopWithSavepointWithDrain() throws Exception {
        SourceOperatorFactory<Integer> sourceOperatorFactory =
                new SourceOperatorFactory<>(
                        new MockSource(Boundedness.CONTINUOUS_UNBOUNDED, 2),
                        WatermarkStrategy.noWatermarks());

        CompletableFuture<Boolean> checkpointCompleted = new CompletableFuture<>();
        CheckpointResponder checkpointResponder =
                new TestCheckpointResponder() {
                    @Override
                    public void acknowledgeCheckpoint(
                            JobID jobID,
                            ExecutionAttemptID executionAttemptID,
                            long checkpointId,
                            CheckpointMetrics checkpointMetrics,
                            TaskStateSnapshot subtaskState) {
                        super.acknowledgeCheckpoint(
                                jobID,
                                executionAttemptID,
                                checkpointId,
                                checkpointMetrics,
                                subtaskState);
                        checkpointCompleted.complete(null);
                    }
                };

        try (StreamTaskMailboxTestHarness<Integer> testHarness =
                new StreamTaskMailboxTestHarnessBuilder<>(
                                SourceOperatorStreamTask::new, BasicTypeInfo.INT_TYPE_INFO)
                        .setupOutputForSingletonOperatorChain(sourceOperatorFactory)
                        .setCheckpointResponder(checkpointResponder)
                        .build()) {

            CompletableFuture<Boolean> triggerResult =
                    testHarness.streamTask.triggerCheckpointAsync(
                            new CheckpointMetaData(2, 2),
                            CheckpointOptions.alignedNoTimeout(
                                    SavepointType.terminate(SavepointFormatType.CANONICAL),
                                    SAVEPOINT_LOCATION));
            checkpointCompleted.whenComplete(
                    (ignored, exception) ->
                            testHarness.streamTask.notifyCheckpointCompleteAsync(2));
            testHarness.waitForTaskCompletion();
            testHarness.finishProcessing();

            assertThat(triggerResult.isDone()).isTrue();
            assertThat(triggerResult.get()).isTrue();
            assertThat(checkpointCompleted.isDone()).isTrue();
        }
    }