StatusWatermarkValve { public void inputWatermark(Watermark watermark, int channelIndex) { if (lastOutputStreamStatus.isActive() && channelStatuses[channelIndex].streamStatus.isActive()) { long watermarkMillis = watermark.getTimestamp(); if (watermarkMillis > channelStatuses[channelIndex].watermark) { channelStatuses[channelIndex].watermark = watermarkMillis; if (!channelStatuses[channelIndex].isWatermarkAligned && watermarkMillis >= lastOutputWatermark) { channelStatuses[channelIndex].isWatermarkAligned = true; } findAndOutputNewMinWatermarkAcrossAlignedChannels(); } } } StatusWatermarkValve(int numInputChannels, ValveOutputHandler outputHandler); void inputWatermark(Watermark watermark, int channelIndex); void inputStreamStatus(StreamStatus streamStatus, int channelIndex);  }
@Test public void testSingleInputIncreasingWatermarks() { BufferedValveOutputHandler valveOutput = new BufferedValveOutputHandler(); StatusWatermarkValve valve = new StatusWatermarkValve(1, valveOutput); valve.inputWatermark(new Watermark(0), 0); assertEquals(new Watermark(0), valveOutput.popLastSeenOutput()); assertEquals(null, valveOutput.popLastSeenOutput()); valve.inputWatermark(new Watermark(25), 0); assertEquals(new Watermark(25), valveOutput.popLastSeenOutput()); assertEquals(null, valveOutput.popLastSeenOutput()); }
@Test public void testSingleInputDecreasingWatermarksYieldsNoOutput() { BufferedValveOutputHandler valveOutput = new BufferedValveOutputHandler(); StatusWatermarkValve valve = new StatusWatermarkValve(1, valveOutput); valve.inputWatermark(new Watermark(25), 0); assertEquals(new Watermark(25), valveOutput.popLastSeenOutput()); valve.inputWatermark(new Watermark(18), 0); assertEquals(null, valveOutput.popLastSeenOutput()); valve.inputWatermark(new Watermark(42), 0); assertEquals(new Watermark(42), valveOutput.popLastSeenOutput()); assertEquals(null, valveOutput.popLastSeenOutput()); }
@Test public void testMultipleInputYieldsWatermarkOnlyWhenAllChannelsReceivesWatermarks() { BufferedValveOutputHandler valveOutput = new BufferedValveOutputHandler(); StatusWatermarkValve valve = new StatusWatermarkValve(3, valveOutput); valve.inputWatermark(new Watermark(0), 0); valve.inputWatermark(new Watermark(0), 1); assertEquals(null, valveOutput.popLastSeenOutput()); valve.inputWatermark(new Watermark(0), 2); assertEquals(new Watermark(0), valveOutput.popLastSeenOutput()); assertEquals(null, valveOutput.popLastSeenOutput()); }
@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); assertEquals(new Watermark(10), valveOutput.popLastSeenOutput()); assertEquals(null, valveOutput.popLastSeenOutput()); valve.inputWatermark(new Watermark(17), 2); assertEquals(new Watermark(12), valveOutput.popLastSeenOutput()); assertEquals(null, valveOutput.popLastSeenOutput()); valve.inputWatermark(new Watermark(20), 0); assertEquals(new Watermark(15), valveOutput.popLastSeenOutput()); assertEquals(null, valveOutput.popLastSeenOutput()); }
@Test public void testMultipleInputDecreasingWatermarksYieldsNoOutput() { BufferedValveOutputHandler valveOutput = new BufferedValveOutputHandler(); StatusWatermarkValve valve = new StatusWatermarkValve(3, valveOutput); valve.inputWatermark(new Watermark(25), 0); valve.inputWatermark(new Watermark(10), 1); valve.inputWatermark(new Watermark(17), 2); assertEquals(new Watermark(10), valveOutput.popLastSeenOutput()); valve.inputWatermark(new Watermark(12), 0); valve.inputWatermark(new Watermark(8), 1); valve.inputWatermark(new Watermark(15), 2); assertEquals(null, valveOutput.popLastSeenOutput()); }