@Override
    public void processWatermark(org.apache.flink.streaming.api.watermark.Watermark mark)
            throws Exception {
        // if we receive a Long.MAX_VALUE watermark we forward it since it is used
        // to signal the end of input and to not block watermark progress downstream
        if (mark.getTimestamp() == Long.MAX_VALUE) {
            wmOutput.emitWatermark(Watermark.MAX_WATERMARK);
        }
    }
@Test
    void inputWatermarksAreNotForwarded() throws Exception {
        OneInputStreamOperatorTestHarness<Long, Long> testHarness =
                createTestHarness(
                        WatermarkStrategy.forGenerator((ctx) -> new PeriodicWatermarkGenerator())
                                .withTimestampAssigner((ctx) -> new LongExtractor()));

        testHarness.processWatermark(createLegacyWatermark(42L));
        testHarness.setProcessingTime(AUTO_WATERMARK_INTERVAL);

        assertThat(testHarness.getOutput()).isEmpty();
    }