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 slots are updated with respect to the latest incoming slot report. This means that
	 * slots for which a report was received are updated accordingly.
	 */
Tests that slots are updated with respect to the latest incoming slot report. This means that
slots for which a report was received are updated accordingly.
Tests that slots are updated with respect to the latest incoming slot report . This means that slots for which a report was received are updated accordingly .
@Test
	public void testUpdateSlotReport() throws Exception {
		final ResourceManagerId resourceManagerId = ResourceManagerId.generate();
		final ResourceActions resourceManagerActions = mock(ResourceActions.class);
		final JobID jobId = new JobID();
		final AllocationID allocationId = new AllocationID();
		final ResourceID resourceId = ResourceID.generate();
		final SlotID slotId1 = new SlotID(resourceId, 0);
		final SlotID slotId2 = new SlotID(resourceId, 1);
		final ResourceProfile resourceProfile = new ResourceProfile(1.0, 1);
		final SlotStatus slotStatus1 = new SlotStatus(slotId1, resourceProfile);
		final SlotStatus slotStatus2 = new SlotStatus(slotId2, resourceProfile);
		final SlotStatus newSlotStatus2 = new SlotStatus(slotId2, resourceProfile, jobId, allocationId);
		final SlotReport slotReport1 = new SlotReport(Arrays.asList(slotStatus1, slotStatus2));
		final SlotReport slotReport2 = new SlotReport(Arrays.asList(newSlotStatus2, slotStatus1));
		final TaskExecutorGateway taskExecutorGateway = mock(TaskExecutorGateway.class);
		final TaskExecutorConnection taskManagerConnection = new TaskExecutorConnection(taskExecutorGateway);
		try (SlotManager slotManager = createSlotManager(resourceManagerId, resourceManagerActions)) {
			assertTrue(0 == slotManager.getNumberRegisteredSlots());
			slotManager.registerTaskManager(taskManagerConnection, slotReport1);
			TaskManagerSlot slot1 = slotManager.getSlot(slotId1);
			TaskManagerSlot slot2 = slotManager.getSlot(slotId2);
			assertTrue(2 == slotManager.getNumberRegisteredSlots());
			assertTrue(slot1.getState() == TaskManagerSlot.State.FREE);
			assertTrue(slot2.getState() == TaskManagerSlot.State.FREE);
			assertTrue(slotManager.reportSlotStatus(taskManagerConnection.getInstanceID(), slotReport2));
			assertTrue(2 == slotManager.getNumberRegisteredSlots());
			assertNotNull(slotManager.getSlot(slotId1));
			assertNotNull(slotManager.getSlot(slotId2));
			assertEquals(allocationId, slotManager.getSlot(slotId2).getAllocationId());
		}
	}
@ Test public void testUpdateSlotReport ( ) throws Exception { final ResourceManagerId resourceManagerId = ResourceManagerId . generate ( ) ; final ResourceActions resourceManagerActions = mock ( ResourceActions . class ) ; final JobID jobId = new JobID ( ) ; final AllocationID allocationId = new AllocationID ( ) ; final ResourceID resourceId = ResourceID . generate ( ) ; final SlotID slotId1 = new SlotID ( resourceId , 0 ) ; final SlotID slotId2 = new SlotID ( resourceId , 1 ) ; final ResourceProfile resourceProfile = new ResourceProfile ( 1.0 , 1 ) ; final SlotStatus slotStatus1 = new SlotStatus ( slotId1 , resourceProfile ) ; final SlotStatus slotStatus2 = new SlotStatus ( slotId2 , resourceProfile ) ; final SlotStatus newSlotStatus2 = new SlotStatus ( slotId2 , resourceProfile , jobId , allocationId ) ; final SlotReport slotReport1 = new SlotReport ( Arrays . asList ( slotStatus1 , slotStatus2 ) ) ; final SlotReport slotReport2 = new SlotReport ( Arrays . asList ( newSlotStatus2 , slotStatus1 ) ) ; final TaskExecutorGateway taskExecutorGateway = mock ( TaskExecutorGateway . class ) ; final TaskExecutorConnection taskManagerConnection = new TaskExecutorConnection ( taskExecutorGateway ) ; try ( SlotManager slotManager = createSlotManager ( resourceManagerId , resourceManagerActions ) ) { assertTrue ( 0 == slotManager . getNumberRegisteredSlots ( ) ) ; slotManager . registerTaskManager ( taskManagerConnection , slotReport1 ) ; TaskManagerSlot slot1 = slotManager . getSlot ( slotId1 ) ; TaskManagerSlot slot2 = slotManager . getSlot ( slotId2 ) ; assertTrue ( 2 == slotManager . getNumberRegisteredSlots ( ) ) ; assertTrue ( slot1 . getState ( ) == TaskManagerSlot . State . FREE ) ; assertTrue ( slot2 . getState ( ) == TaskManagerSlot . State . FREE ) ; assertTrue ( slotManager . reportSlotStatus ( taskManagerConnection . getInstanceID ( ) , slotReport2 ) ) ; assertTrue ( 2 == slotManager . getNumberRegisteredSlots ( ) ) ; assertNotNull ( slotManager . getSlot ( slotId1 ) ) ; assertNotNull ( slotManager . getSlot ( slotId2 ) ) ; assertEquals ( allocationId , slotManager . getSlot ( slotId2 ) . getAllocationId ( ) ) ; } }
Tests that slots are updated with respect to the latest incoming slot report.
Tests that slots are updated with respect to the latest incoming slot report .
// check that we don't have any slots registered // slotId2 should have been allocated for allocationId