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 task manager timeout does not remove the slots from the SlotManager.
	 * A timeout should only trigger the {@link ResourceActions#releaseResource(InstanceID)}
	 * callback. The receiver of the callback can then decide what to do with the TaskManager.
	 *
	 * FLINK-7793
	 */
Tests that a task manager timeout does not remove the slots from the SlotManager.
A timeout should only trigger the  ResourceActions#releaseResource(InstanceID)
callback. The receiver of the callback can then decide what to do with the TaskManager.


Tests that a task manager timeout does not remove the slots from the SlotManager . A timeout should only trigger the ResourceActions#releaseResource ( InstanceID ) callback . The receiver of the callback can then decide what to do with the TaskManager .
@Test
	public void testTaskManagerTimeoutDoesNotRemoveSlots() throws Exception {
		final Time taskManagerTimeout = Time.milliseconds(10L);
		final ResourceManagerId resourceManagerId = ResourceManagerId.generate();
		final ResourceActions resourceActions = mock(ResourceActions.class);
		final TaskExecutorGateway taskExecutorGateway = mock(TaskExecutorGateway.class);
		final TaskExecutorConnection taskExecutorConnection = new TaskExecutorConnection(taskExecutorGateway);
		final SlotStatus slotStatus = new SlotStatus(
			new SlotID(ResourceID.generate(), 0),
			new ResourceProfile(1.0, 1));
		final SlotReport initialSlotReport = new SlotReport(slotStatus);
		try (final SlotManager slotManager = new SlotManager(
			TestingUtils.defaultScheduledExecutor(),
			TestingUtils.infiniteTime(),
			TestingUtils.infiniteTime(),
			taskManagerTimeout)) {
			slotManager.start(resourceManagerId, Executors.directExecutor(), resourceActions);
			slotManager.registerTaskManager(taskExecutorConnection, initialSlotReport);
			assertEquals(1, slotManager.getNumberRegisteredSlots());
			verify(resourceActions, timeout(taskManagerTimeout.toMilliseconds() * 20L).atLeast(1)).releaseResource(eq(taskExecutorConnection.getInstanceID()));
			assertEquals(1, slotManager.getNumberRegisteredSlots());
			slotManager.unregisterTaskManager(taskExecutorConnection.getInstanceID());
			assertEquals(0, slotManager.getNumberRegisteredSlots());
		}
	}
@ Test public void testTaskManagerTimeoutDoesNotRemoveSlots ( ) throws Exception { final Time taskManagerTimeout = Time . milliseconds ( 10L ) ; final ResourceManagerId resourceManagerId = ResourceManagerId . generate ( ) ; final ResourceActions resourceActions = mock ( ResourceActions . class ) ; final TaskExecutorGateway taskExecutorGateway = mock ( TaskExecutorGateway . class ) ; final TaskExecutorConnection taskExecutorConnection = new TaskExecutorConnection ( taskExecutorGateway ) ; final SlotStatus slotStatus = new SlotStatus ( new SlotID ( ResourceID . generate ( ) , 0 ) , new ResourceProfile ( 1.0 , 1 ) ) ; final SlotReport initialSlotReport = new SlotReport ( slotStatus ) ; try ( final SlotManager slotManager = new SlotManager ( TestingUtils . defaultScheduledExecutor ( ) , TestingUtils . infiniteTime ( ) , TestingUtils . infiniteTime ( ) , taskManagerTimeout ) ) { slotManager . start ( resourceManagerId , Executors . directExecutor ( ) , resourceActions ) ; slotManager . registerTaskManager ( taskExecutorConnection , initialSlotReport ) ; assertEquals ( 1 , slotManager . getNumberRegisteredSlots ( ) ) ; verify ( resourceActions , timeout ( taskManagerTimeout . toMilliseconds ( ) * 20L ) . atLeast ( 1 ) ) . releaseResource ( eq ( taskExecutorConnection . getInstanceID ( ) ) ) ; assertEquals ( 1 , slotManager . getNumberRegisteredSlots ( ) ) ; slotManager . unregisterTaskManager ( taskExecutorConnection . getInstanceID ( ) ) ; assertEquals ( 0 , slotManager . getNumberRegisteredSlots ( ) ) ; } }
Tests that a task manager timeout does not remove the slots from the SlotManager.
Tests that a task manager timeout does not remove the slots from the SlotManager .
// wait for the timeout call to happen