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 preserved the order of elements for a
	 * input with a single input channel, and checkpoint events.
	 */
Validates that the buffer preserved the order of elements for a
input with a single input channel, and checkpoint events.
Validates that the buffer preserved the order of elements for a input with a single input channel and checkpoint events .
@Test
	public void testSingleChannelWithBarriers() {
		try {
			BufferOrEvent[] sequence = {
					createBuffer(0), createBuffer(0), createBuffer(0),
					createBarrier(1, 0),
					createBuffer(0), createBuffer(0), createBuffer(0), createBuffer(0),
					createBarrier(2, 0), createBarrier(3, 0),
					createBuffer(0), createBuffer(0),
					createBarrier(4, 0), createBarrier(5, 0), createBarrier(6, 0),
					createBuffer(0), createEndOfPartition(0)
			};
			MockInputGate gate = new MockInputGate(PAGE_SIZE, 1, Arrays.asList(sequence));
			BarrierBuffer buffer = new BarrierBuffer(gate, ioManager);
			ValidatingCheckpointHandler handler = new ValidatingCheckpointHandler();
			buffer.registerCheckpointEventHandler(handler);
			handler.setNextExpectedCheckpointId(1L);
			for (BufferOrEvent boe : sequence) {
				if (boe.isBuffer() || boe.getEvent().getClass() != CheckpointBarrier.class) {
					assertEquals(boe, buffer.getNextNonBlocked());
				}
			}
			assertNull(buffer.getNextNonBlocked());
			assertNull(buffer.getNextNonBlocked());
			buffer.cleanup();
			checkNoTempFilesRemain();
		}
		catch (Exception e) {
			e.printStackTrace();
			fail(e.getMessage());
		}
	}
@ Test public void testSingleChannelWithBarriers ( ) { try { BufferOrEvent [ ] sequence = { createBuffer ( 0 ) , createBuffer ( 0 ) , createBuffer ( 0 ) , createBarrier ( 1 , 0 ) , createBuffer ( 0 ) , createBuffer ( 0 ) , createBuffer ( 0 ) , createBuffer ( 0 ) , createBarrier ( 2 , 0 ) , createBarrier ( 3 , 0 ) , createBuffer ( 0 ) , createBuffer ( 0 ) , createBarrier ( 4 , 0 ) , createBarrier ( 5 , 0 ) , createBarrier ( 6 , 0 ) , createBuffer ( 0 ) , createEndOfPartition ( 0 ) } ; MockInputGate gate = new MockInputGate ( PAGE_SIZE , 1 , Arrays . asList ( sequence ) ) ; BarrierBuffer buffer = new BarrierBuffer ( gate , ioManager ) ; ValidatingCheckpointHandler handler = new ValidatingCheckpointHandler ( ) ; buffer . registerCheckpointEventHandler ( handler ) ; handler . setNextExpectedCheckpointId ( 1L ) ; for ( BufferOrEvent boe : sequence ) { if ( boe . isBuffer ( ) || boe . getEvent ( ) . getClass ( ) != CheckpointBarrier . class ) { assertEquals ( boe , buffer . getNextNonBlocked ( ) ) ; } } assertNull ( buffer . getNextNonBlocked ( ) ) ; assertNull ( buffer . getNextNonBlocked ( ) ) ; buffer . cleanup ( ) ; checkNoTempFilesRemain ( ) ; } catch ( Exception e ) { e . printStackTrace ( ) ; fail ( e . getMessage ( ) ) ; } }
Validates that the buffer preserved the order of elements for a
 input with a single input channel, and checkpoint events.
Validates that the buffer preserved the order of elements for a input with a single input channel and checkpoint events .