@Override
    protected void advanceToEndOfEventTime() throws Exception {
        for (Output<StreamRecord<?>> sourceOutput : operatorChain.getChainedSourceOutputs()) {
            sourceOutput.emitWatermark(Watermark.MAX_WATERMARK);
        }
    }
@TestTemplate
    void testAdvanceToEndOfEventTime() throws Exception {
        try (StreamTaskMailboxTestHarness<String> testHarness =
                buildWatermarkTestHarness(2, false)) {
            testHarness.processElement(Watermark.MAX_WATERMARK, 0, 0);
            testHarness.processElement(Watermark.MAX_WATERMARK, 0, 1);

            testHarness.getStreamTask().advanceToEndOfEventTime();

            testHarness.processElement(Watermark.MAX_WATERMARK, 1, 0);

            assertThat(testHarness.getOutput()).doesNotContain(Watermark.MAX_WATERMARK);

            testHarness.processElement(Watermark.MAX_WATERMARK, 1, 1);
            assertThat(testHarness.getOutput()).containsExactly(Watermark.MAX_WATERMARK);
        }
    }