flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionTest.java
/**
	 * Tests {@link ResultPartition#add(Buffer, int)} on a working partition.
	 *
	 * @param pipelined the result partition type to set up
	 */
Tests  ResultPartition#add(Buffer, int) on a working partition.
@param pipelined the result partition type to set up
Tests ResultPartition#add ( Buffer int ) on a working partition . @param pipelined the result partition type to set up
protected void testAddOnPartition(final ResultPartitionType pipelined)
		throws Exception {
		ResultPartitionConsumableNotifier notifier = mock(ResultPartitionConsumableNotifier.class);
		ResultPartition partition = createPartition(notifier, pipelined, true);
		Buffer buffer = TestBufferFactory.createBuffer();
		try {
			partition.add(buffer, 0);
			assertFalse("buffer should not be recycled (still in the queue)", buffer.isRecycled());
		} finally {
			if (!buffer.isRecycled()) {
				buffer.recycle();
			}
			if (pipelined.isPipelined()) {
				verify(notifier, times(1))
					.notifyPartitionConsumable(
						eq(partition.getJobId()),
						eq(partition.getPartitionId()),
						any(TaskActions.class));
			}
		}
	}
protected void testAddOnPartition ( final ResultPartitionType pipelined ) throws Exception { ResultPartitionConsumableNotifier notifier = mock ( ResultPartitionConsumableNotifier . class ) ; ResultPartition partition = createPartition ( notifier , pipelined , true ) ; Buffer buffer = TestBufferFactory . createBuffer ( ) ; try { partition . add ( buffer , 0 ) ; assertFalse ( "buffer should not be recycled (still in the queue)" , buffer . isRecycled ( ) ) ; } finally { if ( ! buffer . isRecycled ( ) ) { buffer . recycle ( ) ; } if ( pipelined . isPipelined ( ) ) { verify ( notifier , times ( 1 ) ) . notifyPartitionConsumable ( eq ( partition . getJobId ( ) ) , eq ( partition . getPartitionId ( ) ) , any ( TaskActions . class ) ) ; } } }
Tests {@link ResultPartition#add(Buffer, int)} on a working partition.
Tests { @link ResultPartition#add ( Buffer int ) } on a working partition .
// partition.add() adds the buffer without recycling it (if not spilling) // should have been notified for pipelined partitions