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 duplicate slot requests (requests with an already registered allocation id) are
	 * also detected after a pending slot request has been fulfilled but not yet freed.
	 */
Tests that duplicate slot requests (requests with an already registered allocation id) are
also detected after a pending slot request has been fulfilled but not yet freed.
Tests that duplicate slot requests ( requests with an already registered allocation id ) are also detected after a pending slot request has been fulfilled but not yet freed .
@Test
	public void testDuplicatePendingSlotRequestAfterSuccessfulAllocation() 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, resourceProfile1);
		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());
			assertFalse(slotManager.registerSlotRequest(slotRequest2));
		}
		verify(resourceManagerActions, never()).allocateResource(any(ResourceProfile.class));
	}
@ Test public void testDuplicatePendingSlotRequestAfterSuccessfulAllocation ( ) 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 , resourceProfile1 ) ; 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 ( ) ) ; assertFalse ( slotManager . registerSlotRequest ( slotRequest2 ) ) ; } verify ( resourceManagerActions , never ( ) ) . allocateResource ( any ( ResourceProfile . class ) ) ; }
Tests that duplicate slot requests (requests with an already registered allocation id) are
 also detected after a pending slot request has been fulfilled but not yet freed.
Tests that duplicate slot requests ( requests with an already registered allocation id ) are also detected after a pending slot request has been fulfilled but not yet freed .
// check that we have only called the resource allocation only for the first slot request, // since the second request is a duplicate