@Test public void testFreeSlot() throws Exception { final ResourceManagerId resourceManagerId = ResourceManagerId.generate(); final ResourceID resourceID = ResourceID.generate(); final JobID jobId = new JobID(); final SlotID slotId = new SlotID(resourceID, 0); final AllocationID allocationId = new AllocationID(); final ResourceProfile resourceProfile = new ResourceProfile(42.0, 1337); ResourceActions resourceManagerActions = mock(ResourceActions.class); final TaskExecutorGateway taskExecutorGateway = mock(TaskExecutorGateway.class); final TaskExecutorConnection taskExecutorConnection = new TaskExecutorConnection(resourceID, taskExecutorGateway); final SlotStatus slotStatus = new SlotStatus(slotId, resourceProfile, jobId, allocationId); final SlotReport slotReport = new SlotReport(slotStatus); try (SlotManager slotManager = createSlotManager(resourceManagerId, resourceManagerActions)) { slotManager.registerTaskManager( taskExecutorConnection, slotReport); TaskManagerSlot slot = slotManager.getSlot(slotId); assertEquals("The slot has not been allocated to the expected allocation id.", allocationId, slot.getAllocationId()); slotManager.freeSlot(slotId, new AllocationID()); assertTrue(slot.getState() == TaskManagerSlot.State.ALLOCATED); 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()); } }
public void freeSlot(SlotID slotId, AllocationID allocationId) { checkInit(); TaskManagerSlot slot = slots.get(slotId); if (null != slot) { if (slot.getState() == TaskManagerSlot.State.ALLOCATED) { if (Objects.equals(allocationId, slot.getAllocationId())) { TaskManagerRegistration taskManagerRegistration = taskManagerRegistrations.get(slot.getInstanceId()); if (taskManagerRegistration == null) { throw new IllegalStateException("Trying to free a slot from a TaskManager " + slot.getInstanceId() + " which has not been registered."); } updateSlotState(slot, taskManagerRegistration, null, null); } else { LOG.debug("Received request to free slot {} with expected allocation id {}, " + "but actual allocation id {} differs. Ignoring the request.", slotId, allocationId, slot.getAllocationId()); } } else { LOG.debug("Slot {} has not been allocated.", allocationId); } } else { LOG.debug("Trying to free a slot {} which has not been registered. Ignoring this message.", slotId); } }
SlotManager implements AutoCloseable { public void freeSlot(SlotID slotId, AllocationID allocationId) { checkInit(); TaskManagerSlot slot = slots.get(slotId); if (null != slot) { if (slot.getState() == TaskManagerSlot.State.ALLOCATED) { if (Objects.equals(allocationId, slot.getAllocationId())) { TaskManagerRegistration taskManagerRegistration = taskManagerRegistrations.get(slot.getInstanceId()); if (taskManagerRegistration == null) { throw new IllegalStateException("Trying to free a slot from a TaskManager " + slot.getInstanceId() + " which has not been registered."); } updateSlotState(slot, taskManagerRegistration, null, null); } else { LOG.debug("Received request to free slot {} with expected allocation id {}, " + "but actual allocation id {} differs. Ignoring the request.", slotId, allocationId, slot.getAllocationId()); } } else { LOG.debug("Slot {} has not been allocated.", allocationId); } } else { LOG.debug("Trying to free a slot {} which has not been registered. Ignoring this message.", slotId); } } }
SlotManager implements AutoCloseable { public void freeSlot(SlotID slotId, AllocationID allocationId) { checkInit(); TaskManagerSlot slot = slots.get(slotId); if (null != slot) { if (slot.getState() == TaskManagerSlot.State.ALLOCATED) { if (Objects.equals(allocationId, slot.getAllocationId())) { TaskManagerRegistration taskManagerRegistration = taskManagerRegistrations.get(slot.getInstanceId()); if (taskManagerRegistration == null) { throw new IllegalStateException("Trying to free a slot from a TaskManager " + slot.getInstanceId() + " which has not been registered."); } updateSlotState(slot, taskManagerRegistration, null, null); } else { LOG.debug("Received request to free slot {} with expected allocation id {}, " + "but actual allocation id {} differs. Ignoring the request.", slotId, allocationId, slot.getAllocationId()); } } else { LOG.debug("Slot {} has not been allocated.", allocationId); } } else { LOG.debug("Trying to free a slot {} which has not been registered. Ignoring this message.", slotId); } } SlotManager(
			ScheduledExecutor scheduledExecutor,
			Time taskManagerRequestTimeout,
			Time slotRequestTimeout,
			Time taskManagerTimeout,
			boolean waitResultConsumedBeforeRelease); }
SlotManager implements AutoCloseable { public void freeSlot(SlotID slotId, AllocationID allocationId) { checkInit(); TaskManagerSlot slot = slots.get(slotId); if (null != slot) { if (slot.getState() == TaskManagerSlot.State.ALLOCATED) { if (Objects.equals(allocationId, slot.getAllocationId())) { TaskManagerRegistration taskManagerRegistration = taskManagerRegistrations.get(slot.getInstanceId()); if (taskManagerRegistration == null) { throw new IllegalStateException("Trying to free a slot from a TaskManager " + slot.getInstanceId() + " which has not been registered."); } updateSlotState(slot, taskManagerRegistration, null, null); } else { LOG.debug("Received request to free slot {} with expected allocation id {}, " + "but actual allocation id {} differs. Ignoring the request.", slotId, allocationId, slot.getAllocationId()); } } else { LOG.debug("Slot {} has not been allocated.", allocationId); } } else { LOG.debug("Trying to free a slot {} which has not been registered. Ignoring this message.", slotId); } } SlotManager(
			ScheduledExecutor scheduledExecutor,
			Time taskManagerRequestTimeout,
			Time slotRequestTimeout,
			Time taskManagerTimeout,
			boolean waitResultConsumedBeforeRelease); int getNumberRegisteredSlots(); int getNumberRegisteredSlotsOf(InstanceID instanceId); int getNumberFreeSlots(); int getNumberFreeSlotsOf(InstanceID instanceId); int getNumberPendingTaskManagerSlots(); void start(ResourceManagerId newResourceManagerId, Executor newMainThreadExecutor, ResourceActions newResourceActions); void suspend(); @Override void close(); boolean registerSlotRequest(SlotRequest slotRequest); boolean unregisterSlotRequest(AllocationID allocationId); void registerTaskManager(final TaskExecutorConnection taskExecutorConnection, SlotReport initialSlotReport); boolean unregisterTaskManager(InstanceID instanceId); boolean reportSlotStatus(InstanceID instanceId, SlotReport slotReport); void freeSlot(SlotID slotId, AllocationID allocationId); @VisibleForTesting void unregisterTaskManagersAndReleaseResources(); }
SlotManager implements AutoCloseable { public void freeSlot(SlotID slotId, AllocationID allocationId) { checkInit(); TaskManagerSlot slot = slots.get(slotId); if (null != slot) { if (slot.getState() == TaskManagerSlot.State.ALLOCATED) { if (Objects.equals(allocationId, slot.getAllocationId())) { TaskManagerRegistration taskManagerRegistration = taskManagerRegistrations.get(slot.getInstanceId()); if (taskManagerRegistration == null) { throw new IllegalStateException("Trying to free a slot from a TaskManager " + slot.getInstanceId() + " which has not been registered."); } updateSlotState(slot, taskManagerRegistration, null, null); } else { LOG.debug("Received request to free slot {} with expected allocation id {}, " + "but actual allocation id {} differs. Ignoring the request.", slotId, allocationId, slot.getAllocationId()); } } else { LOG.debug("Slot {} has not been allocated.", allocationId); } } else { LOG.debug("Trying to free a slot {} which has not been registered. Ignoring this message.", slotId); } } SlotManager(
			ScheduledExecutor scheduledExecutor,
			Time taskManagerRequestTimeout,
			Time slotRequestTimeout,
			Time taskManagerTimeout,
			boolean waitResultConsumedBeforeRelease); int getNumberRegisteredSlots(); int getNumberRegisteredSlotsOf(InstanceID instanceId); int getNumberFreeSlots(); int getNumberFreeSlotsOf(InstanceID instanceId); int getNumberPendingTaskManagerSlots(); void start(ResourceManagerId newResourceManagerId, Executor newMainThreadExecutor, ResourceActions newResourceActions); void suspend(); @Override void close(); boolean registerSlotRequest(SlotRequest slotRequest); boolean unregisterSlotRequest(AllocationID allocationId); void registerTaskManager(final TaskExecutorConnection taskExecutorConnection, SlotReport initialSlotReport); boolean unregisterTaskManager(InstanceID instanceId); boolean reportSlotStatus(InstanceID instanceId, SlotReport slotReport); void freeSlot(SlotID slotId, AllocationID allocationId); @VisibleForTesting void unregisterTaskManagersAndReleaseResources();  }