flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManagerTest.java
/**
	 * Tests that a slot request which can be fulfilled will trigger a slot allocation.
	 */
Tests that a slot request which can be fulfilled will trigger a slot allocation.
Tests that a slot request which can be fulfilled will trigger a slot allocation .
@Test
	public void testSlotRequestWithFreeSlot() throws Exception {
		final ResourceManagerId resourceManagerId = ResourceManagerId.generate();
		final ResourceID resourceID = ResourceID.generate();
		final JobID jobId = new JobID();
		final SlotID slotId = new SlotID(resourceID, 0);
		final String targetAddress = "localhost";
		final AllocationID allocationId = new AllocationID();
		final ResourceProfile resourceProfile = new ResourceProfile(42.0, 1337);
		final SlotRequest slotRequest = new SlotRequest(
			jobId,
			allocationId,
			resourceProfile,
			targetAddress);
		ResourceActions resourceManagerActions = mock(ResourceActions.class);
		try (SlotManager slotManager = createSlotManager(resourceManagerId, resourceManagerActions)) {
			final TaskExecutorGateway taskExecutorGateway = mock(TaskExecutorGateway.class);
			when(taskExecutorGateway.requestSlot(
				eq(slotId),
				eq(jobId),
				eq(allocationId),
				anyString(),
				eq(resourceManagerId),
				any(Time.class))).thenReturn(CompletableFuture.completedFuture(Acknowledge.get()));
			final TaskExecutorConnection taskExecutorConnection = new TaskExecutorConnection(taskExecutorGateway);
			final SlotStatus slotStatus = new SlotStatus(slotId, resourceProfile);
			final SlotReport slotReport = new SlotReport(slotStatus);
			slotManager.registerTaskManager(
				taskExecutorConnection,
				slotReport);
			assertTrue("The slot request should be accepted", slotManager.registerSlotRequest(slotRequest));
			verify(taskExecutorGateway).requestSlot(eq(slotId), eq(jobId), eq(allocationId), eq(targetAddress), eq(resourceManagerId), any(Time.class));
			TaskManagerSlot slot = slotManager.getSlot(slotId);
			assertEquals("The slot has not been allocated to the expected allocation id.", allocationId, slot.getAllocationId());
		}
	}
@ Test public void testSlotRequestWithFreeSlot ( ) throws Exception { final ResourceManagerId resourceManagerId = ResourceManagerId . generate ( ) ; final ResourceID resourceID = ResourceID . generate ( ) ; final JobID jobId = new JobID ( ) ; final SlotID slotId = new SlotID ( resourceID , 0 ) ; final String targetAddress = "localhost" ; final AllocationID allocationId = new AllocationID ( ) ; final ResourceProfile resourceProfile = new ResourceProfile ( 42.0 , 1337 ) ; final SlotRequest slotRequest = new SlotRequest ( jobId , allocationId , resourceProfile , targetAddress ) ; ResourceActions resourceManagerActions = mock ( ResourceActions . class ) ; try ( SlotManager slotManager = createSlotManager ( resourceManagerId , resourceManagerActions ) ) { final TaskExecutorGateway taskExecutorGateway = mock ( TaskExecutorGateway . class ) ; when ( taskExecutorGateway . requestSlot ( eq ( slotId ) , eq ( jobId ) , eq ( allocationId ) , anyString ( ) , eq ( resourceManagerId ) , any ( Time . class ) ) ) . thenReturn ( CompletableFuture . completedFuture ( Acknowledge . get ( ) ) ) ; final TaskExecutorConnection taskExecutorConnection = new TaskExecutorConnection ( taskExecutorGateway ) ; final SlotStatus slotStatus = new SlotStatus ( slotId , resourceProfile ) ; final SlotReport slotReport = new SlotReport ( slotStatus ) ; slotManager . registerTaskManager ( taskExecutorConnection , slotReport ) ; assertTrue ( "The slot request should be accepted" , slotManager . registerSlotRequest ( slotRequest ) ) ; verify ( taskExecutorGateway ) . requestSlot ( eq ( slotId ) , eq ( jobId ) , eq ( allocationId ) , eq ( targetAddress ) , eq ( resourceManagerId ) , any ( Time . class ) ) ; TaskManagerSlot slot = slotManager . getSlot ( slotId ) ; assertEquals ( "The slot has not been allocated to the expected allocation id." , allocationId , slot . getAllocationId ( ) ) ; } }
Tests that a slot request which can be fulfilled will trigger a slot allocation.
Tests that a slot request which can be fulfilled will trigger a slot allocation .
// accept an incoming slot request