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 the current checkpoint if it
	 * receives a barrier from a later checkpoint on a non-blocked input.
	 */
Validates that the buffer skips over the current checkpoint if it
receives a barrier from a later checkpoint on a non-blocked input.
Validates that the buffer skips over the current checkpoint if it receives a barrier from a later checkpoint on a non - blocked input .
@Test
	public void testMultiChannelSkippingCheckpoints() {
		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(2), createBuffer(0),
					createBarrier(3, 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);
			StatefulTask toNotify = mock(StatefulTask.class);
			buffer.registerCheckpointEventHandler(toNotify);
			long startTs;
			check(sequence[0], buffer.getNextNonBlocked());
			check(sequence[1], buffer.getNextNonBlocked());
			check(sequence[2], buffer.getNextNonBlocked());
			startTs = System.nanoTime();
			check(sequence[7], buffer.getNextNonBlocked());
			assertEquals(1L, buffer.getCurrentCheckpointId());
			check(sequence[5], buffer.getNextNonBlocked());
			validateAlignmentTime(startTs, buffer);
			verify(toNotify).triggerCheckpointOnBarrier(argThat(new CheckpointMatcher(1L)), any(CheckpointOptions.class), any(CheckpointMetrics.class));
			check(sequence[6], buffer.getNextNonBlocked());
			check(sequence[9], buffer.getNextNonBlocked());
			check(sequence[10], buffer.getNextNonBlocked());
			startTs = System.nanoTime();
			check(sequence[13], buffer.getNextNonBlocked());
			check(sequence[15], buffer.getNextNonBlocked());
			check(sequence[12], buffer.getNextNonBlocked());
			assertEquals(3L, buffer.getCurrentCheckpointId());
			validateAlignmentTime(startTs, buffer);
			verify(toNotify).abortCheckpointOnBarrier(eq(2L), any(CheckpointDeclineSubsumedException.class));
			check(sequence[16], buffer.getNextNonBlocked());
			check(sequence[19], buffer.getNextNonBlocked());
			check(sequence[20], buffer.getNextNonBlocked());
			verify(toNotify).abortCheckpointOnBarrier(eq(3L), any(CheckpointDeclineSubsumedException.class));
			check(sequence[18], buffer.getNextNonBlocked());
			check(sequence[21], buffer.getNextNonBlocked());
			check(sequence[22], buffer.getNextNonBlocked());
			check(sequence[23], buffer.getNextNonBlocked());
			check(sequence[24], buffer.getNextNonBlocked());
			assertNull(buffer.getNextNonBlocked());
			assertNull(buffer.getNextNonBlocked());
			buffer.cleanup();
			checkNoTempFilesRemain();
		}
		catch (Exception e) {
			e.printStackTrace();
			fail(e.getMessage());
		}
	}
@ Test public void testMultiChannelSkippingCheckpoints ( ) { 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 ( 2 ) , createBuffer ( 0 ) , createBarrier ( 3 , 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 ) ; StatefulTask toNotify = mock ( StatefulTask . class ) ; buffer . registerCheckpointEventHandler ( toNotify ) ; long startTs ; check ( sequence [ 0 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 1 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 2 ] , buffer . getNextNonBlocked ( ) ) ; startTs = System . nanoTime ( ) ; check ( sequence [ 7 ] , buffer . getNextNonBlocked ( ) ) ; assertEquals ( 1L , buffer . getCurrentCheckpointId ( ) ) ; check ( sequence [ 5 ] , buffer . getNextNonBlocked ( ) ) ; validateAlignmentTime ( startTs , buffer ) ; verify ( toNotify ) . triggerCheckpointOnBarrier ( argThat ( new CheckpointMatcher ( 1L ) ) , any ( CheckpointOptions . class ) , any ( CheckpointMetrics . class ) ) ; check ( sequence [ 6 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 9 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 10 ] , buffer . getNextNonBlocked ( ) ) ; startTs = System . nanoTime ( ) ; check ( sequence [ 13 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 15 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 12 ] , buffer . getNextNonBlocked ( ) ) ; assertEquals ( 3L , buffer . getCurrentCheckpointId ( ) ) ; validateAlignmentTime ( startTs , buffer ) ; verify ( toNotify ) . abortCheckpointOnBarrier ( eq ( 2L ) , any ( CheckpointDeclineSubsumedException . class ) ) ; check ( sequence [ 16 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 19 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 20 ] , buffer . getNextNonBlocked ( ) ) ; verify ( toNotify ) . abortCheckpointOnBarrier ( eq ( 3L ) , any ( CheckpointDeclineSubsumedException . class ) ) ; check ( sequence [ 18 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 21 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 22 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 23 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 24 ] , 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 the current checkpoint if it
 receives a barrier from a later checkpoint on a non-blocked input.
Validates that the buffer skips over the current checkpoint if it receives a barrier from a later checkpoint on a non - blocked input .
// checkpoint 1 - with blocked data // checkpoint 2 will not complete: pre-mature barrier from checkpoint 3 // initial data // align checkpoint 1 // checkpoint done - replay buffered // alignment of checkpoint 2 // checkpoint 2 aborted, checkpoint 3 started // checkpoint 3 alignment in progress // checkpoint 3 aborted (end of partition) // replay buffered data from checkpoint 3 // all the remaining messages