BarrierTracker implements CheckpointBarrierHandler { @Override public BufferOrEvent getNextNonBlocked() throws Exception { while (true) { BufferOrEvent next = inputGate.getNextBufferOrEvent(); if (next == null || next.isBuffer()) { return next; } else if (next.getEvent().getClass() == CheckpointBarrier.class) { processBarrier((CheckpointBarrier) next.getEvent(), next.getChannelIndex()); } else if (next.getEvent().getClass() == CancelCheckpointMarker.class) { processCheckpointAbortBarrier((CancelCheckpointMarker) next.getEvent(), next.getChannelIndex()); } else { return next; } } } BarrierTracker(InputGate inputGate); @Override BufferOrEvent getNextNonBlocked(); @Override void registerCheckpointEventHandler(StatefulTask toNotifyOnCheckpoint); @Override void cleanup(); @Override boolean isEmpty(); @Override long getAlignmentDurationNanos();  }
@Test public void testSingleChannelNoBarriers() { try { BufferOrEvent[] sequence = { createBuffer(0), createBuffer(0), createBuffer(0) }; MockInputGate gate = new MockInputGate(PAGE_SIZE, 1, Arrays.asList(sequence)); BarrierTracker tracker = new BarrierTracker(gate); for (BufferOrEvent boe : sequence) { assertEquals(boe, tracker.getNextNonBlocked()); } assertNull(tracker.getNextNonBlocked()); assertNull(tracker.getNextNonBlocked()); } catch (Exception e) { e.printStackTrace(); fail(e.getMessage()); } }
@Test public void testMultiChannelNoBarriers() { try { BufferOrEvent[] sequence = { createBuffer(2), createBuffer(2), createBuffer(0), createBuffer(1), createBuffer(0), createBuffer(3), createBuffer(1), createBuffer(1), createBuffer(2) }; MockInputGate gate = new MockInputGate(PAGE_SIZE, 4, Arrays.asList(sequence)); BarrierTracker tracker = new BarrierTracker(gate); for (BufferOrEvent boe : sequence) { assertEquals(boe, tracker.getNextNonBlocked()); } assertNull(tracker.getNextNonBlocked()); assertNull(tracker.getNextNonBlocked()); } catch (Exception e) { e.printStackTrace(); fail(e.getMessage()); } }