@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());
	}
}