@Override
    protected CompletableFuture<Void> getCompletionFuture() {
        return sourceThread.getCompletionFuture();
    }
@Test
    void testClosedOnRestoreSourceSkipExecution() throws Exception {
        LifeCycleMonitorSource testSource = new LifeCycleMonitorSource();
        List<Object> output = new ArrayList<>();
        try (StreamTaskMailboxTestHarness<String> harness =
                new StreamTaskMailboxTestHarnessBuilder<>(SourceStreamTask::new, STRING_TYPE_INFO)
                        .setTaskStateSnapshot(1, TaskStateSnapshot.FINISHED_ON_RESTORE)
                        .addAdditionalOutput(
                                new RecordOrEventCollectingResultPartitionWriter<StreamElement>(
                                        output,
                                        new StreamElementSerializer<>(IntSerializer.INSTANCE)) {
                                    @Override
                                    public void notifyEndOfData(StopMode mode) throws IOException {
                                        broadcastEvent(new EndOfData(mode), false);
                                    }
                                })
                        .setupOperatorChain(new StreamSource<>(testSource))
                        .chain(new TestFinishedOnRestoreStreamOperator(), StringSerializer.INSTANCE)
                        .finish()
                        .build()) {
            harness.getStreamTask().invoke();
            harness.processAll();
            harness.streamTask.getCompletionFuture().get();

            assertThat(output)
                    .containsExactly(Watermark.MAX_WATERMARK, new EndOfData(StopMode.DRAIN));

            LifeCycleMonitorSource source =
                    (LifeCycleMonitorSource)
                            ((StreamSource<?, ?>) harness.getStreamTask().getMainOperator())
                                    .getUserFunction();
            source.getLifeCycleMonitor()
                    .assertCallTimes(
                            0,
                            LifeCyclePhase.OPEN,
                            LifeCyclePhase.PROCESS_ELEMENT,
                            LifeCyclePhase.CLOSE);
        }
    }