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) SlotManager accepts a slot request
	 * 2) SlotRequest leads to a container allocation
	 * 3) Slot becomes available and TaskExecutor gets a SlotRequest
	 */
Tests whether
1) SlotManager accepts a slot request
2) SlotRequest leads to a container allocation
3) Slot becomes available and TaskExecutor gets a SlotRequest
Tests whether 1 ) SlotManager accepts a slot request 2 ) SlotRequest leads to a container allocation 3 ) Slot becomes available and TaskExecutor gets a SlotRequest
@Test
	public void testSlotsUnavailableRequest() throws Exception {
		final JobID jobID = new JobID();
		final ResourceManagerId rmLeaderID = ResourceManagerId.generate();
		try (SlotManager slotManager = new SlotManager(
			scheduledExecutor,
			TestingUtils.infiniteTime(),
			TestingUtils.infiniteTime(),
			TestingUtils.infiniteTime())) {
			ResourceActions resourceManagerActions = mock(ResourceActions.class);
			slotManager.start(rmLeaderID, Executors.directExecutor(), resourceManagerActions);
			final AllocationID allocationID = new AllocationID();
			final ResourceProfile resourceProfile = new ResourceProfile(1.0, 100);
			final String targetAddress = "foobar";
			SlotRequest slotRequest = new SlotRequest(jobID, allocationID, resourceProfile, targetAddress);
			slotManager.registerSlotRequest(slotRequest);
			verify(resourceManagerActions).allocateResource(eq(slotRequest.getResourceProfile()));
			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));
			final ResourceID resourceID = ResourceID.generate();
			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);
			verify(taskExecutorGateway, timeout(5000L))
				.requestSlot(eq(slotID), eq(jobID), eq(allocationID), any(String.class), any(ResourceManagerId.class), any(Time.class));
		}
	}
@ Test public void testSlotsUnavailableRequest ( ) throws Exception { final JobID jobID = new JobID ( ) ; final ResourceManagerId rmLeaderID = ResourceManagerId . generate ( ) ; try ( SlotManager slotManager = new SlotManager ( scheduledExecutor , TestingUtils . infiniteTime ( ) , TestingUtils . infiniteTime ( ) , TestingUtils . infiniteTime ( ) ) ) { ResourceActions resourceManagerActions = mock ( ResourceActions . class ) ; slotManager . start ( rmLeaderID , Executors . directExecutor ( ) , resourceManagerActions ) ; final AllocationID allocationID = new AllocationID ( ) ; final ResourceProfile resourceProfile = new ResourceProfile ( 1.0 , 100 ) ; final String targetAddress = "foobar" ; SlotRequest slotRequest = new SlotRequest ( jobID , allocationID , resourceProfile , targetAddress ) ; slotManager . registerSlotRequest ( slotRequest ) ; verify ( resourceManagerActions ) . allocateResource ( eq ( slotRequest . getResourceProfile ( ) ) ) ; 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 ) ) ; final ResourceID resourceID = ResourceID . generate ( ) ; 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 ) ; verify ( taskExecutorGateway , timeout ( 5000L ) ) . requestSlot ( eq ( slotID ) , eq ( jobID ) , eq ( allocationID ) , any ( String . class ) , any ( ResourceManagerId . class ) , any ( Time . class ) ) ; } }
Tests whether
 1) SlotManager accepts a slot request
 2) SlotRequest leads to a container allocation
 3) Slot becomes available and TaskExecutor gets a SlotRequest
Tests whether 1 ) SlotManager accepts a slot request 2 ) SlotRequest leads to a container allocation 3 ) Slot becomes available and TaskExecutor gets a SlotRequest
// slot becomes available // register slot at SlotManager // 4) Slot becomes available and TaskExecutor gets a SlotRequest