flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionTest.java
/**
	 * Tests cleanup of {@link PipelinedSubpartition#release()}.
	 *
	 * @param createView
	 * 		whether the partition should have a view attached to it (<tt>true</tt>) or not (<tt>false</tt>)
	 */
Tests cleanup of  PipelinedSubpartition#release().
@param createView
whether the partition should have a view attached to it (true) or not (false)
Tests cleanup of PipelinedSubpartition#release () . @param createView whether the partition should have a view attached to it ( true ) or not ( false )
private void testCleanupReleasedPartition(boolean createView) throws Exception {
		PipelinedSubpartition partition = createSubpartition();
		Buffer buffer1 = new Buffer(MemorySegmentFactory.allocateUnpooledSegment(4096),
			FreeingBufferRecycler.INSTANCE);
		Buffer buffer2 = new Buffer(MemorySegmentFactory.allocateUnpooledSegment(4096),
			FreeingBufferRecycler.INSTANCE);
		boolean buffer1Recycled;
		boolean buffer2Recycled;
		try {
			partition.add(buffer1);
			partition.add(buffer2);
			ResultSubpartitionView view = null;
			if (createView) {
				view = partition.createReadView(numBuffers -> {});
			}
			partition.release();
			assertTrue(partition.isReleased());
			if (createView) {
				assertTrue(view.isReleased());
			}
			assertTrue(buffer1.isRecycled());
		} finally {
			buffer1Recycled = buffer1.isRecycled();
			if (!buffer1Recycled) {
				buffer1.recycle();
			}
			buffer2Recycled = buffer2.isRecycled();
			if (!buffer2Recycled) {
				buffer2.recycle();
			}
		}
		if (!buffer1Recycled) {
			Assert.fail("buffer 1 not recycled");
		}
		if (!buffer2Recycled) {
			Assert.fail("buffer 2 not recycled");
		}
		assertEquals(2, partition.getTotalNumberOfBuffers());
		assertEquals(2 * 4096, partition.getTotalNumberOfBytes());
	}
private void testCleanupReleasedPartition ( boolean createView ) throws Exception { PipelinedSubpartition partition = createSubpartition ( ) ; Buffer buffer1 = new Buffer ( MemorySegmentFactory . allocateUnpooledSegment ( 4096 ) , FreeingBufferRecycler . INSTANCE ) ; Buffer buffer2 = new Buffer ( MemorySegmentFactory . allocateUnpooledSegment ( 4096 ) , FreeingBufferRecycler . INSTANCE ) ; boolean buffer1Recycled ; boolean buffer2Recycled ; try { partition . add ( buffer1 ) ; partition . add ( buffer2 ) ; ResultSubpartitionView view = null ; if ( createView ) { view = partition . createReadView ( numBuffers -> { } ) ; } partition . release ( ) ; assertTrue ( partition . isReleased ( ) ) ; if ( createView ) { assertTrue ( view . isReleased ( ) ) ; } assertTrue ( buffer1 . isRecycled ( ) ) ; } finally { buffer1Recycled = buffer1 . isRecycled ( ) ; if ( ! buffer1Recycled ) { buffer1 . recycle ( ) ; } buffer2Recycled = buffer2 . isRecycled ( ) ; if ( ! buffer2Recycled ) { buffer2 . recycle ( ) ; } } if ( ! buffer1Recycled ) { Assert . fail ( "buffer 1 not recycled" ) ; } if ( ! buffer2Recycled ) { Assert . fail ( "buffer 2 not recycled" ) ; } assertEquals ( 2 , partition . getTotalNumberOfBuffers ( ) ) ; assertEquals ( 2 * 4096 , partition . getTotalNumberOfBytes ( ) ) ; }
Tests cleanup of {@link PipelinedSubpartition#release()}.
Tests cleanup of { @link PipelinedSubpartition#release () } .
// create the read view first