TimestampsAndPunctuatedWatermarksOperator extends AbstractUdfStreamOperator<T, AssignerWithPunctuatedWatermarks<T>> implements OneInputStreamOperator<T, T> { @Override public void processElement(StreamRecord<T> element) throws Exception { final T value = element.getValue(); final long newTimestamp = userFunction.extractTimestamp(value, element.hasTimestamp() ? element.getTimestamp() : Long.MIN_VALUE); output.collect(element.replace(element.getValue(), newTimestamp)); final Watermark nextWatermark = userFunction.checkAndGetNextWatermark(value, newTimestamp); if (nextWatermark != null && nextWatermark.getTimestamp() > currentWatermark) { currentWatermark = nextWatermark.getTimestamp(); output.emitWatermark(nextWatermark); } } TimestampsAndPunctuatedWatermarksOperator(AssignerWithPunctuatedWatermarks<T> assigner); @Override void processElement(StreamRecord<T> element); @Override void processWatermark(Watermark mark);  }
@Test public void testZeroOnNegativeTimestamps() throws Exception { final AssignerWithPunctuatedWatermarks<Long> assigner = new NeverWatermarkExtractor(); final TimestampsAndPunctuatedWatermarksOperator<Long> operator = new TimestampsAndPunctuatedWatermarksOperator<Long>(assigner); OneInputStreamOperatorTestHarness<Long, Long> testHarness = new OneInputStreamOperatorTestHarness<Long, Long>(operator); testHarness.open(); long[] values = { Long.MIN_VALUE, -1L, 0L, 1L, 2L, 3L, Long.MAX_VALUE }; for (long value : values) { testHarness.processElement(new StreamRecord<>(value)); } ConcurrentLinkedQueue<Object> output = testHarness.getOutput(); for (long value: values) { assertEquals(value, ((StreamRecord<?>) output.poll()).getTimestamp()); } }