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 correctly aligns the streams in cases
	 * where some channels receive barriers from multiple successive checkpoints
	 * before the pending checkpoint is complete.
	 */
Validates that the buffer correctly aligns the streams in cases
where some channels receive barriers from multiple successive checkpoints
before the pending checkpoint is complete.
Validates that the buffer correctly aligns the streams in cases where some channels receive barriers from multiple successive checkpoints before the pending checkpoint is complete .
@Test
	public void testMultiChannelWithQueuedFutureBarriers() {
		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),
					createBarrier(3, 0), createBuffer(0),
					createBarrier(3, 1), createBuffer(0), createBuffer(1), createBuffer(2),
					createBarrier(4, 1), createBuffer(1), createBuffer(2),
					createBarrier(2, 2),
					createBuffer(2), createBuffer(1), createBuffer(2), createBuffer(0),
					createBarrier(4, 0),
					createBuffer(2), createBuffer(1), createBuffer(2), createBuffer(0),
					createBarrier(5, 1),
					createBarrier(3, 2),
					createBuffer(2), createBuffer(1), createBuffer(2), createBuffer(0),
					createBarrier(6, 1),
					createBarrier(4, 2),
					createBuffer(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);
			ValidatingCheckpointHandler handler = new ValidatingCheckpointHandler();
			buffer.registerCheckpointEventHandler(handler);
			handler.setNextExpectedCheckpointId(1L);
			check(sequence[0], buffer.getNextNonBlocked());
			check(sequence[1], buffer.getNextNonBlocked());
			check(sequence[2], buffer.getNextNonBlocked());
			check(sequence[7], buffer.getNextNonBlocked());
			check(sequence[5], buffer.getNextNonBlocked());
			assertEquals(2L, handler.getNextExpectedCheckpointId());
			check(sequence[6], buffer.getNextNonBlocked());
			check(sequence[9], buffer.getNextNonBlocked());
			check(sequence[10], buffer.getNextNonBlocked());
			long startTs = System.nanoTime();
			check(sequence[13], buffer.getNextNonBlocked());
			check(sequence[20], buffer.getNextNonBlocked());
			check(sequence[23], buffer.getNextNonBlocked());
			check(sequence[12], buffer.getNextNonBlocked());
			validateAlignmentTime(startTs, buffer);
			check(sequence[25], buffer.getNextNonBlocked());
			check(sequence[27], buffer.getNextNonBlocked());
			check(sequence[30], buffer.getNextNonBlocked());
			check(sequence[32], buffer.getNextNonBlocked());
			check(sequence[16], buffer.getNextNonBlocked());
			check(sequence[18], buffer.getNextNonBlocked());
			check(sequence[19], buffer.getNextNonBlocked());
			check(sequence[28], buffer.getNextNonBlocked());
			check(sequence[36], buffer.getNextNonBlocked());
			check(sequence[38], buffer.getNextNonBlocked());
			check(sequence[22], buffer.getNextNonBlocked());
			check(sequence[26], buffer.getNextNonBlocked());
			check(sequence[31], buffer.getNextNonBlocked());
			check(sequence[33], buffer.getNextNonBlocked());
			check(sequence[39], buffer.getNextNonBlocked());
			check(sequence[42], buffer.getNextNonBlocked());
			check(sequence[45], buffer.getNextNonBlocked());
			check(sequence[46], buffer.getNextNonBlocked());
			check(sequence[37], buffer.getNextNonBlocked());
			check(sequence[47], buffer.getNextNonBlocked());
			check(sequence[48], buffer.getNextNonBlocked());
			check(sequence[43], buffer.getNextNonBlocked());
			check(sequence[44], buffer.getNextNonBlocked());
			assertNull(buffer.getNextNonBlocked());
			assertNull(buffer.getNextNonBlocked());
			buffer.cleanup();
			checkNoTempFilesRemain();
		}
		catch (Exception e) {
			e.printStackTrace();
			fail(e.getMessage());
		}
	}
@ Test public void testMultiChannelWithQueuedFutureBarriers ( ) { 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 ) , createBarrier ( 3 , 0 ) , createBuffer ( 0 ) , createBarrier ( 3 , 1 ) , createBuffer ( 0 ) , createBuffer ( 1 ) , createBuffer ( 2 ) , createBarrier ( 4 , 1 ) , createBuffer ( 1 ) , createBuffer ( 2 ) , createBarrier ( 2 , 2 ) , createBuffer ( 2 ) , createBuffer ( 1 ) , createBuffer ( 2 ) , createBuffer ( 0 ) , createBarrier ( 4 , 0 ) , createBuffer ( 2 ) , createBuffer ( 1 ) , createBuffer ( 2 ) , createBuffer ( 0 ) , createBarrier ( 5 , 1 ) , createBarrier ( 3 , 2 ) , createBuffer ( 2 ) , createBuffer ( 1 ) , createBuffer ( 2 ) , createBuffer ( 0 ) , createBarrier ( 6 , 1 ) , createBarrier ( 4 , 2 ) , createBuffer ( 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 ) ; ValidatingCheckpointHandler handler = new ValidatingCheckpointHandler ( ) ; buffer . registerCheckpointEventHandler ( handler ) ; handler . setNextExpectedCheckpointId ( 1L ) ; check ( sequence [ 0 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 1 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 2 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 7 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 5 ] , buffer . getNextNonBlocked ( ) ) ; assertEquals ( 2L , handler . getNextExpectedCheckpointId ( ) ) ; check ( sequence [ 6 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 9 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 10 ] , buffer . getNextNonBlocked ( ) ) ; long startTs = System . nanoTime ( ) ; check ( sequence [ 13 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 20 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 23 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 12 ] , buffer . getNextNonBlocked ( ) ) ; validateAlignmentTime ( startTs , buffer ) ; check ( sequence [ 25 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 27 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 30 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 32 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 16 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 18 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 19 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 28 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 36 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 38 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 22 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 26 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 31 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 33 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 39 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 42 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 45 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 46 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 37 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 47 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 48 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 43 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 44 ] , buffer . getNextNonBlocked ( ) ) ; assertNull ( buffer . getNextNonBlocked ( ) ) ; assertNull ( buffer . getNextNonBlocked ( ) ) ; buffer . cleanup ( ) ; checkNoTempFilesRemain ( ) ; } catch ( Exception e ) { e . printStackTrace ( ) ; fail ( e . getMessage ( ) ) ; } }
Validates that the buffer correctly aligns the streams in cases
 where some channels receive barriers from multiple successive checkpoints
 before the pending checkpoint is complete.
Validates that the buffer correctly aligns the streams in cases where some channels receive barriers from multiple successive checkpoints before the pending checkpoint is complete .
// checkpoint 1 - with blocked data // checkpoint 2 - where future checkpoint barriers come before // the current checkpoint is complete // complete checkpoint 2, send a barrier for checkpoints 4 and 5 // complete checkpoint 3 // complete checkpoint 4, checkpoint 5 remains not fully triggered // around checkpoint 1 // alignment of checkpoint 2 - buffering also some barriers for // checkpoints 3 and 4 // checkpoint 2 completed // checkpoint 3 completed (emit buffered) // past checkpoint 3 // checkpoint 4 completed (emit buffered) // past checkpoint 4, alignment for checkpoint 5 // abort checkpoint 5 (end of partition) // start checkpoint 6 alignment // end of input, emit remainder