@Override
    public void maybeInterruptOnCancel(
            Thread toInterrupt, @Nullable String taskName, @Nullable Long timeout) {
        super.maybeInterruptOnCancel(toInterrupt, taskName, timeout);
        interruptSourceThread();
    }
@Test
    void testWaitsForSourceThreadOnCancel() throws Exception {
        StreamTaskTestHarness<String> harness =
                new StreamTaskTestHarness<>(SourceStreamTask::new, STRING_TYPE_INFO);

        harness.setupOutputForSingletonOperatorChain();
        harness.getStreamConfig().setStreamOperator(new StreamSource<>(new NonStoppingSource()));

        harness.invoke();
        NonStoppingSource.waitForStart();

        // SourceStreamTask should be still waiting for NonStoppingSource after cancellation
        harness.getTask().cancel();
        harness.waitForTaskCompletion(50, true); // allow task to exit prematurely
        assertThat(harness.taskThread.isAlive()).isTrue();

        // SourceStreamTask should be still waiting for NonStoppingSource after interruptions
        for (int i = 0; i < 10; i++) {
            harness.getTask().maybeInterruptOnCancel(harness.getTaskThread(), null, null);
            harness.waitForTaskCompletion(50, true); // allow task to exit prematurely
            assertThat(harness.taskThread.isAlive()).isTrue();
        }

        // It should only exit once NonStoppingSource allows for it
        NonStoppingSource.forceCancel();
        harness.waitForTaskCompletion(Long.MAX_VALUE, true);
    }