flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/BarrierBufferTest.java
/**
	 * Validates that the buffer skips over a later checkpoint if it
	 * receives a barrier from an even later checkpoint on a blocked input.
	 */
Validates that the buffer skips over a later checkpoint if it
receives a barrier from an even later checkpoint on a blocked input.
Validates that the buffer skips over a later checkpoint if it receives a barrier from an even later checkpoint on a blocked input .
@Test
	public void testMultiChannelSkippingCheckpointsViaBlockedInputs() {
		try {
			BufferOrEvent[] sequence = {
					createBuffer(0), createBuffer(2), createBuffer(0),
					createBarrier(1, 1), createBarrier(1, 2),
					createBuffer(2), createBuffer(1), createBuffer(0),
					createBarrier(1, 0),
					createBuffer(1), createBuffer(0),
					createBarrier(2, 1),
					createBuffer(1), createBuffer(2),
					createBarrier(2, 0),
					createBuffer(1), createBuffer(0),
					createBarrier(3, 0), 
					createBuffer(0),
					createBarrier(4, 1), 
					createBuffer(1),
					createBuffer(0),
					createBuffer(2),
					createBarrier(2, 2),
					createBuffer(0),
					createBarrier(3, 2), 
					createBuffer(2),
					createBarrier(4, 0),
					createBuffer(0), createBuffer(1), createBuffer(2),
					createBarrier(4, 2),
					createBuffer(1), createEndOfPartition(1),
					createBuffer(2), createEndOfPartition(2),
					createBuffer(0), createEndOfPartition(0)
			};
			MockInputGate gate = new MockInputGate(PAGE_SIZE, 3, Arrays.asList(sequence));
			BarrierBuffer buffer = new BarrierBuffer(gate, ioManager);
			check(sequence[0], buffer.getNextNonBlocked());
			check(sequence[1], buffer.getNextNonBlocked());
			check(sequence[2], buffer.getNextNonBlocked());
			check(sequence[7], buffer.getNextNonBlocked());
			assertEquals(1L, buffer.getCurrentCheckpointId());
			check(sequence[5], buffer.getNextNonBlocked());
			check(sequence[6], buffer.getNextNonBlocked());
			check(sequence[9], buffer.getNextNonBlocked());
			check(sequence[10], buffer.getNextNonBlocked());
			check(sequence[13], buffer.getNextNonBlocked());
			check(sequence[22], buffer.getNextNonBlocked());
			assertEquals(2L, buffer.getCurrentCheckpointId());
			check(sequence[12], buffer.getNextNonBlocked());
			check(sequence[15], buffer.getNextNonBlocked());
			check(sequence[16], buffer.getNextNonBlocked());
			check(sequence[18], buffer.getNextNonBlocked());
			assertEquals(4L, buffer.getCurrentCheckpointId());
			check(sequence[21], buffer.getNextNonBlocked());
			check(sequence[24], buffer.getNextNonBlocked());
			check(sequence[26], buffer.getNextNonBlocked());
			check(sequence[30], buffer.getNextNonBlocked());
			check(sequence[20], buffer.getNextNonBlocked());
			check(sequence[28], buffer.getNextNonBlocked());
			check(sequence[29], buffer.getNextNonBlocked());
			check(sequence[32], buffer.getNextNonBlocked());
			check(sequence[33], buffer.getNextNonBlocked());
			check(sequence[34], buffer.getNextNonBlocked());
			check(sequence[35], buffer.getNextNonBlocked());
			check(sequence[36], buffer.getNextNonBlocked());
			check(sequence[37], buffer.getNextNonBlocked());
			assertNull(buffer.getNextNonBlocked());
			assertNull(buffer.getNextNonBlocked());
			buffer.cleanup();
			checkNoTempFilesRemain();
		}
		catch (Exception e) {
			e.printStackTrace();
			fail(e.getMessage());
		}
	}
@ Test public void testMultiChannelSkippingCheckpointsViaBlockedInputs ( ) { try { BufferOrEvent [ ] sequence = { createBuffer ( 0 ) , createBuffer ( 2 ) , createBuffer ( 0 ) , createBarrier ( 1 , 1 ) , createBarrier ( 1 , 2 ) , createBuffer ( 2 ) , createBuffer ( 1 ) , createBuffer ( 0 ) , createBarrier ( 1 , 0 ) , createBuffer ( 1 ) , createBuffer ( 0 ) , createBarrier ( 2 , 1 ) , createBuffer ( 1 ) , createBuffer ( 2 ) , createBarrier ( 2 , 0 ) , createBuffer ( 1 ) , createBuffer ( 0 ) , createBarrier ( 3 , 0 ) , createBuffer ( 0 ) , createBarrier ( 4 , 1 ) , createBuffer ( 1 ) , createBuffer ( 0 ) , createBuffer ( 2 ) , createBarrier ( 2 , 2 ) , createBuffer ( 0 ) , createBarrier ( 3 , 2 ) , createBuffer ( 2 ) , createBarrier ( 4 , 0 ) , createBuffer ( 0 ) , createBuffer ( 1 ) , createBuffer ( 2 ) , createBarrier ( 4 , 2 ) , createBuffer ( 1 ) , createEndOfPartition ( 1 ) , createBuffer ( 2 ) , createEndOfPartition ( 2 ) , createBuffer ( 0 ) , createEndOfPartition ( 0 ) } ; MockInputGate gate = new MockInputGate ( PAGE_SIZE , 3 , Arrays . asList ( sequence ) ) ; BarrierBuffer buffer = new BarrierBuffer ( gate , ioManager ) ; check ( sequence [ 0 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 1 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 2 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 7 ] , buffer . getNextNonBlocked ( ) ) ; assertEquals ( 1L , buffer . getCurrentCheckpointId ( ) ) ; check ( sequence [ 5 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 6 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 9 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 10 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 13 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 22 ] , buffer . getNextNonBlocked ( ) ) ; assertEquals ( 2L , buffer . getCurrentCheckpointId ( ) ) ; check ( sequence [ 12 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 15 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 16 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 18 ] , buffer . getNextNonBlocked ( ) ) ; assertEquals ( 4L , buffer . getCurrentCheckpointId ( ) ) ; check ( sequence [ 21 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 24 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 26 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 30 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 20 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 28 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 29 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 32 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 33 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 34 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 35 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 36 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 37 ] , buffer . getNextNonBlocked ( ) ) ; assertNull ( buffer . getNextNonBlocked ( ) ) ; assertNull ( buffer . getNextNonBlocked ( ) ) ; buffer . cleanup ( ) ; checkNoTempFilesRemain ( ) ; } catch ( Exception e ) { e . printStackTrace ( ) ; fail ( e . getMessage ( ) ) ; } }
Validates that the buffer skips over a later checkpoint if it
 receives a barrier from an even later checkpoint on a blocked input.
Validates that the buffer skips over a later checkpoint if it receives a barrier from an even later checkpoint on a blocked input .
// checkpoint 1 - with blocked data // checkpoint 2 will not complete: pre-mature barrier from checkpoint 3 // queued barrier on blocked input // pre-mature barrier on blocked input // complete checkpoint 2 // should be ignored // checkpoint 1 // alignment of checkpoint 2 // checkpoint 2 completed // checkpoint 3 skipped, alignment for 4 started // checkpoint 4 completed