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 an already registered allocation id can be reused after the initial slot request
	 * has been freed.
	 */
Tests that an already registered allocation id can be reused after the initial slot request
has been freed.
Tests that an already registered allocation id can be reused after the initial slot request has been freed .
@Test
	public void testAcceptingDuplicateSlotRequestAfterAllocationRelease() throws Exception {
		final ResourceManagerId resourceManagerId = ResourceManagerId.generate();
		final ResourceActions resourceManagerActions = mock(ResourceActions.class);
		final AllocationID allocationId = new AllocationID();
		final ResourceProfile resourceProfile1 = new ResourceProfile(1.0, 2);
		final ResourceProfile resourceProfile2 = new ResourceProfile(2.0, 1);
		final SlotRequest slotRequest1 = new SlotRequest(new JobID(), allocationId, resourceProfile1, "foobar");
		final SlotRequest slotRequest2 = new SlotRequest(new JobID(), allocationId, resourceProfile2, "barfoo");
		final TaskExecutorGateway taskExecutorGateway = mock(TaskExecutorGateway.class);
		when(taskExecutorGateway.requestSlot(
			any(SlotID.class),
			any(JobID.class),
			any(AllocationID.class),
			anyString(),
			eq(resourceManagerId),
			any(Time.class))).thenReturn(CompletableFuture.completedFuture(Acknowledge.get()));
		final TaskExecutorConnection taskManagerConnection = new TaskExecutorConnection(taskExecutorGateway);
		final SlotID slotId = new SlotID(ResourceID.generate(), 0);
		final SlotStatus slotStatus = new SlotStatus(slotId, new ResourceProfile(2.0, 2));
		final SlotReport slotReport = new SlotReport(slotStatus);
		try (SlotManager slotManager = createSlotManager(resourceManagerId, resourceManagerActions)) {
			slotManager.registerTaskManager(taskManagerConnection, slotReport);
			assertTrue(slotManager.registerSlotRequest(slotRequest1));
			TaskManagerSlot slot = slotManager.getSlot(slotId);
			assertEquals("The slot has not been allocated to the expected allocation id.", allocationId, slot.getAllocationId());
			slotManager.freeSlot(slotId, allocationId);
			assertTrue(slot.getState() == TaskManagerSlot.State.FREE);
			assertNull(slot.getAllocationId());
			assertTrue(slotManager.registerSlotRequest(slotRequest2));
			assertEquals("The slot has not been allocated to the expected allocation id.", allocationId, slot.getAllocationId());
		}
		verify(resourceManagerActions, never()).allocateResource(any(ResourceProfile.class));
	}
@ Test public void testAcceptingDuplicateSlotRequestAfterAllocationRelease ( ) throws Exception { final ResourceManagerId resourceManagerId = ResourceManagerId . generate ( ) ; final ResourceActions resourceManagerActions = mock ( ResourceActions . class ) ; final AllocationID allocationId = new AllocationID ( ) ; final ResourceProfile resourceProfile1 = new ResourceProfile ( 1.0 , 2 ) ; final ResourceProfile resourceProfile2 = new ResourceProfile ( 2.0 , 1 ) ; final SlotRequest slotRequest1 = new SlotRequest ( new JobID ( ) , allocationId , resourceProfile1 , "foobar" ) ; final SlotRequest slotRequest2 = new SlotRequest ( new JobID ( ) , allocationId , resourceProfile2 , "barfoo" ) ; final TaskExecutorGateway taskExecutorGateway = mock ( TaskExecutorGateway . class ) ; when ( taskExecutorGateway . requestSlot ( any ( SlotID . class ) , any ( JobID . class ) , any ( AllocationID . class ) , anyString ( ) , eq ( resourceManagerId ) , any ( Time . class ) ) ) . thenReturn ( CompletableFuture . completedFuture ( Acknowledge . get ( ) ) ) ; final TaskExecutorConnection taskManagerConnection = new TaskExecutorConnection ( taskExecutorGateway ) ; final SlotID slotId = new SlotID ( ResourceID . generate ( ) , 0 ) ; final SlotStatus slotStatus = new SlotStatus ( slotId , new ResourceProfile ( 2.0 , 2 ) ) ; final SlotReport slotReport = new SlotReport ( slotStatus ) ; try ( SlotManager slotManager = createSlotManager ( resourceManagerId , resourceManagerActions ) ) { slotManager . registerTaskManager ( taskManagerConnection , slotReport ) ; assertTrue ( slotManager . registerSlotRequest ( slotRequest1 ) ) ; TaskManagerSlot slot = slotManager . getSlot ( slotId ) ; assertEquals ( "The slot has not been allocated to the expected allocation id." , allocationId , slot . getAllocationId ( ) ) ; slotManager . freeSlot ( slotId , allocationId ) ; assertTrue ( slot . getState ( ) == TaskManagerSlot . State . FREE ) ; assertNull ( slot . getAllocationId ( ) ) ; assertTrue ( slotManager . registerSlotRequest ( slotRequest2 ) ) ; assertEquals ( "The slot has not been allocated to the expected allocation id." , allocationId , slot . getAllocationId ( ) ) ; } verify ( resourceManagerActions , never ( ) ) . allocateResource ( any ( ResourceProfile . class ) ) ; }
Tests that an already registered allocation id can be reused after the initial slot request
 has been freed.
Tests that an already registered allocation id can be reused after the initial slot request has been freed .
// check that the slot has been freed // check that we have only called the resource allocation only for the first slot request, // since the second request is a duplicate