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 the schedule or update consumers message sending behaviour depending on the relevant flags.
	 */
Tests the schedule or update consumers message sending behaviour depending on the relevant flags.
Tests the schedule or update consumers message sending behaviour depending on the relevant flags .
@Test
	public void testSendScheduleOrUpdateConsumersMessage() throws Exception {
		{
			ResultPartitionConsumableNotifier notifier = mock(ResultPartitionConsumableNotifier.class);
			ResultPartition partition = createPartition(notifier, ResultPartitionType.PIPELINED, true);
			partition.add(TestBufferFactory.createBuffer(), 0);
			verify(notifier, times(1))
				.notifyPartitionConsumable(
					eq(partition.getJobId()),
					eq(partition.getPartitionId()),
					any(TaskActions.class));
		}
		{
			ResultPartitionConsumableNotifier notifier = mock(ResultPartitionConsumableNotifier.class);
			ResultPartition partition = createPartition(notifier, ResultPartitionType.PIPELINED, false);
			partition.add(TestBufferFactory.createBuffer(), 0);
			verify(notifier, never()).notifyPartitionConsumable(any(JobID.class), any(ResultPartitionID.class), any(TaskActions.class));
		}
		{
			ResultPartitionConsumableNotifier notifier = mock(ResultPartitionConsumableNotifier.class);
			ResultPartition partition = createPartition(notifier, ResultPartitionType.BLOCKING, true);
			partition.add(TestBufferFactory.createBuffer(), 0);
			verify(notifier, never()).notifyPartitionConsumable(any(JobID.class), any(ResultPartitionID.class), any(TaskActions.class));
		}
		{
			ResultPartitionConsumableNotifier notifier = mock(ResultPartitionConsumableNotifier.class);
			ResultPartition partition = createPartition(notifier, ResultPartitionType.BLOCKING, false);
			partition.add(TestBufferFactory.createBuffer(), 0);
			verify(notifier, never()).notifyPartitionConsumable(any(JobID.class), any(ResultPartitionID.class), any(TaskActions.class));
		}
	}
@ Test public void testSendScheduleOrUpdateConsumersMessage ( ) throws Exception { { ResultPartitionConsumableNotifier notifier = mock ( ResultPartitionConsumableNotifier . class ) ; ResultPartition partition = createPartition ( notifier , ResultPartitionType . PIPELINED , true ) ; partition . add ( TestBufferFactory . createBuffer ( ) , 0 ) ; verify ( notifier , times ( 1 ) ) . notifyPartitionConsumable ( eq ( partition . getJobId ( ) ) , eq ( partition . getPartitionId ( ) ) , any ( TaskActions . class ) ) ; } { ResultPartitionConsumableNotifier notifier = mock ( ResultPartitionConsumableNotifier . class ) ; ResultPartition partition = createPartition ( notifier , ResultPartitionType . PIPELINED , false ) ; partition . add ( TestBufferFactory . createBuffer ( ) , 0 ) ; verify ( notifier , never ( ) ) . notifyPartitionConsumable ( any ( JobID . class ) , any ( ResultPartitionID . class ) , any ( TaskActions . class ) ) ; } { ResultPartitionConsumableNotifier notifier = mock ( ResultPartitionConsumableNotifier . class ) ; ResultPartition partition = createPartition ( notifier , ResultPartitionType . BLOCKING , true ) ; partition . add ( TestBufferFactory . createBuffer ( ) , 0 ) ; verify ( notifier , never ( ) ) . notifyPartitionConsumable ( any ( JobID . class ) , any ( ResultPartitionID . class ) , any ( TaskActions . class ) ) ; } { ResultPartitionConsumableNotifier notifier = mock ( ResultPartitionConsumableNotifier . class ) ; ResultPartition partition = createPartition ( notifier , ResultPartitionType . BLOCKING , false ) ; partition . add ( TestBufferFactory . createBuffer ( ) , 0 ) ; verify ( notifier , never ( ) ) . notifyPartitionConsumable ( any ( JobID . class ) , any ( ResultPartitionID . class ) , any ( TaskActions . class ) ) ; } }
Tests the schedule or update consumers message sending behaviour depending on the relevant flags.
Tests the schedule or update consumers message sending behaviour depending on the relevant flags .
// Pipelined, send message => notify // Pipelined, don't send message => don't notify // Blocking, send message => don't notify // Blocking, don't send message => don't notify