@Test public void testOfferSlot() throws Exception { final CompletableFuture<SlotRequest> slotRequestFuture = new CompletableFuture<>(); resourceManagerGateway.setRequestSlotConsumer(slotRequestFuture::complete); try (SlotPoolImpl slotPool = createSlotPoolImpl()) { setupSlotPool(slotPool, resourceManagerGateway, mainThreadExecutor); Scheduler scheduler = setupScheduler(slotPool, mainThreadExecutor); slotPool.registerTaskManager(taskManagerLocation.getResourceID()); CompletableFuture<LogicalSlot> future = scheduler.allocateSlot( new SlotRequestId(), new DummyScheduledUnit(), SlotProfile.noLocality(DEFAULT_TESTING_PROFILE), true, timeout); assertFalse(future.isDone()); final SlotRequest slotRequest = slotRequestFuture.get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS); final SlotOffer slotOffer = new SlotOffer( slotRequest.getAllocationId(), 0, DEFAULT_TESTING_PROFILE); final TaskManagerLocation invalidTaskManagerLocation = new LocalTaskManagerLocation(); assertFalse(slotPool.offerSlot(invalidTaskManagerLocation, taskManagerGateway, slotOffer)); final SlotOffer nonRequestedSlotOffer = new SlotOffer( new AllocationID(), 0, DEFAULT_TESTING_PROFILE); assertTrue(slotPool.offerSlot(taskManagerLocation, taskManagerGateway, nonRequestedSlotOffer)); assertTrue(slotPool.offerSlot(taskManagerLocation, taskManagerGateway, slotOffer)); LogicalSlot slot = future.get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS); assertTrue(slot.isAlive()); assertTrue(slotPool.offerSlot(taskManagerLocation, taskManagerGateway, slotOffer)); assertTrue(slot.isAlive()); final SlotOffer anotherSlotOfferWithSameAllocationId = new SlotOffer( slotRequest.getAllocationId(), 1, DEFAULT_TESTING_PROFILE); assertFalse(slotPool.offerSlot(taskManagerLocation, taskManagerGateway, anotherSlotOfferWithSameAllocationId)); TaskManagerLocation anotherTaskManagerLocation = new LocalTaskManagerLocation(); assertFalse(slotPool.offerSlot(anotherTaskManagerLocation, taskManagerGateway, slotOffer)); slot.releaseSlot(); assertTrue(slotPool.offerSlot(taskManagerLocation, taskManagerGateway, slotOffer)); assertFalse(slotPool.offerSlot(taskManagerLocation, taskManagerGateway, anotherSlotOfferWithSameAllocationId)); assertFalse(slotPool.offerSlot(anotherTaskManagerLocation, taskManagerGateway, slotOffer)); } }
boolean offerSlot( final TaskManagerLocation taskManagerLocation, final TaskManagerGateway taskManagerGateway, final SlotOffer slotOffer) { componentMainThreadExecutor.assertRunningInMainThread(); final ResourceID resourceID = taskManagerLocation.getResourceID(); final AllocationID allocationID = slotOffer.getAllocationId(); if (!registeredTaskManagers.contains(resourceID)) { log.debug("Received outdated slot offering [{}] from unregistered TaskManager: {}", slotOffer.getAllocationId(), taskManagerLocation); return false; } AllocatedSlot existingSlot; if ((existingSlot = allocatedSlots.get(allocationID)) != null || (existingSlot = availableSlots.get(allocationID)) != null) { final SlotID existingSlotId = existingSlot.getSlotId(); final SlotID newSlotId = new SlotID(taskManagerLocation.getResourceID(), slotOffer.getSlotIndex()); if (existingSlotId.equals(newSlotId)) { log.info("Received repeated offer for slot [{}]. Ignoring.", allocationID); return true; } else { return false; } } final AllocatedSlot allocatedSlot = new AllocatedSlot( allocationID, taskManagerLocation, slotOffer.getSlotIndex(), slotOffer.getResourceProfile(), taskManagerGateway); PendingRequest pendingRequest = pendingRequests.removeKeyB(allocationID); if (pendingRequest != null) { allocatedSlots.add(pendingRequest.getSlotRequestId(), allocatedSlot); if (!pendingRequest.getAllocatedSlotFuture().complete(allocatedSlot)) { allocatedSlots.remove(pendingRequest.getSlotRequestId()); tryFulfillSlotRequestOrMakeAvailable(allocatedSlot); } else { log.debug("Fulfilled slot request [{}] with allocated slot [{}].", pendingRequest.getSlotRequestId(), allocationID); } } else { tryFulfillSlotRequestOrMakeAvailable(allocatedSlot); } return true; }
SlotPoolImpl implements SlotPool { boolean offerSlot( final TaskManagerLocation taskManagerLocation, final TaskManagerGateway taskManagerGateway, final SlotOffer slotOffer) { componentMainThreadExecutor.assertRunningInMainThread(); final ResourceID resourceID = taskManagerLocation.getResourceID(); final AllocationID allocationID = slotOffer.getAllocationId(); if (!registeredTaskManagers.contains(resourceID)) { log.debug("Received outdated slot offering [{}] from unregistered TaskManager: {}", slotOffer.getAllocationId(), taskManagerLocation); return false; } AllocatedSlot existingSlot; if ((existingSlot = allocatedSlots.get(allocationID)) != null || (existingSlot = availableSlots.get(allocationID)) != null) { final SlotID existingSlotId = existingSlot.getSlotId(); final SlotID newSlotId = new SlotID(taskManagerLocation.getResourceID(), slotOffer.getSlotIndex()); if (existingSlotId.equals(newSlotId)) { log.info("Received repeated offer for slot [{}]. Ignoring.", allocationID); return true; } else { return false; } } final AllocatedSlot allocatedSlot = new AllocatedSlot( allocationID, taskManagerLocation, slotOffer.getSlotIndex(), slotOffer.getResourceProfile(), taskManagerGateway); PendingRequest pendingRequest = pendingRequests.removeKeyB(allocationID); if (pendingRequest != null) { allocatedSlots.add(pendingRequest.getSlotRequestId(), allocatedSlot); if (!pendingRequest.getAllocatedSlotFuture().complete(allocatedSlot)) { allocatedSlots.remove(pendingRequest.getSlotRequestId()); tryFulfillSlotRequestOrMakeAvailable(allocatedSlot); } else { log.debug("Fulfilled slot request [{}] with allocated slot [{}].", pendingRequest.getSlotRequestId(), allocationID); } } else { tryFulfillSlotRequestOrMakeAvailable(allocatedSlot); } return true; } }
SlotPoolImpl implements SlotPool { boolean offerSlot( final TaskManagerLocation taskManagerLocation, final TaskManagerGateway taskManagerGateway, final SlotOffer slotOffer) { componentMainThreadExecutor.assertRunningInMainThread(); final ResourceID resourceID = taskManagerLocation.getResourceID(); final AllocationID allocationID = slotOffer.getAllocationId(); if (!registeredTaskManagers.contains(resourceID)) { log.debug("Received outdated slot offering [{}] from unregistered TaskManager: {}", slotOffer.getAllocationId(), taskManagerLocation); return false; } AllocatedSlot existingSlot; if ((existingSlot = allocatedSlots.get(allocationID)) != null || (existingSlot = availableSlots.get(allocationID)) != null) { final SlotID existingSlotId = existingSlot.getSlotId(); final SlotID newSlotId = new SlotID(taskManagerLocation.getResourceID(), slotOffer.getSlotIndex()); if (existingSlotId.equals(newSlotId)) { log.info("Received repeated offer for slot [{}]. Ignoring.", allocationID); return true; } else { return false; } } final AllocatedSlot allocatedSlot = new AllocatedSlot( allocationID, taskManagerLocation, slotOffer.getSlotIndex(), slotOffer.getResourceProfile(), taskManagerGateway); PendingRequest pendingRequest = pendingRequests.removeKeyB(allocationID); if (pendingRequest != null) { allocatedSlots.add(pendingRequest.getSlotRequestId(), allocatedSlot); if (!pendingRequest.getAllocatedSlotFuture().complete(allocatedSlot)) { allocatedSlots.remove(pendingRequest.getSlotRequestId()); tryFulfillSlotRequestOrMakeAvailable(allocatedSlot); } else { log.debug("Fulfilled slot request [{}] with allocated slot [{}].", pendingRequest.getSlotRequestId(), allocationID); } } else { tryFulfillSlotRequestOrMakeAvailable(allocatedSlot); } return true; } SlotPoolImpl(
			JobID jobId,
			Clock clock,
			Time rpcTimeout,
			Time idleSlotTimeout,
			Time batchSlotTimeout); }
SlotPoolImpl implements SlotPool { boolean offerSlot( final TaskManagerLocation taskManagerLocation, final TaskManagerGateway taskManagerGateway, final SlotOffer slotOffer) { componentMainThreadExecutor.assertRunningInMainThread(); final ResourceID resourceID = taskManagerLocation.getResourceID(); final AllocationID allocationID = slotOffer.getAllocationId(); if (!registeredTaskManagers.contains(resourceID)) { log.debug("Received outdated slot offering [{}] from unregistered TaskManager: {}", slotOffer.getAllocationId(), taskManagerLocation); return false; } AllocatedSlot existingSlot; if ((existingSlot = allocatedSlots.get(allocationID)) != null || (existingSlot = availableSlots.get(allocationID)) != null) { final SlotID existingSlotId = existingSlot.getSlotId(); final SlotID newSlotId = new SlotID(taskManagerLocation.getResourceID(), slotOffer.getSlotIndex()); if (existingSlotId.equals(newSlotId)) { log.info("Received repeated offer for slot [{}]. Ignoring.", allocationID); return true; } else { return false; } } final AllocatedSlot allocatedSlot = new AllocatedSlot( allocationID, taskManagerLocation, slotOffer.getSlotIndex(), slotOffer.getResourceProfile(), taskManagerGateway); PendingRequest pendingRequest = pendingRequests.removeKeyB(allocationID); if (pendingRequest != null) { allocatedSlots.add(pendingRequest.getSlotRequestId(), allocatedSlot); if (!pendingRequest.getAllocatedSlotFuture().complete(allocatedSlot)) { allocatedSlots.remove(pendingRequest.getSlotRequestId()); tryFulfillSlotRequestOrMakeAvailable(allocatedSlot); } else { log.debug("Fulfilled slot request [{}] with allocated slot [{}].", pendingRequest.getSlotRequestId(), allocationID); } } else { tryFulfillSlotRequestOrMakeAvailable(allocatedSlot); } return true; } SlotPoolImpl(
			JobID jobId,
			Clock clock,
			Time rpcTimeout,
			Time idleSlotTimeout,
			Time batchSlotTimeout); void start(
		@Nonnull JobMasterId jobMasterId,
		@Nonnull String newJobManagerAddress,
		@Nonnull ComponentMainThreadExecutor componentMainThreadExecutor); @Override void suspend(); @Override void close(); @Override void connectToResourceManager(@Nonnull ResourceManagerGateway resourceManagerGateway); @Override void disconnectResourceManager(); @Override void releaseSlot(@Nonnull SlotRequestId slotRequestId, @Nullable Throwable cause); @Override Optional<PhysicalSlot> allocateAvailableSlot(
		@Nonnull SlotRequestId slotRequestId,
		@Nonnull AllocationID allocationID); @Nonnull @Override CompletableFuture<PhysicalSlot> requestNewAllocatedSlot(
			@Nonnull SlotRequestId slotRequestId,
			@Nonnull ResourceProfile resourceProfile,
			Time timeout); @Nonnull @Override CompletableFuture<PhysicalSlot> requestNewAllocatedBatchSlot(
		@Nonnull SlotRequestId slotRequestId,
		@Nonnull ResourceProfile resourceProfile); @Override @Nonnull Collection<SlotInfo> getAvailableSlotsInformation(); @Override Collection<SlotOffer> offerSlots(
			TaskManagerLocation taskManagerLocation,
			TaskManagerGateway taskManagerGateway,
			Collection<SlotOffer> offers); @Override Optional<ResourceID> failAllocation(final AllocationID allocationID, final Exception cause); @Override boolean registerTaskManager(final ResourceID resourceID); @Override boolean releaseTaskManager(final ResourceID resourceId, final Exception cause); @Override AllocatedSlotReport createAllocatedSlotReport(ResourceID taskManagerId); }
SlotPoolImpl implements SlotPool { boolean offerSlot( final TaskManagerLocation taskManagerLocation, final TaskManagerGateway taskManagerGateway, final SlotOffer slotOffer) { componentMainThreadExecutor.assertRunningInMainThread(); final ResourceID resourceID = taskManagerLocation.getResourceID(); final AllocationID allocationID = slotOffer.getAllocationId(); if (!registeredTaskManagers.contains(resourceID)) { log.debug("Received outdated slot offering [{}] from unregistered TaskManager: {}", slotOffer.getAllocationId(), taskManagerLocation); return false; } AllocatedSlot existingSlot; if ((existingSlot = allocatedSlots.get(allocationID)) != null || (existingSlot = availableSlots.get(allocationID)) != null) { final SlotID existingSlotId = existingSlot.getSlotId(); final SlotID newSlotId = new SlotID(taskManagerLocation.getResourceID(), slotOffer.getSlotIndex()); if (existingSlotId.equals(newSlotId)) { log.info("Received repeated offer for slot [{}]. Ignoring.", allocationID); return true; } else { return false; } } final AllocatedSlot allocatedSlot = new AllocatedSlot( allocationID, taskManagerLocation, slotOffer.getSlotIndex(), slotOffer.getResourceProfile(), taskManagerGateway); PendingRequest pendingRequest = pendingRequests.removeKeyB(allocationID); if (pendingRequest != null) { allocatedSlots.add(pendingRequest.getSlotRequestId(), allocatedSlot); if (!pendingRequest.getAllocatedSlotFuture().complete(allocatedSlot)) { allocatedSlots.remove(pendingRequest.getSlotRequestId()); tryFulfillSlotRequestOrMakeAvailable(allocatedSlot); } else { log.debug("Fulfilled slot request [{}] with allocated slot [{}].", pendingRequest.getSlotRequestId(), allocationID); } } else { tryFulfillSlotRequestOrMakeAvailable(allocatedSlot); } return true; } SlotPoolImpl(
			JobID jobId,
			Clock clock,
			Time rpcTimeout,
			Time idleSlotTimeout,
			Time batchSlotTimeout); void start(
		@Nonnull JobMasterId jobMasterId,
		@Nonnull String newJobManagerAddress,
		@Nonnull ComponentMainThreadExecutor componentMainThreadExecutor); @Override void suspend(); @Override void close(); @Override void connectToResourceManager(@Nonnull ResourceManagerGateway resourceManagerGateway); @Override void disconnectResourceManager(); @Override void releaseSlot(@Nonnull SlotRequestId slotRequestId, @Nullable Throwable cause); @Override Optional<PhysicalSlot> allocateAvailableSlot(
		@Nonnull SlotRequestId slotRequestId,
		@Nonnull AllocationID allocationID); @Nonnull @Override CompletableFuture<PhysicalSlot> requestNewAllocatedSlot(
			@Nonnull SlotRequestId slotRequestId,
			@Nonnull ResourceProfile resourceProfile,
			Time timeout); @Nonnull @Override CompletableFuture<PhysicalSlot> requestNewAllocatedBatchSlot(
		@Nonnull SlotRequestId slotRequestId,
		@Nonnull ResourceProfile resourceProfile); @Override @Nonnull Collection<SlotInfo> getAvailableSlotsInformation(); @Override Collection<SlotOffer> offerSlots(
			TaskManagerLocation taskManagerLocation,
			TaskManagerGateway taskManagerGateway,
			Collection<SlotOffer> offers); @Override Optional<ResourceID> failAllocation(final AllocationID allocationID, final Exception cause); @Override boolean registerTaskManager(final ResourceID resourceID); @Override boolean releaseTaskManager(final ResourceID resourceId, final Exception cause); @Override AllocatedSlotReport createAllocatedSlotReport(ResourceID taskManagerId);  }