docs: @Override
  	public CompletableFuture<Acknowledge> submitTask(
  			TaskDeploymentDescriptor tdd,
  			JobMasterId jobMasterId,
  			Time timeout) {
  
  		try {
  			final JobID jobId = tdd.getJobId();
  			final JobManagerConnection jobManagerConnection = jobManagerTable.get(jobId);
  
  			if (jobManagerConnection == null) {
  				final String message = "Could not submit task because there is no JobManager " +
  					"associated for the job " + jobId + '.';
  
  				log.debug(message);
  				throw new TaskSubmissionException(message);
  			}
  
  			if (!Objects.equals(jobManagerConnection.getJobMasterId(), jobMasterId)) {
  				final String message = "Rejecting the task submission because the job manager leader id " +
  					jobMasterId + " does not match the expected job manager leader id " +
  					jobManagerConnection.getJobMasterId() + '.';
  
  				log.debug(message);
  				throw new TaskSubmissionException(message);
  			}
  
  			if (!taskSlotTable.tryMarkSlotActive(jobId, tdd.getAllocationId())) {
  				final String message = "No task slot allocated for job ID " + jobId +
  					" and allocation ID " + tdd.getAllocationId() + '.';
  				log.debug(message);
  				throw new TaskSubmissionException(message);
  			}
  
  			// re-integrate offloaded data:
  			try {
  				tdd.loadBigData(blobCacheService.getPermanentBlobService());
  			} catch (IOException | ClassNotFoundException e) {
  				throw new TaskSubmissionException("Could not re-integrate offloaded TaskDeploymentDescriptor data.", e);
  			}
  
  			// deserialize the pre-serialized information
  			final JobInformation jobInformation;
  			final TaskInformation taskInformation;
  			try {
  				jobInformation = tdd.getSerializedJobInformation().deserializeValue(getClass().getClassLoader());
  				taskInformation = tdd.getSerializedTaskInformation().deserializeValue(getClass().getClassLoader());
  			} catch (IOException | ClassNotFoundException e) {
  				throw new TaskSubmissionException("Could not deserialize the job or task information.", e);
  			}
  
  			if (!jobId.equals(jobInformation.getJobId())) {
  				throw new TaskSubmissionException(
  					"Inconsistent job ID information inside TaskDeploymentDescriptor (" +
  						tdd.getJobId() + " vs. " + jobInformation.getJobId() + ")");
  			}
  
  			TaskMetricGroup taskMetricGroup = taskManagerMetricGroup.addTaskForJob(
  				jobInformation.getJobId(),
  				jobInformation.getJobName(),
  				taskInformation.getJobVertexId(),
  				tdd.getExecutionAttemptId(),
  				taskInformation.getTaskName(),
  				tdd.getSubtaskIndex(),
  				tdd.getAttemptNumber());
  
  			InputSplitProvider inputSplitProvider = new RpcInputSplitProvider(
  				jobManagerConnection.getJobManagerGateway(),
  				taskInformation.getJobVertexId(),
  				tdd.getExecutionAttemptId(),
  				taskManagerConfiguration.getTimeout());
  
  			TaskManagerActions taskManagerActions = jobManagerConnection.getTaskManagerActions();
  			CheckpointResponder checkpointResponder = jobManagerConnection.getCheckpointResponder();
  			GlobalAggregateManager aggregateManager = jobManagerConnection.getGlobalAggregateManager();
  
  			LibraryCacheManager libraryCache = jobManagerConnection.getLibraryCacheManager();
  			ResultPartitionConsumableNotifier resultPartitionConsumableNotifier = jobManagerConnection.getResultPartitionConsumableNotifier();
  			PartitionProducerStateChecker partitionStateChecker = jobManagerConnection.getPartitionStateChecker();
  
  			final TaskLocalStateStore localStateStore = localStateStoresManager.localStateStoreForSubtask(
  				jobId,
  				tdd.getAllocationId(),
  				taskInformation.getJobVertexId(),
  				tdd.getSubtaskIndex());
  
  			final JobManagerTaskRestore taskRestore = tdd.getTaskRestore();
  
  			final TaskStateManager taskStateManager = new TaskStateManagerImpl(
  				jobId,
  				tdd.getExecutionAttemptId(),
  				localStateStore,
  				taskRestore,
  				checkpointResponder);
  
  			Task task = new Task(
  				jobInformation,
  				taskInformation,
  				tdd.getExecutionAttemptId(),
  				tdd.getAllocationId(),
  				tdd.getSubtaskIndex(),
  				tdd.getAttemptNumber(),
  				tdd.getProducedPartitions(),
  				tdd.getInputGates(),
  				tdd.getTargetSlotNumber(),
  				taskExecutorServices.getMemoryManager(),
  				taskExecutorServices.getIOManager(),
  				taskExecutorServices.getNetworkEnvironment(),
  				taskExecutorServices.getKvStateService(),
  				taskExecutorServices.getBroadcastVariableManager(),
  				taskExecutorServices.getTaskEventDispatcher(),
  				taskStateManager,
  				taskManagerActions,
  				inputSplitProvider,
  				checkpointResponder,
  				aggregateManager,
  				blobCacheService,
  				libraryCache,
  				fileCache,
  				taskManagerConfiguration,
  				taskMetricGroup,
  				resultPartitionConsumableNotifier,
  				partitionStateChecker,
  				getRpcService().getExecutor());
  
  			log.info("Received task {}.", task.getTaskInfo().getTaskNameWithSubtasks());
  
  			boolean taskAdded;
  
  			try {
  				taskAdded = taskSlotTable.addTask(task);
  			} catch (SlotNotFoundException | SlotNotActiveException e) {
  				throw new TaskSubmissionException("Could not submit task.", e);
  			}
  
  			if (taskAdded) {
  				task.startTaskThread();
  
  				return CompletableFuture.completedFuture(Acknowledge.get());
  			} else {
  				final String message = "TaskManager already contains a task for id " +
  					task.getExecutionId() + '.';
  
  				log.debug(message);
  				throw new TaskSubmissionException(message);
  			}
  		} catch (TaskSubmissionException e) {
  			return FutureUtils.completedExceptionally(e);
  		}
  	}