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 behaves correctly if no checkpoint barriers come,
	 * for an input with multiple input channels.
	 */
Validates that the buffer behaves correctly if no checkpoint barriers come,
for an input with multiple input channels.
Validates that the buffer behaves correctly if no checkpoint barriers come for an input with multiple input channels .
@Test
	public void testMultiChannelNoBarriers() {
		try {
			BufferOrEvent[] sequence = { createBuffer(2), createBuffer(2), createBuffer(0),
					createBuffer(1), createBuffer(0), createEndOfPartition(0),
					createBuffer(3), createBuffer(1), createEndOfPartition(3),
					createBuffer(1), createEndOfPartition(1), createBuffer(2), createEndOfPartition(2)
			};
			MockInputGate gate = new MockInputGate(PAGE_SIZE, 4, Arrays.asList(sequence));
			BarrierBuffer buffer = new BarrierBuffer(gate, ioManager);
			for (BufferOrEvent boe : sequence) {
				assertEquals(boe, buffer.getNextNonBlocked());
			}
			assertEquals(0L, buffer.getAlignmentDurationNanos());
			assertNull(buffer.getNextNonBlocked());
			assertNull(buffer.getNextNonBlocked());
			buffer.cleanup();
			checkNoTempFilesRemain();
		}
		catch (Exception e) {
			e.printStackTrace();
			fail(e.getMessage());
		}
	}
@ Test public void testMultiChannelNoBarriers ( ) { try { BufferOrEvent [ ] sequence = { createBuffer ( 2 ) , createBuffer ( 2 ) , createBuffer ( 0 ) , createBuffer ( 1 ) , createBuffer ( 0 ) , createEndOfPartition ( 0 ) , createBuffer ( 3 ) , createBuffer ( 1 ) , createEndOfPartition ( 3 ) , createBuffer ( 1 ) , createEndOfPartition ( 1 ) , createBuffer ( 2 ) , createEndOfPartition ( 2 ) } ; MockInputGate gate = new MockInputGate ( PAGE_SIZE , 4 , Arrays . asList ( sequence ) ) ; BarrierBuffer buffer = new BarrierBuffer ( gate , ioManager ) ; for ( BufferOrEvent boe : sequence ) { assertEquals ( boe , buffer . getNextNonBlocked ( ) ) ; } assertEquals ( 0L , buffer . getAlignmentDurationNanos ( ) ) ; assertNull ( buffer . getNextNonBlocked ( ) ) ; assertNull ( buffer . getNextNonBlocked ( ) ) ; buffer . cleanup ( ) ; checkNoTempFilesRemain ( ) ; } catch ( Exception e ) { e . printStackTrace ( ) ; fail ( e . getMessage ( ) ) ; } }
Validates that the buffer behaves correctly if no checkpoint barriers come,
 for an input with multiple input channels.
Validates that the buffer behaves correctly if no checkpoint barriers come for an input with multiple input channels .