flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotProtocolTest.java
/**
	 * Tests whether
	 * 1) a SlotRequest is routed to the SlotManager
	 * 2) a SlotRequest is confirmed
	 * 3) a SlotRequest leads to an allocation of a registered slot
	 * 4) a SlotRequest is routed to the TaskExecutor
	 */
Tests whether
1) a SlotRequest is routed to the SlotManager
2) a SlotRequest is confirmed
3) a SlotRequest leads to an allocation of a registered slot
4) a SlotRequest is routed to the TaskExecutor
Tests whether 1 ) a SlotRequest is routed to the SlotManager 2 ) a SlotRequest is confirmed 3 ) a SlotRequest leads to an allocation of a registered slot 4 ) a SlotRequest is routed to the TaskExecutor
@Test
	public void testSlotAvailableRequest() throws Exception {
		final JobID jobID = new JobID();
		final ResourceManagerId rmLeaderID = ResourceManagerId.generate();
		TaskExecutorGateway taskExecutorGateway = mock(TaskExecutorGateway.class);
		Mockito.when(
			taskExecutorGateway
				.requestSlot(any(SlotID.class), any(JobID.class), any(AllocationID.class), any(String.class), any(ResourceManagerId.class), any(Time.class)))
			.thenReturn(mock(CompletableFuture.class));
		try (SlotManager slotManager = new SlotManager(
			scheduledExecutor,
			TestingUtils.infiniteTime(),
			TestingUtils.infiniteTime(),
			TestingUtils.infiniteTime())) {
			ResourceActions resourceManagerActions = mock(ResourceActions.class);
			slotManager.start(rmLeaderID, Executors.directExecutor(), resourceManagerActions);
			final ResourceID resourceID = ResourceID.generate();
			final AllocationID allocationID = new AllocationID();
			final ResourceProfile resourceProfile = new ResourceProfile(1.0, 100);
			final SlotID slotID = new SlotID(resourceID, 0);
			final SlotStatus slotStatus =
				new SlotStatus(slotID, resourceProfile);
			final SlotReport slotReport =
				new SlotReport(Collections.singletonList(slotStatus));
			slotManager.registerTaskManager(
				new TaskExecutorConnection(taskExecutorGateway), slotReport);
			final String targetAddress = "foobar";
			SlotRequest slotRequest = new SlotRequest(jobID, allocationID, resourceProfile, targetAddress);
			slotManager.registerSlotRequest(slotRequest);
			verify(taskExecutorGateway, timeout(5000))
				.requestSlot(eq(slotID), eq(jobID), eq(allocationID), any(String.class), any(ResourceManagerId.class), any(Time.class));
		}
	}
@ Test public void testSlotAvailableRequest ( ) throws Exception { final JobID jobID = new JobID ( ) ; final ResourceManagerId rmLeaderID = ResourceManagerId . generate ( ) ; TaskExecutorGateway taskExecutorGateway = mock ( TaskExecutorGateway . class ) ; Mockito . when ( taskExecutorGateway . requestSlot ( any ( SlotID . class ) , any ( JobID . class ) , any ( AllocationID . class ) , any ( String . class ) , any ( ResourceManagerId . class ) , any ( Time . class ) ) ) . thenReturn ( mock ( CompletableFuture . class ) ) ; try ( SlotManager slotManager = new SlotManager ( scheduledExecutor , TestingUtils . infiniteTime ( ) , TestingUtils . infiniteTime ( ) , TestingUtils . infiniteTime ( ) ) ) { ResourceActions resourceManagerActions = mock ( ResourceActions . class ) ; slotManager . start ( rmLeaderID , Executors . directExecutor ( ) , resourceManagerActions ) ; final ResourceID resourceID = ResourceID . generate ( ) ; final AllocationID allocationID = new AllocationID ( ) ; final ResourceProfile resourceProfile = new ResourceProfile ( 1.0 , 100 ) ; final SlotID slotID = new SlotID ( resourceID , 0 ) ; final SlotStatus slotStatus = new SlotStatus ( slotID , resourceProfile ) ; final SlotReport slotReport = new SlotReport ( Collections . singletonList ( slotStatus ) ) ; slotManager . registerTaskManager ( new TaskExecutorConnection ( taskExecutorGateway ) , slotReport ) ; final String targetAddress = "foobar" ; SlotRequest slotRequest = new SlotRequest ( jobID , allocationID , resourceProfile , targetAddress ) ; slotManager . registerSlotRequest ( slotRequest ) ; verify ( taskExecutorGateway , timeout ( 5000 ) ) . requestSlot ( eq ( slotID ) , eq ( jobID ) , eq ( allocationID ) , any ( String . class ) , any ( ResourceManagerId . class ) , any ( Time . class ) ) ; } }
Tests whether
 1) a SlotRequest is routed to the SlotManager
 2) a SlotRequest is confirmed
 3) a SlotRequest leads to an allocation of a registered slot
 4) a SlotRequest is routed to the TaskExecutor
Tests whether 1 ) a SlotRequest is routed to the SlotManager 2 ) a SlotRequest is confirmed 3 ) a SlotRequest leads to an allocation of a registered slot 4 ) a SlotRequest is routed to the TaskExecutor
// register slot at SlotManager // a SlotRequest is routed to the TaskExecutor