docs: public void deploy() throws JobException {
  		assertRunningInJobMasterMainThread();
  
  		final LogicalSlot slot  = assignedResource;
  
  		checkNotNull(slot, "In order to deploy the execution we first have to assign a resource via tryAssignResource.");
  
  		// Check if the TaskManager died in the meantime
  		// This only speeds up the response to TaskManagers failing concurrently to deployments.
  		// The more general check is the rpcTimeout of the deployment call
  		if (!slot.isAlive()) {
  			throw new JobException("Target slot (TaskManager) for deployment is no longer alive.");
  		}
  
  		// make sure exactly one deployment call happens from the correct state
  		// note: the transition from CREATED to DEPLOYING is for testing purposes only
  		ExecutionState previous = this.state;
  		if (previous == SCHEDULED || previous == CREATED) {
  			if (!transitionState(previous, DEPLOYING)) {
  				// race condition, someone else beat us to the deploying call.
  				// this should actually not happen and indicates a race somewhere else
  				throw new IllegalStateException("Cannot deploy task: Concurrent deployment call race.");
  			}
  		}
  		else {
  			// vertex may have been cancelled, or it was already scheduled
  			throw new IllegalStateException("The vertex must be in CREATED or SCHEDULED state to be deployed. Found state " + previous);
  		}
  
  		if (this != slot.getPayload()) {
  			throw new IllegalStateException(
  				String.format("The execution %s has not been assigned to the assigned slot.", this));
  		}
  
  		try {
  
  			// race double check, did we fail/cancel and do we need to release the slot?
  			if (this.state != DEPLOYING) {
  				slot.releaseSlot(new FlinkException("Actual state of execution " + this + " (" + state + ") does not match expected state DEPLOYING."));
  				return;
  			}
  
  			if (LOG.isInfoEnabled()) {
  				LOG.info(String.format("Deploying %s (attempt #%d) to %s", vertex.getTaskNameWithSubtaskIndex(),
  						attemptNumber, getAssignedResourceLocation()));
  			}
  
  			final TaskDeploymentDescriptor deployment = vertex.createDeploymentDescriptor(
  				attemptId,
  				slot,
  				taskRestore,
  				attemptNumber);
  
  			// null taskRestore to let it be GC'ed
  			taskRestore = null;
  
  			final TaskManagerGateway taskManagerGateway = slot.getTaskManagerGateway();
  
  			final ComponentMainThreadExecutor jobMasterMainThreadExecutor =
  				vertex.getExecutionGraph().getJobMasterMainThreadExecutor();
  
  
  			// We run the submission in the future executor so that the serialization of large TDDs does not block
  			// the main thread and sync back to the main thread once submission is completed.
  			CompletableFuture.supplyAsync(() -> taskManagerGateway.submitTask(deployment, rpcTimeout), executor)
  				.thenCompose(Function.identity())
  				.whenCompleteAsync(
  					(ack, failure) -> {
  						// only respond to the failure case
  						if (failure != null) {
  							if (failure instanceof TimeoutException) {
  								String taskname = vertex.getTaskNameWithSubtaskIndex() + " (" + attemptId + ')';
  
  								markFailed(new Exception(
  									"Cannot deploy task " + taskname + " - TaskManager (" + getAssignedResourceLocation()
  										+ ") not responding after a rpcTimeout of " + rpcTimeout, failure));
  							} else {
  								markFailed(failure);
  							}
  						}
  					},
  					jobMasterMainThreadExecutor);
  
  		}
  		catch (Throwable t) {
  			markFailed(t);
  			ExceptionUtils.rethrow(t);
  		}
  	}