flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartitionTest.java
/**
	 * Tests {@link SpillableSubpartition#releaseMemory()} with a spillable partition which has a a
	 * writer that does not do any write to check for correct buffer recycling.
	 */
Tests  SpillableSubpartition#releaseMemory() with a spillable partition which has a a
writer that does not do any write to check for correct buffer recycling.
Tests SpillableSubpartition#releaseMemory () with a spillable partition which has a a writer that does not do any write to check for correct buffer recycling .
private void testReleaseOnSpillablePartitionWithSlowWriter(boolean createView) throws Exception {
		IOManager ioManager = new IOManagerAsyncWithNoOpBufferFileWriter();
		SpillableSubpartition partition = createSubpartition(ioManager);
		Buffer buffer1 = new Buffer(MemorySegmentFactory.allocateUnpooledSegment(4096),
			FreeingBufferRecycler.INSTANCE);
		Buffer buffer2 = new Buffer(MemorySegmentFactory.allocateUnpooledSegment(4096),
			FreeingBufferRecycler.INSTANCE);
		try {
			partition.add(buffer1);
			partition.add(buffer2);
			assertFalse("buffer1 should not be recycled (still in the queue)", buffer1.isRecycled());
			assertFalse("buffer2 should not be recycled (still in the queue)", buffer2.isRecycled());
			assertEquals(2, partition.getTotalNumberOfBuffers());
			assertEquals(4096 * 2, partition.getTotalNumberOfBytes());
			if (createView) {
				partition.finish();
				partition.createReadView(numBuffers -> {});
			}
			assertEquals(2, partition.releaseMemory());
			assertFalse("buffer1 should not be recycled (advertised as nextBuffer)", buffer1.isRecycled());
			assertFalse("buffer2 should not be recycled (not written yet)", buffer2.isRecycled());
		} finally {
			ioManager.shutdown();
			if (!buffer1.isRecycled()) {
				buffer1.recycle();
			}
			if (!buffer2.isRecycled()) {
				buffer2.recycle();
			}
			assertEquals(2 + (createView ? 1 : 0), partition.getTotalNumberOfBuffers());
			assertEquals(4096 * 2 + (createView ? 4 : 0), partition.getTotalNumberOfBytes());
		}
	}
private void testReleaseOnSpillablePartitionWithSlowWriter ( boolean createView ) throws Exception { IOManager ioManager = new IOManagerAsyncWithNoOpBufferFileWriter ( ) ; SpillableSubpartition partition = createSubpartition ( ioManager ) ; Buffer buffer1 = new Buffer ( MemorySegmentFactory . allocateUnpooledSegment ( 4096 ) , FreeingBufferRecycler . INSTANCE ) ; Buffer buffer2 = new Buffer ( MemorySegmentFactory . allocateUnpooledSegment ( 4096 ) , FreeingBufferRecycler . INSTANCE ) ; try { partition . add ( buffer1 ) ; partition . add ( buffer2 ) ; assertFalse ( "buffer1 should not be recycled (still in the queue)" , buffer1 . isRecycled ( ) ) ; assertFalse ( "buffer2 should not be recycled (still in the queue)" , buffer2 . isRecycled ( ) ) ; assertEquals ( 2 , partition . getTotalNumberOfBuffers ( ) ) ; assertEquals ( 4096 * 2 , partition . getTotalNumberOfBytes ( ) ) ; if ( createView ) { partition . finish ( ) ; partition . createReadView ( numBuffers -> { } ) ; } assertEquals ( 2 , partition . releaseMemory ( ) ) ; assertFalse ( "buffer1 should not be recycled (advertised as nextBuffer)" , buffer1 . isRecycled ( ) ) ; assertFalse ( "buffer2 should not be recycled (not written yet)" , buffer2 . isRecycled ( ) ) ; } finally { ioManager . shutdown ( ) ; if ( ! buffer1 . isRecycled ( ) ) { buffer1 . recycle ( ) ; } if ( ! buffer2 . isRecycled ( ) ) { buffer2 . recycle ( ) ; } assertEquals ( 2 + ( createView ? 1 : 0 ) , partition . getTotalNumberOfBuffers ( ) ) ; assertEquals ( 4096 * 2 + ( createView ? 4 : 0 ) , partition . getTotalNumberOfBytes ( ) ) ; } }
Tests {@link SpillableSubpartition#releaseMemory()} with a spillable partition which has a a
 writer that does not do any write to check for correct buffer recycling.
Tests { @link SpillableSubpartition#releaseMemory () } with a spillable partition which has a a writer that does not do any write to check for correct buffer recycling .
// simulate slow writer by a no-op write operation // we need two buffers because the view will use one of them and not release it // Create a read view // one instance of the buffers is placed in the view's nextBuffer and not released // (if there is no view, there will be no additional EndOfPartitionEvent) // note: a view requires a finished partition which has an additional EndOfPartitionEvent