@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)); // this watermark should be ignored
	testHarness.processElement(new StreamRecord<>(3L, 3));
	testHarness.processElement(new StreamRecord<>(4L, 3));
	
	// validate first part of the sequence. we poll elements until our
	// watermark updates to "3", which must be the result of the "4" element.
	{
		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;
				
				// check the invariant
				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));
	// validate the next part of the sequence. we poll elements until our
	// watermark updates to "7", which must be the result of the "8" element.
	{
		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;
				
				// check the invariant
				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());
}