@Test
public void testMultipleInputIncreasingWatermarks() {

	BufferedValveOutputHandler valveOutput = new BufferedValveOutputHandler();
	StatusWatermarkValve valve = new StatusWatermarkValve(3, valveOutput);
	valve.inputWatermark(new Watermark(0), 0);
	valve.inputWatermark(new Watermark(0), 1);
	valve.inputWatermark(new Watermark(0), 2);
	assertEquals(new Watermark(0), valveOutput.popLastSeenOutput());
	valve.inputWatermark(new Watermark(12), 0);
	valve.inputWatermark(new Watermark(8), 2);
	valve.inputWatermark(new Watermark(10), 2);
	assertEquals(null, valveOutput.popLastSeenOutput());
	valve.inputWatermark(new Watermark(15), 1);
	// lowest watermark across all channels is now channel 2, with watermark @ 10
	assertEquals(new Watermark(10), valveOutput.popLastSeenOutput());
	assertEquals(null, valveOutput.popLastSeenOutput());
	valve.inputWatermark(new Watermark(17), 2);
	// lowest watermark across all channels is now channel 0, with watermark @ 12
	assertEquals(new Watermark(12), valveOutput.popLastSeenOutput());
	assertEquals(null, valveOutput.popLastSeenOutput());
	valve.inputWatermark(new Watermark(20), 0);
	// lowest watermark across all channels is now channel 1, with watermark @ 15
	assertEquals(new Watermark(15), valveOutput.popLastSeenOutput());
	assertEquals(null, valveOutput.popLastSeenOutput());
}