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
/**
	 * This tests the where a cancellation barrier is received for a checkpoint already
	 * canceled due to receiving a newer checkpoint barrier.
	 */
This tests the where a cancellation barrier is received for a checkpoint already
canceled due to receiving a newer checkpoint barrier.
This tests the where a cancellation barrier is received for a checkpoint already canceled due to receiving a newer checkpoint barrier .
@Test
	public void testIgnoreCancelBarrierIfCheckpointSubsumed() throws Exception {
		BufferOrEvent[] sequence = {
				/*  0 */ createBuffer(2),
				/*  1 */ createBarrier(3, 1), createBarrier(3, 0),
				/*  3 */ createBuffer(0), createBuffer(1), createBuffer(2),
				/*  6 */ createBarrier(5, 2),
				/*  7 */ createBuffer(2), createBuffer(1), createBuffer(0),
				/* 10 */ createCancellationBarrier(3, 2),
				/* 11 */ createBuffer(2), createBuffer(0), createBuffer(1),
				/* 14 */ createBarrier(5, 0), createBarrier(5, 1),
				/* 16 */ createBuffer(0), createBuffer(1), createBuffer(2)
		};
		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[5], buffer.getNextNonBlocked());
		startTs = System.nanoTime();
		check(sequence[3], buffer.getNextNonBlocked());
		verify(toNotify, times(1)).abortCheckpointOnBarrier(eq(3L), any(CheckpointDeclineSubsumedException.class));
		check(sequence[4], buffer.getNextNonBlocked());
		check(sequence[8], buffer.getNextNonBlocked());
		check(sequence[9], buffer.getNextNonBlocked());
		check(sequence[12], buffer.getNextNonBlocked());
		check(sequence[13], buffer.getNextNonBlocked());
		check(sequence[7], buffer.getNextNonBlocked());
		validateAlignmentTime(startTs, buffer);
		verify(toNotify, times(1)).triggerCheckpointOnBarrier(argThat(new CheckpointMatcher(5L)), any(CheckpointOptions.class), any(CheckpointMetrics.class));
		check(sequence[11], buffer.getNextNonBlocked());
		check(sequence[16], buffer.getNextNonBlocked());
		check(sequence[17], buffer.getNextNonBlocked());
		check(sequence[18], buffer.getNextNonBlocked());
		assertNull(buffer.getNextNonBlocked());
		assertNull(buffer.getNextNonBlocked());
		buffer.cleanup();
		checkNoTempFilesRemain();
		verify(toNotify, times(1)).triggerCheckpointOnBarrier(any(CheckpointMetaData.class), any(CheckpointOptions.class), any(CheckpointMetrics.class));
		verify(toNotify, times(1)).abortCheckpointOnBarrier(anyLong(), any(Throwable.class));
	}
@ Test public void testIgnoreCancelBarrierIfCheckpointSubsumed ( ) throws Exception { BufferOrEvent [ ] sequence = { /*  0 */ createBuffer ( 2 ) , /*  1 */ createBarrier ( 3 , 1 ) , createBarrier ( 3 , 0 ) , /*  3 */ createBuffer ( 0 ) , createBuffer ( 1 ) , createBuffer ( 2 ) , /*  6 */ createBarrier ( 5 , 2 ) , /*  7 */ createBuffer ( 2 ) , createBuffer ( 1 ) , createBuffer ( 0 ) , /* 10 */ createCancellationBarrier ( 3 , 2 ) , /* 11 */ createBuffer ( 2 ) , createBuffer ( 0 ) , createBuffer ( 1 ) , /* 14 */ createBarrier ( 5 , 0 ) , createBarrier ( 5 , 1 ) , /* 16 */ createBuffer ( 0 ) , createBuffer ( 1 ) , createBuffer ( 2 ) } ; 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 [ 5 ] , buffer . getNextNonBlocked ( ) ) ; startTs = System . nanoTime ( ) ; check ( sequence [ 3 ] , buffer . getNextNonBlocked ( ) ) ; verify ( toNotify , times ( 1 ) ) . abortCheckpointOnBarrier ( eq ( 3L ) , any ( CheckpointDeclineSubsumedException . class ) ) ; check ( sequence [ 4 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 8 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 9 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 12 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 13 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 7 ] , buffer . getNextNonBlocked ( ) ) ; validateAlignmentTime ( startTs , buffer ) ; verify ( toNotify , times ( 1 ) ) . triggerCheckpointOnBarrier ( argThat ( new CheckpointMatcher ( 5L ) ) , any ( CheckpointOptions . class ) , any ( CheckpointMetrics . class ) ) ; check ( sequence [ 11 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 16 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 17 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 18 ] , buffer . getNextNonBlocked ( ) ) ; assertNull ( buffer . getNextNonBlocked ( ) ) ; assertNull ( buffer . getNextNonBlocked ( ) ) ; buffer . cleanup ( ) ; checkNoTempFilesRemain ( ) ; verify ( toNotify , times ( 1 ) ) . triggerCheckpointOnBarrier ( any ( CheckpointMetaData . class ) , any ( CheckpointOptions . class ) , any ( CheckpointMetrics . class ) ) ; verify ( toNotify , times ( 1 ) ) . abortCheckpointOnBarrier ( anyLong ( ) , any ( Throwable . class ) ) ; }
This tests the where a cancellation barrier is received for a checkpoint already
 canceled due to receiving a newer checkpoint barrier.
This tests the where a cancellation barrier is received for a checkpoint already canceled due to receiving a newer checkpoint barrier .
// starting a checkpoint // newer checkpoint barrier cancels/subsumes pending checkpoint // some queued buffers // cancel barrier the initial checkpoint /it is already canceled) // some more buffers // complete next checkpoint regularly // some more buffers // validate the sequence // beginning of first checkpoint // future barrier aborts checkpoint // alignment of next checkpoint // checkpoint finished // remaining data // all done // check overall notifications