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 replay of queued checkpoint barriers meets
	 * a canceled checkpoint.
	 *
	 * <p>The replayed newer checkpoint barrier must not try to cancel the
	 * already canceled checkpoint.
	 */
This tests the where a replay of queued checkpoint barriers meets
a canceled checkpoint.

The replayed newer checkpoint barrier must not try to cancel the
already canceled checkpoint.
This tests the where a replay of queued checkpoint barriers meets a canceled checkpoint . The replayed newer checkpoint barrier must not try to cancel the already canceled checkpoint .
@Test
	public void testAbortWhileHavingQueuedBarriers() throws Exception {
		BufferOrEvent[] sequence = {
				/*  0 */ createBuffer(1),
				/*  1 */ createBarrier(1, 1),
				/*  2 */ createBuffer(2), createBuffer(0), createBuffer(1),
				/*  5 */ createBarrier(2, 1),
				/*  6 */ createBuffer(2), createBuffer(1),
				/*  8 */ createCancellationBarrier(1, 0),
				/*  9 */ createBuffer(2), createBuffer(1), createBuffer(0),
				/* 12 */ createBarrier(1, 2),
				/* 13 */ createBuffer(0), createBuffer(1), createBuffer(2),
				/* 16 */ createBarrier(2, 0), createBarrier(2, 2),
				/* 18 */ 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());
		startTs = System.nanoTime();
		check(sequence[2], buffer.getNextNonBlocked());
		check(sequence[3], buffer.getNextNonBlocked());
		check(sequence[6], buffer.getNextNonBlocked());
		check(sequence[4], buffer.getNextNonBlocked());
		validateAlignmentTime(startTs, buffer);
		verify(toNotify).abortCheckpointOnBarrier(eq(1L), any(CheckpointDeclineOnCancellationBarrierException.class));
		startTs = System.nanoTime();
		check(sequence[9], buffer.getNextNonBlocked());
		check(sequence[11], buffer.getNextNonBlocked());
		check(sequence[13], buffer.getNextNonBlocked());
		check(sequence[15], buffer.getNextNonBlocked());
		check(sequence[7], buffer.getNextNonBlocked());
		validateAlignmentTime(startTs, buffer);
		verify(toNotify).triggerCheckpointOnBarrier(argThat(new CheckpointMatcher(2L)), any(CheckpointOptions.class), any(CheckpointMetrics.class));
		check(sequence[10], buffer.getNextNonBlocked());
		check(sequence[14], buffer.getNextNonBlocked());
		check(sequence[18], buffer.getNextNonBlocked());
		check(sequence[19], buffer.getNextNonBlocked());
		check(sequence[20], 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 testAbortWhileHavingQueuedBarriers ( ) throws Exception { BufferOrEvent [ ] sequence = { /*  0 */ createBuffer ( 1 ) , /*  1 */ createBarrier ( 1 , 1 ) , /*  2 */ createBuffer ( 2 ) , createBuffer ( 0 ) , createBuffer ( 1 ) , /*  5 */ createBarrier ( 2 , 1 ) , /*  6 */ createBuffer ( 2 ) , createBuffer ( 1 ) , /*  8 */ createCancellationBarrier ( 1 , 0 ) , /*  9 */ createBuffer ( 2 ) , createBuffer ( 1 ) , createBuffer ( 0 ) , /* 12 */ createBarrier ( 1 , 2 ) , /* 13 */ createBuffer ( 0 ) , createBuffer ( 1 ) , createBuffer ( 2 ) , /* 16 */ createBarrier ( 2 , 0 ) , createBarrier ( 2 , 2 ) , /* 18 */ 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 ( ) ) ; startTs = System . nanoTime ( ) ; check ( sequence [ 2 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 3 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 6 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 4 ] , buffer . getNextNonBlocked ( ) ) ; validateAlignmentTime ( startTs , buffer ) ; verify ( toNotify ) . abortCheckpointOnBarrier ( eq ( 1L ) , any ( CheckpointDeclineOnCancellationBarrierException . class ) ) ; startTs = System . nanoTime ( ) ; check ( sequence [ 9 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 11 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 13 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 15 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 7 ] , buffer . getNextNonBlocked ( ) ) ; validateAlignmentTime ( startTs , buffer ) ; verify ( toNotify ) . triggerCheckpointOnBarrier ( argThat ( new CheckpointMatcher ( 2L ) ) , any ( CheckpointOptions . class ) , any ( CheckpointMetrics . class ) ) ; check ( sequence [ 10 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 14 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 18 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 19 ] , buffer . getNextNonBlocked ( ) ) ; check ( sequence [ 20 ] , 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 replay of queued checkpoint barriers meets
 a canceled checkpoint.
This tests the where a replay of queued checkpoint barriers meets a canceled checkpoint .
// starting a checkpoint // queued barrier and cancellation barrier // some queued buffers // cancel the initial checkpoint // some more buffers // ignored barrier - already canceled and moved to next checkpoint // some more buffers // complete next checkpoint regularly // some more buffers // starting first checkpoint // cancelled by cancellation barrier // the next checkpoint alignment starts now // checkpoint done // queued data // trailing data // all done // check overall notifications