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 un-registration of task managers will free and remove all registered slots.
	 */
Tests that un-registration of task managers will free and remove all registered slots.
Tests that un - registration of task managers will free and remove all registered slots .
@Test
	public void testTaskManagerUnregistration() throws Exception {
		final ResourceManagerId resourceManagerId = ResourceManagerId.generate();
		final ResourceActions resourceManagerActions = mock(ResourceActions.class);
		final JobID jobId = new JobID();
		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(new CompletableFuture<>());
		final TaskExecutorConnection taskManagerConnection = new TaskExecutorConnection(taskExecutorGateway);
		ResourceID resourceId = ResourceID.generate();
		final SlotID slotId1 = new SlotID(resourceId, 0);
		final SlotID slotId2 = new SlotID(resourceId, 1);
		final AllocationID allocationId1 = new AllocationID();
		final AllocationID allocationId2 = new AllocationID();
		final ResourceProfile resourceProfile = new ResourceProfile(42.0, 1337);
		final SlotStatus slotStatus1 = new SlotStatus(slotId1, resourceProfile, jobId, allocationId1);
		final SlotStatus slotStatus2 = new SlotStatus(slotId2, resourceProfile);
		final SlotReport slotReport = new SlotReport(Arrays.asList(slotStatus1, slotStatus2));
		final SlotRequest slotRequest = new SlotRequest(
			new JobID(),
			allocationId2,
			resourceProfile,
			"foobar");
		try (SlotManager slotManager = createSlotManager(resourceManagerId, resourceManagerActions)) {
			slotManager.registerTaskManager(taskManagerConnection, slotReport);
			assertTrue("The number registered slots does not equal the expected number.",2 == slotManager.getNumberRegisteredSlots());
			TaskManagerSlot slot1 = slotManager.getSlot(slotId1);
			TaskManagerSlot slot2 = slotManager.getSlot(slotId2);
			assertTrue(slot1.getState() == TaskManagerSlot.State.ALLOCATED);
			assertTrue(slot2.getState() == TaskManagerSlot.State.FREE);
			assertTrue(slotManager.registerSlotRequest(slotRequest));
			assertFalse(slot2.getState() == TaskManagerSlot.State.FREE);
			assertTrue(slot2.getState() == TaskManagerSlot.State.PENDING);
			PendingSlotRequest pendingSlotRequest = slotManager.getSlotRequest(allocationId2);
			assertTrue("The pending slot request should have been assigned to slot 2", pendingSlotRequest.isAssigned());
			slotManager.unregisterTaskManager(taskManagerConnection.getInstanceID());
			assertTrue(0 == slotManager.getNumberRegisteredSlots());
			assertFalse(pendingSlotRequest.isAssigned());
		}
	}
@ Test public void testTaskManagerUnregistration ( ) throws Exception { final ResourceManagerId resourceManagerId = ResourceManagerId . generate ( ) ; final ResourceActions resourceManagerActions = mock ( ResourceActions . class ) ; final JobID jobId = new JobID ( ) ; 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 ( new CompletableFuture < > ( ) ) ; final TaskExecutorConnection taskManagerConnection = new TaskExecutorConnection ( taskExecutorGateway ) ; ResourceID resourceId = ResourceID . generate ( ) ; final SlotID slotId1 = new SlotID ( resourceId , 0 ) ; final SlotID slotId2 = new SlotID ( resourceId , 1 ) ; final AllocationID allocationId1 = new AllocationID ( ) ; final AllocationID allocationId2 = new AllocationID ( ) ; final ResourceProfile resourceProfile = new ResourceProfile ( 42.0 , 1337 ) ; final SlotStatus slotStatus1 = new SlotStatus ( slotId1 , resourceProfile , jobId , allocationId1 ) ; final SlotStatus slotStatus2 = new SlotStatus ( slotId2 , resourceProfile ) ; final SlotReport slotReport = new SlotReport ( Arrays . asList ( slotStatus1 , slotStatus2 ) ) ; final SlotRequest slotRequest = new SlotRequest ( new JobID ( ) , allocationId2 , resourceProfile , "foobar" ) ; try ( SlotManager slotManager = createSlotManager ( resourceManagerId , resourceManagerActions ) ) { slotManager . registerTaskManager ( taskManagerConnection , slotReport ) ; assertTrue ( "The number registered slots does not equal the expected number." , 2 == slotManager . getNumberRegisteredSlots ( ) ) ; TaskManagerSlot slot1 = slotManager . getSlot ( slotId1 ) ; TaskManagerSlot slot2 = slotManager . getSlot ( slotId2 ) ; assertTrue ( slot1 . getState ( ) == TaskManagerSlot . State . ALLOCATED ) ; assertTrue ( slot2 . getState ( ) == TaskManagerSlot . State . FREE ) ; assertTrue ( slotManager . registerSlotRequest ( slotRequest ) ) ; assertFalse ( slot2 . getState ( ) == TaskManagerSlot . State . FREE ) ; assertTrue ( slot2 . getState ( ) == TaskManagerSlot . State . PENDING ) ; PendingSlotRequest pendingSlotRequest = slotManager . getSlotRequest ( allocationId2 ) ; assertTrue ( "The pending slot request should have been assigned to slot 2" , pendingSlotRequest . isAssigned ( ) ) ; slotManager . unregisterTaskManager ( taskManagerConnection . getInstanceID ( ) ) ; assertTrue ( 0 == slotManager . getNumberRegisteredSlots ( ) ) ; assertFalse ( pendingSlotRequest . isAssigned ( ) ) ; } }
Tests that un-registration of task managers will free and remove all registered slots.
Tests that un - registration of task managers will free and remove all registered slots .