@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 testNegativeTimestamps() throws Exception {

        OneInputStreamOperatorTestHarness<Long, Long> testHarness =
                createTestHarness(
                        WatermarkStrategy.forGenerator((ctx) -> new NeverWatermarkGenerator())
                                .withTimestampAssigner((ctx) -> new LongExtractor()));

        long[] values = {Long.MIN_VALUE, -1L, 0L, 1L, 2L, 3L, Long.MAX_VALUE};

        for (long value : values) {
            testHarness.processElement(new StreamRecord<>(value));
        }

        for (long value : values) {
            assertThat(pollNextStreamRecord(testHarness).getTimestamp()).isEqualTo(value);
        }
    }