docs: @Override
  	public CompletableFuture<LogicalSlot> allocateSlot(
  		SlotRequestId slotRequestId,
  		ScheduledUnit scheduledUnit,
  		SlotProfile slotProfile,
  		boolean allowQueuedScheduling,
  		Time allocationTimeout) {
  		log.debug("Received slot request [{}] for task: {}", slotRequestId, scheduledUnit.getTaskToExecute());
  
  		componentMainThreadExecutor.assertRunningInMainThread();
  
  		final CompletableFuture<LogicalSlot> allocationResultFuture = new CompletableFuture<>();
  
  		CompletableFuture<LogicalSlot> allocationFuture = scheduledUnit.getSlotSharingGroupId() == null ?
  			allocateSingleSlot(slotRequestId, slotProfile, allowQueuedScheduling, allocationTimeout) :
  			allocateSharedSlot(slotRequestId, scheduledUnit, slotProfile, allowQueuedScheduling, allocationTimeout);
  
  		allocationFuture.whenComplete((LogicalSlot slot, Throwable failure) -> {
  			if (failure != null) {
  				cancelSlotRequest(
  					slotRequestId,
  					scheduledUnit.getSlotSharingGroupId(),
  					failure);
  				allocationResultFuture.completeExceptionally(failure);
  			} else {
  				allocationResultFuture.complete(slot);
  			}
  		});
  
  		return allocationResultFuture;
  	}