docs: public CompletableFuture<Execution> allocateAndAssignSlotForExecution(
  			SlotProvider slotProvider,
  			boolean queued,
  			LocationPreferenceConstraint locationPreferenceConstraint,
  			@Nonnull Set<AllocationID> allPreviousExecutionGraphAllocationIds,
  			Time allocationTimeout) throws IllegalExecutionStateException {
  
  		checkNotNull(slotProvider);
  
  		assertRunningInJobMasterMainThread();
  
  		final SlotSharingGroup sharingGroup = vertex.getJobVertex().getSlotSharingGroup();
  		final CoLocationConstraint locationConstraint = vertex.getLocationConstraint();
  
  		// sanity check
  		if (locationConstraint != null && sharingGroup == null) {
  			throw new IllegalStateException(
  					"Trying to schedule with co-location constraint but without slot sharing allowed.");
  		}
  
  		// this method only works if the execution is in the state 'CREATED'
  		if (transitionState(CREATED, SCHEDULED)) {
  
  			final SlotSharingGroupId slotSharingGroupId = sharingGroup != null ? sharingGroup.getSlotSharingGroupId() : null;
  
  			ScheduledUnit toSchedule = locationConstraint == null ?
  					new ScheduledUnit(this, slotSharingGroupId) :
  					new ScheduledUnit(this, slotSharingGroupId, locationConstraint);
  
  			// try to extract previous allocation ids, if applicable, so that we can reschedule to the same slot
  			ExecutionVertex executionVertex = getVertex();
  			AllocationID lastAllocation = executionVertex.getLatestPriorAllocation();
  
  			Collection<AllocationID> previousAllocationIDs =
  				lastAllocation != null ? Collections.singletonList(lastAllocation) : Collections.emptyList();
  
  			// calculate the preferred locations
  			final CompletableFuture<Collection<TaskManagerLocation>> preferredLocationsFuture =
  				calculatePreferredLocations(locationPreferenceConstraint);
  
  			final SlotRequestId slotRequestId = new SlotRequestId();
  
  			final CompletableFuture<LogicalSlot> logicalSlotFuture =
  				preferredLocationsFuture.thenCompose(
  					(Collection<TaskManagerLocation> preferredLocations) ->
  						slotProvider.allocateSlot(
  							slotRequestId,
  							toSchedule,
  							new SlotProfile(
  								ResourceProfile.UNKNOWN,
  								preferredLocations,
  								previousAllocationIDs,
  								allPreviousExecutionGraphAllocationIds),
  							queued,
  							allocationTimeout));
  
  			// register call back to cancel slot request in case that the execution gets canceled
  			releaseFuture.whenComplete(
  				(Object ignored, Throwable throwable) -> {
  					if (logicalSlotFuture.cancel(false)) {
  						slotProvider.cancelSlotRequest(
  							slotRequestId,
  							slotSharingGroupId,
  							new FlinkException("Execution " + this + " was released."));
  					}
  				});
  
  			// This forces calls to the slot pool back into the main thread, for normal and exceptional completion
  			return logicalSlotFuture.handle(
  				(LogicalSlot logicalSlot, Throwable failure) -> {
  
  					if (failure != null) {
  						throw new CompletionException(failure);
  					}
  
  					if (tryAssignResource(logicalSlot)) {
  						return this;
  					} else {
  						// release the slot
  						logicalSlot.releaseSlot(new FlinkException("Could not assign logical slot to execution " + this + '.'));
  						throw new CompletionException(
  							new FlinkException(
  								"Could not assign slot " + logicalSlot + " to execution " + this + " because it has already been assigned "));
  					}
  				});
  		} else {
  			// call race, already deployed, or already done
  			throw new IllegalExecutionStateException(this, CREATED, state);
  		}
  	}