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 for inputs with
	 * multiple input channels, by buffering and blocking certain inputs.
	 */
Validates that the buffer correctly aligns the streams for inputs with
multiple input channels, by buffering and blocking certain inputs.
Validates that the buffer correctly aligns the streams for inputs with multiple input channels by buffering and blocking certain inputs .
@Test
	public void testMultiChannelWithBarriers() {
		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(0), createBuffer(0), createBuffer(1), createBuffer(1), createBuffer(2),
					createBarrier(2, 0), createBarrier(2, 1), createBarrier(2, 2),
					createBuffer(2), createBuffer(2),
					createBarrier(3, 2),
					createBuffer(2), createBuffer(2),
					createBarrier(3, 0), createBarrier(3, 1),
					createBarrier(4, 1), createBarrier(4, 2), createBarrier(4, 0),
					 in mixed order
					createBuffer(0), createBuffer(2), createBuffer(0),
					createBarrier(5, 1),
					createBuffer(2), createBuffer(0), createBuffer(2), createBuffer(1),
					createBarrier(5, 2),
					createBuffer(1), createBuffer(0), createBuffer(2), createBuffer(1),
					createBarrier(5, 0),
					createBuffer(0),
					createEndOfPartition(0), createEndOfPartition(1), createEndOfPartition(2)
			};
			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());
			assertEquals(1L, handler.getNextExpectedCheckpointId());
			long startTs = System.nanoTime();
			check(sequence[7], buffer.getNextNonBlocked());
			assertEquals(1L, handler.getNextExpectedCheckpointId());
			check(sequence[5], buffer.getNextNonBlocked());
			assertEquals(2L, handler.getNextExpectedCheckpointId());
			validateAlignmentTime(startTs, buffer);
			validateAlignmentBuffered(handler.getLastReportedBytesBufferedInAlignment(), sequence[5], sequence[6]);
			check(sequence[6], buffer.getNextNonBlocked());
			check(sequence[9], buffer.getNextNonBlocked());
			check(sequence[10], buffer.getNextNonBlocked());
			check(sequence[11], buffer.getNextNonBlocked());
			check(sequence[12], buffer.getNextNonBlocked());
			check(sequence[13], buffer.getNextNonBlocked());
			assertEquals(2L, handler.getNextExpectedCheckpointId());
			startTs = System.nanoTime();
			check(sequence[17], buffer.getNextNonBlocked());
			assertEquals(3L, handler.getNextExpectedCheckpointId());
			validateAlignmentTime(startTs, buffer);
			validateAlignmentBuffered(handler.getLastReportedBytesBufferedInAlignment());
			check(sequence[18], buffer.getNextNonBlocked());
			check(sequence[20], buffer.getNextNonBlocked());
			validateAlignmentBuffered(handler.getLastReportedBytesBufferedInAlignment(), sequence[20], sequence[21]);
			assertEquals(4L, handler.getNextExpectedCheckpointId());
			check(sequence[21], buffer.getNextNonBlocked());
			check(sequence[27], buffer.getNextNonBlocked());
			validateAlignmentBuffered(handler.getLastReportedBytesBufferedInAlignment());
			assertEquals(5L, handler.getNextExpectedCheckpointId());
			check(sequence[28], buffer.getNextNonBlocked());
			check(sequence[29], buffer.getNextNonBlocked());
			check(sequence[31], buffer.getNextNonBlocked());
			check(sequence[32], buffer.getNextNonBlocked());
			check(sequence[33], buffer.getNextNonBlocked());
			check(sequence[37], buffer.getNextNonBlocked());
			check(sequence[34], buffer.getNextNonBlocked());
			check(sequence[36], buffer.getNextNonBlocked());
			check(sequence[38], buffer.getNextNonBlocked());
			check(sequence[39], buffer.getNextNonBlocked());
			check(sequence[41], buffer.getNextNonBlocked());
			check(sequence[42], buffer.getNextNonBlocked());
			check(sequence[43], buffer.getNextNonBlocked());
			check(sequence[44], buffer.getNextNonBlocked());
			assertNull(buffer.getNextNonBlocked());
			assertNull(buffer.getNextNonBlocked());
			validateAlignmentBuffered(handler.getLastReportedBytesBufferedInAlignment(),
				sequence[34], sequence[36], sequence[38], sequence[39]);
			buffer.cleanup();
			checkNoTempFilesRemain();
		}
		catch (Exception e) {
			e.printStackTrace();
			fail(e.getMessage());
		}
	}
@ Test public void testMultiChannelWithBarriers ( ) { 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 ( 0 ) , createBuffer ( 0 ) , createBuffer ( 1 ) , createBuffer ( 1 ) , createBuffer ( 2 ) , createBarrier ( 2 , 0 ) , createBarrier ( 2 , 1 ) , createBarrier ( 2 , 2 ) , createBuffer ( 2 ) , createBuffer ( 2 ) , createBarrier ( 3 , 2 ) , createBuffer ( 2 ) , createBuffer ( 2 ) , createBarrier ( 3 , 0 ) , createBarrier ( 3 , 1 ) , createBarrier ( 4 , 1 ) , createBarrier ( 4 , 2 ) , createBarrier ( 4 , 0 ) , createBuffer ( 0 ) , createBuffer ( 2 ) , createBuffer ( 0 ) , createBarrier ( 5 , 1 ) , createBuffer ( 2 ) , createBuffer ( 0 ) , createBuffer ( 2 ) , createBuffer ( 1 ) , createBarrier ( 5 , 2 ) , createBuffer ( 1 ) , createBuffer ( 0 ) , createBuffer ( 2 ) , createBuffer ( 1 ) , createBarrier ( 5 , 0 ) , createBuffer ( 0 ) , createEndOfPartition ( 0 ) , createEndOfPartition ( 1 ) , createEndOfPartition ( 2 ) } ; 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 ( ) ) ; assertEquals ( 1L , handler . getNextExpectedCheckpointId ( ) ) ; long startTs = System . nanoTime ( ) ; check ( sequence [ 7 ] , buffer . getNextNonBlocked ( ) ) ; assertEquals ( 1L , handler . getNextExpectedCheckpointId ( ) ) ; check ( sequence [ 5 ] , buffer . getNextNonBlocked ( ) ) ; assertEquals ( 2L , handler . getNextExpectedCheckpointId ( ) ) ; validateAlignmentTime ( startTs , buffer ) ; validateAlignmentBuffered ( handler . getLastReportedBytesBufferedInAlignment ( ) , sequence [ 5 ] , sequence [ 6 ] ) ; check ( sequence [ 6 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 9 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 10 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 11 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 12 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 13 ] , buffer . getNextNonBlocked ( ) ) ; assertEquals ( 2L , handler . getNextExpectedCheckpointId ( ) ) ; startTs = System . nanoTime ( ) ; check ( sequence [ 17 ] , buffer . getNextNonBlocked ( ) ) ; assertEquals ( 3L , handler . getNextExpectedCheckpointId ( ) ) ; validateAlignmentTime ( startTs , buffer ) ; validateAlignmentBuffered ( handler . getLastReportedBytesBufferedInAlignment ( ) ) ; check ( sequence [ 18 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 20 ] , buffer . getNextNonBlocked ( ) ) ; validateAlignmentBuffered ( handler . getLastReportedBytesBufferedInAlignment ( ) , sequence [ 20 ] , sequence [ 21 ] ) ; assertEquals ( 4L , handler . getNextExpectedCheckpointId ( ) ) ; check ( sequence [ 21 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 27 ] , buffer . getNextNonBlocked ( ) ) ; validateAlignmentBuffered ( handler . getLastReportedBytesBufferedInAlignment ( ) ) ; assertEquals ( 5L , handler . getNextExpectedCheckpointId ( ) ) ; check ( sequence [ 28 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 29 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 31 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 32 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 33 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 37 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 34 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 36 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 38 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 39 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 41 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 42 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 43 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 44 ] , buffer . getNextNonBlocked ( ) ) ; assertNull ( buffer . getNextNonBlocked ( ) ) ; assertNull ( buffer . getNextNonBlocked ( ) ) ; validateAlignmentBuffered ( handler . getLastReportedBytesBufferedInAlignment ( ) , sequence [ 34 ] , sequence [ 36 ] , sequence [ 38 ] , sequence [ 39 ] ) ; buffer . cleanup ( ) ; checkNoTempFilesRemain ( ) ; } catch ( Exception e ) { e . printStackTrace ( ) ; fail ( e . getMessage ( ) ) ; } }
Validates that the buffer correctly aligns the streams for inputs with
 multiple input channels, by buffering and blocking certain inputs.
Validates that the buffer correctly aligns the streams for inputs with multiple input channels by buffering and blocking certain inputs .
// checkpoint with blocked data // checkpoint without blocked data // checkpoint with data only from one channel // empty checkpoint // checkpoint with blocked data in mixed order // some trailing data // pre checkpoint 1 // blocking while aligning for checkpoint 1 // checkpoint 1 done, returning buffered data // pre checkpoint 2 // checkpoint 2 barriers come together // checkpoint 3 starts, data buffered // checkpoint 4 happens without extra data // pre checkpoint 5 // checkpoint 5 aligning // buffered data from checkpoint 5 alignment // remaining data