code: @Override
  	public CompletableFuture<LogicalSlot> allocateSlot(
  			SlotRequestId slotRequestId,
  			ScheduledUnit task,
  			SlotProfile slotProfile,
  			boolean allowQueued,
  			Time allocationTimeout) {
  
  		try {
  			final Object ret = scheduleTask(task, allowQueued, slotProfile.getPreferredLocations());
  
  			if (ret instanceof SimpleSlot) {
  				return CompletableFuture.completedFuture((SimpleSlot) ret);
  			}
  			else if (ret instanceof CompletableFuture) {
  				@SuppressWarnings("unchecked")
  				CompletableFuture<LogicalSlot> typed = (CompletableFuture<LogicalSlot>) ret;
  				return FutureUtils.orTimeout(typed, allocationTimeout.toMilliseconds(), TimeUnit.MILLISECONDS);
  			}
  			else {
  				// this should never happen, simply guard this case with an exception
  				throw new RuntimeException();
  			}
  		} catch (NoResourceAvailableException e) {
  			return FutureUtils.completedExceptionally(e);
  		}
  	}
docs: ------------------------------------------------------------------------