@Override
    public void processElement(final StreamRecord<T> element) throws Exception {
        final T event = element.getValue();
        final long previousTimestamp =
                element.hasTimestamp() ? element.getTimestamp() : Long.MIN_VALUE;
        final long newTimestamp = timestampAssigner.extractTimestamp(event, previousTimestamp);

        element.setTimestamp(newTimestamp);
        output.collect(element);
        watermarkGenerator.onEvent(event, newTimestamp, wmOutput);
    }
@Test
    void punctuatedWatermarksDoNotRegressStreamMode() throws Exception {
        OneInputStreamOperatorTestHarness<Tuple2<Boolean, Long>, Tuple2<Boolean, Long>>
                testHarness =
                        createTestHarness(
                                WatermarkStrategy.forGenerator(
                                                (ctx) -> new PunctuatedWatermarkGenerator())
                                        .withTimestampAssigner((ctx) -> new TupleExtractor()));

        testHarness.processElement(new StreamRecord<>(new Tuple2<>(true, 4L), 1));

        assertThat(pollNextStreamRecord(testHarness))
                .is(matching(streamRecord(new Tuple2<>(true, 4L), 4L)));
        assertThat(pollNextLegacyWatermark(testHarness)).is(matching(legacyWatermark(4L)));

        testHarness.processElement(new StreamRecord<>(new Tuple2<>(true, 2L), 1));

        assertThat(pollNextStreamRecord(testHarness))
                .is(matching(streamRecord(new Tuple2<>(true, 2L), 2L)));
        assertThat(testHarness.getOutput()).isEmpty();
    }