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 that a spilled partition is correctly read back in via a spilled
	 * read view.
	 */
Tests that a spilled partition is correctly read back in via a spilled
read view.
Tests that a spilled partition is correctly read back in via a spilled read view .
@Test
	public void testConsumeSpillablePartitionSpilledDuringConsume() throws Exception {
		SpillableSubpartition partition = createSubpartition();
		Buffer buffer = new Buffer(MemorySegmentFactory.allocateUnpooledSegment(4096), FreeingBufferRecycler.INSTANCE);
		buffer.retain();
		buffer.retain();
		partition.add(buffer);
		partition.add(buffer);
		partition.add(buffer);
		partition.finish();
		assertEquals(4, partition.getTotalNumberOfBuffers());
		assertEquals(4096 * 3 + 4, partition.getTotalNumberOfBytes());
		AwaitableBufferAvailablityListener listener = new AwaitableBufferAvailablityListener();
		SpillableSubpartitionView reader = (SpillableSubpartitionView) partition.createReadView(listener);
		assertEquals(1, listener.getNumNotifiedBuffers());
		assertFalse(buffer.isRecycled());
		Buffer read = reader.getNextBuffer();
		assertSame(buffer, read);
		read.recycle();
		assertEquals(2, listener.getNumNotifiedBuffers());
		assertFalse(buffer.isRecycled());
		assertEquals(2, partition.releaseMemory());
		assertFalse(buffer.isRecycled()); 
		assertEquals(4, partition.getTotalNumberOfBuffers());
		assertEquals(4096 * 3 + 4, partition.getTotalNumberOfBytes());
		listener.awaitNotifications(4, 30_000);
		assertEquals(4, listener.getNumNotifiedBuffers());
		read = reader.getNextBuffer();
		assertSame(buffer, read);
		read.recycle();
		read = reader.getNextBuffer();
		assertNotNull(read);
		assertNotSame(buffer, read);
		assertFalse(read.isRecycled());
		read.recycle();
		assertTrue(read.isRecycled());
		read = reader.getNextBuffer();
		assertNotNull(read);
		assertEquals(EndOfPartitionEvent.class, EventSerializer.fromBuffer(read, ClassLoader.getSystemClassLoader()).getClass());
		assertFalse(read.isRecycled());
		read.recycle();
		assertTrue(read.isRecycled());
		final long deadline = System.currentTimeMillis() + 30_000L; 
		while (!buffer.isRecycled() && System.currentTimeMillis() < deadline) {
			Thread.sleep(1);
		}
		assertTrue(buffer.isRecycled());
	}
@ Test public void testConsumeSpillablePartitionSpilledDuringConsume ( ) throws Exception { SpillableSubpartition partition = createSubpartition ( ) ; Buffer buffer = new Buffer ( MemorySegmentFactory . allocateUnpooledSegment ( 4096 ) , FreeingBufferRecycler . INSTANCE ) ; buffer . retain ( ) ; buffer . retain ( ) ; partition . add ( buffer ) ; partition . add ( buffer ) ; partition . add ( buffer ) ; partition . finish ( ) ; assertEquals ( 4 , partition . getTotalNumberOfBuffers ( ) ) ; assertEquals ( 4096 * 3 + 4 , partition . getTotalNumberOfBytes ( ) ) ; AwaitableBufferAvailablityListener listener = new AwaitableBufferAvailablityListener ( ) ; SpillableSubpartitionView reader = ( SpillableSubpartitionView ) partition . createReadView ( listener ) ; assertEquals ( 1 , listener . getNumNotifiedBuffers ( ) ) ; assertFalse ( buffer . isRecycled ( ) ) ; Buffer read = reader . getNextBuffer ( ) ; assertSame ( buffer , read ) ; read . recycle ( ) ; assertEquals ( 2 , listener . getNumNotifiedBuffers ( ) ) ; assertFalse ( buffer . isRecycled ( ) ) ; assertEquals ( 2 , partition . releaseMemory ( ) ) ; assertFalse ( buffer . isRecycled ( ) ) ; assertEquals ( 4 , partition . getTotalNumberOfBuffers ( ) ) ; assertEquals ( 4096 * 3 + 4 , partition . getTotalNumberOfBytes ( ) ) ; listener . awaitNotifications ( 4 , 30_000 ) ; assertEquals ( 4 , listener . getNumNotifiedBuffers ( ) ) ; read = reader . getNextBuffer ( ) ; assertSame ( buffer , read ) ; read . recycle ( ) ; read = reader . getNextBuffer ( ) ; assertNotNull ( read ) ; assertNotSame ( buffer , read ) ; assertFalse ( read . isRecycled ( ) ) ; read . recycle ( ) ; assertTrue ( read . isRecycled ( ) ) ; read = reader . getNextBuffer ( ) ; assertNotNull ( read ) ; assertEquals ( EndOfPartitionEvent . class , EventSerializer . fromBuffer ( read , ClassLoader . getSystemClassLoader ( ) ) . getClass ( ) ) ; assertFalse ( read . isRecycled ( ) ) ; read . recycle ( ) ; assertTrue ( read . isRecycled ( ) ) ; final long deadline = System . currentTimeMillis ( ) + 30_000L ; while ( ! buffer . isRecycled ( ) && System . currentTimeMillis ( ) < deadline ) { Thread . sleep ( 1 ) ; } assertTrue ( buffer . isRecycled ( ) ) ; }
Tests that a spilled partition is correctly read back in via a spilled
 read view.
Tests that a spilled partition is correctly read back in via a spilled read view .
// Initial notification // Spill now // still one in the reader! // still same statistics: // now the buffer may be freed, depending on the timing of the write operation // -> let's do this check at the end of the test (to save some time) // End of partition // finally check that the buffer has been freed after a successful (or failed) write // 30 secs