@Override
    public CompletableFuture<Boolean> triggerCheckpointAsync(
            CheckpointMetaData metadata, CheckpointOptions options) {

        if (operatorChain.getSourceTaskInputs().size() == 0) {
            return super.triggerCheckpointAsync(metadata, options);
        }

        // If there are chained sources, we would always only trigger the
        // chained sources for checkpoint. This means that for the checkpoints
        // during the upstream task finished and this task receives the
        // EndOfPartitionEvent, we would not complement barriers for the
        // unfinished network inputs, and the checkpoint would be triggered
        // after received all the EndOfPartitionEvent.
        if (isSynchronous(options.getCheckpointType())) {
            return triggerStopWithSavepointAsync(metadata, options);
        } else {
            return triggerSourcesCheckpointAsync(metadata, options);
        }
    }
@TestTemplate
    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<String> testHarness =
                new StreamTaskMailboxTestHarnessBuilder<>(
                                MultipleInputStreamTask::new, BasicTypeInfo.STRING_TYPE_INFO)
                        .setCollectNetworkEvents()
                        .modifyStreamConfig(config -> config.setCheckpointingEnabled(true))
                        .modifyExecutionConfig(applyObjectReuse(objectReuse))
                        .addInput(BasicTypeInfo.INT_TYPE_INFO)
                        .addInput(BasicTypeInfo.INT_TYPE_INFO)
                        .addInput(BasicTypeInfo.INT_TYPE_INFO)
                        .setTaskStateSnapshot(1, TaskStateSnapshot.FINISHED_ON_RESTORE)
                        .setupOperatorChain(new LifeCycleMonitorMultipleInputOperatorFactory())
                        .finishForSingletonOperatorChain(StringSerializer.INSTANCE)
                        .setCheckpointResponder(checkpointResponder)
                        .build()) {
            CompletableFuture<Boolean> triggerResult =
                    testHarness.streamTask.triggerCheckpointAsync(
                            new CheckpointMetaData(2, 2),
                            CheckpointOptions.alignedNoTimeout(
                                    SavepointType.terminate(SavepointFormatType.CANONICAL),
                                    CheckpointStorageLocationReference.getDefault()));
            checkpointCompleted.whenComplete(
                    (ignored, exception) ->
                            testHarness.streamTask.notifyCheckpointCompleteAsync(2));
            testHarness.waitForTaskCompletion();
            testHarness.finishProcessing();

            assertThat(triggerResult).isCompletedWithValue(true);
            assertThat(checkpointCompleted).isDone();
        }
    }