TimestampsAndPeriodicWatermarksOperator extends AbstractUdfStreamOperator<T, AssignerWithPeriodicWatermarks<T>> implements OneInputStreamOperator<T, T>, Triggerable { public TimestampsAndPeriodicWatermarksOperator(AssignerWithPeriodicWatermarks<T> assigner) { super(assigner); this.chainingStrategy = ChainingStrategy.ALWAYS; } TimestampsAndPeriodicWatermarksOperator(AssignerWithPeriodicWatermarks<T> assigner); @Override void open(); @Override void processElement(StreamRecord<T> element); @Override void trigger(long timestamp); @Override void processWatermark(Watermark mark); @Override void close();  }
@Test public void testTimestampsAndPeriodicWatermarksOperator() throws Exception { final TimestampsAndPeriodicWatermarksOperator<Long> operator = new TimestampsAndPeriodicWatermarksOperator<Long>(new LongExtractor()); final ExecutionConfig config = new ExecutionConfig(); config.setAutoWatermarkInterval(50); OneInputStreamOperatorTestHarness<Long, Long> testHarness = new OneInputStreamOperatorTestHarness<Long, Long>(operator, config); testHarness.open(); testHarness.processElement(new StreamRecord<>(1L, 1)); testHarness.processElement(new StreamRecord<>(2L, 1)); testHarness.processWatermark(new Watermark(2)); testHarness.processElement(new StreamRecord<>(3L, 3)); testHarness.processElement(new StreamRecord<>(4L, 3)); { ConcurrentLinkedQueue<Object> output = testHarness.getOutput(); long nextElementValue = 1L; long lastWatermark = -1L; while (lastWatermark < 3) { if (output.size() > 0) { Object next = output.poll(); assertNotNull(next); Tuple2<Long, Long> update = validateElement(next, nextElementValue, lastWatermark); nextElementValue = update.f0; lastWatermark = update.f1; assertTrue(lastWatermark < nextElementValue); } else { Thread.sleep(10); } } output.clear(); } testHarness.processElement(new StreamRecord<>(4L, 4)); testHarness.processElement(new StreamRecord<>(5L, 4)); testHarness.processElement(new StreamRecord<>(6L, 4)); testHarness.processElement(new StreamRecord<>(7L, 4)); testHarness.processElement(new StreamRecord<>(8L, 4)); { ConcurrentLinkedQueue<Object> output = testHarness.getOutput(); long nextElementValue = 4L; long lastWatermark = 2L; while (lastWatermark < 7) { if (output.size() > 0) { Object next = output.poll(); assertNotNull(next); Tuple2<Long, Long> update = validateElement(next, nextElementValue, lastWatermark); nextElementValue = update.f0; lastWatermark = update.f1; assertTrue(lastWatermark < nextElementValue); } else { Thread.sleep(10); } } output.clear(); } testHarness.processWatermark(new Watermark(Long.MAX_VALUE)); assertEquals(Long.MAX_VALUE, ((Watermark) testHarness.getOutput().poll()).getTimestamp()); }