@Test public void testDuplicatePendingSlotRequest() throws Exception { final ResourceManagerId resourceManagerId = ResourceManagerId.generate(); final AtomicInteger numberAllocateResourceFunctionCalls = new AtomicInteger(0); final ResourceActions resourceManagerActions = new TestingResourceActionsBuilder() .setAllocateResourceConsumer(resourceProfile -> numberAllocateResourceFunctionCalls.incrementAndGet()) .build(); 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"); try (SlotManager slotManager = createSlotManager(resourceManagerId, resourceManagerActions)) { assertTrue(slotManager.registerSlotRequest(slotRequest1)); assertFalse(slotManager.registerSlotRequest(slotRequest2)); } assertThat(numberAllocateResourceFunctionCalls.get(), is(1)); }
public boolean registerSlotRequest(SlotRequest slotRequest) throws SlotManagerException { checkInit(); if (checkDuplicateRequest(slotRequest.getAllocationId())) { LOG.debug("Ignoring a duplicate slot request with allocation id {}.", slotRequest.getAllocationId()); return false; } else { PendingSlotRequest pendingSlotRequest = new PendingSlotRequest(slotRequest); pendingSlotRequests.put(slotRequest.getAllocationId(), pendingSlotRequest); try { internalRequestSlot(pendingSlotRequest); } catch (ResourceManagerException e) { pendingSlotRequests.remove(slotRequest.getAllocationId()); throw new SlotManagerException("Could not fulfill slot request " + slotRequest.getAllocationId() + '.', e); } return true; } }
SlotManager implements AutoCloseable { public boolean registerSlotRequest(SlotRequest slotRequest) throws SlotManagerException { checkInit(); if (checkDuplicateRequest(slotRequest.getAllocationId())) { LOG.debug("Ignoring a duplicate slot request with allocation id {}.", slotRequest.getAllocationId()); return false; } else { PendingSlotRequest pendingSlotRequest = new PendingSlotRequest(slotRequest); pendingSlotRequests.put(slotRequest.getAllocationId(), pendingSlotRequest); try { internalRequestSlot(pendingSlotRequest); } catch (ResourceManagerException e) { pendingSlotRequests.remove(slotRequest.getAllocationId()); throw new SlotManagerException("Could not fulfill slot request " + slotRequest.getAllocationId() + '.', e); } return true; } } }
SlotManager implements AutoCloseable { public boolean registerSlotRequest(SlotRequest slotRequest) throws SlotManagerException { checkInit(); if (checkDuplicateRequest(slotRequest.getAllocationId())) { LOG.debug("Ignoring a duplicate slot request with allocation id {}.", slotRequest.getAllocationId()); return false; } else { PendingSlotRequest pendingSlotRequest = new PendingSlotRequest(slotRequest); pendingSlotRequests.put(slotRequest.getAllocationId(), pendingSlotRequest); try { internalRequestSlot(pendingSlotRequest); } catch (ResourceManagerException e) { pendingSlotRequests.remove(slotRequest.getAllocationId()); throw new SlotManagerException("Could not fulfill slot request " + slotRequest.getAllocationId() + '.', e); } return true; } } SlotManager(
			ScheduledExecutor scheduledExecutor,
			Time taskManagerRequestTimeout,
			Time slotRequestTimeout,
			Time taskManagerTimeout,
			boolean waitResultConsumedBeforeRelease); }
SlotManager implements AutoCloseable { public boolean registerSlotRequest(SlotRequest slotRequest) throws SlotManagerException { checkInit(); if (checkDuplicateRequest(slotRequest.getAllocationId())) { LOG.debug("Ignoring a duplicate slot request with allocation id {}.", slotRequest.getAllocationId()); return false; } else { PendingSlotRequest pendingSlotRequest = new PendingSlotRequest(slotRequest); pendingSlotRequests.put(slotRequest.getAllocationId(), pendingSlotRequest); try { internalRequestSlot(pendingSlotRequest); } catch (ResourceManagerException e) { pendingSlotRequests.remove(slotRequest.getAllocationId()); throw new SlotManagerException("Could not fulfill slot request " + slotRequest.getAllocationId() + '.', e); } return true; } } 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 boolean registerSlotRequest(SlotRequest slotRequest) throws SlotManagerException { checkInit(); if (checkDuplicateRequest(slotRequest.getAllocationId())) { LOG.debug("Ignoring a duplicate slot request with allocation id {}.", slotRequest.getAllocationId()); return false; } else { PendingSlotRequest pendingSlotRequest = new PendingSlotRequest(slotRequest); pendingSlotRequests.put(slotRequest.getAllocationId(), pendingSlotRequest); try { internalRequestSlot(pendingSlotRequest); } catch (ResourceManagerException e) { pendingSlotRequests.remove(slotRequest.getAllocationId()); throw new SlotManagerException("Could not fulfill slot request " + slotRequest.getAllocationId() + '.', e); } return true; } } 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();  }