docs: @Override
  	public CompletableFuture<Acknowledge> requestSlot(
  		final SlotID slotId,
  		final JobID jobId,
  		final AllocationID allocationId,
  		final String targetAddress,
  		final ResourceManagerId resourceManagerId,
  		final Time timeout) {
  		// TODO: Filter invalid requests from the resource manager by using the instance/registration Id
  
  		log.info("Receive slot request {} for job {} from resource manager with leader id {}.",
  			allocationId, jobId, resourceManagerId);
  
  		try {
  			if (!isConnectedToResourceManager(resourceManagerId)) {
  				final String message = String.format("TaskManager is not connected to the resource manager %s.", resourceManagerId);
  				log.debug(message);
  				throw new TaskManagerException(message);
  			}
  
  			if (taskSlotTable.isSlotFree(slotId.getSlotNumber())) {
  				if (taskSlotTable.allocateSlot(slotId.getSlotNumber(), jobId, allocationId, taskManagerConfiguration.getTimeout())) {
  					log.info("Allocated slot for {}.", allocationId);
  				} else {
  					log.info("Could not allocate slot for {}.", allocationId);
  					throw new SlotAllocationException("Could not allocate slot.");
  				}
  			} else if (!taskSlotTable.isAllocated(slotId.getSlotNumber(), jobId, allocationId)) {
  				final String message = "The slot " + slotId + " has already been allocated for a different job.";
  
  				log.info(message);
  
  				final AllocationID allocationID = taskSlotTable.getCurrentAllocation(slotId.getSlotNumber());
  				throw new SlotOccupiedException(message, allocationID, taskSlotTable.getOwningJob(allocationID));
  			}
  
  			if (jobManagerTable.contains(jobId)) {
  				offerSlotsToJobManager(jobId);
  			} else {
  				try {
  					jobLeaderService.addJob(jobId, targetAddress);
  				} catch (Exception e) {
  					// free the allocated slot
  					try {
  						taskSlotTable.freeSlot(allocationId);
  					} catch (SlotNotFoundException slotNotFoundException) {
  						// slot no longer existent, this should actually never happen, because we've
  						// just allocated the slot. So let's fail hard in this case!
  						onFatalError(slotNotFoundException);
  					}
  
  					// release local state under the allocation id.
  					localStateStoresManager.releaseLocalStateForAllocationId(allocationId);
  
  					// sanity check
  					if (!taskSlotTable.isSlotFree(slotId.getSlotNumber())) {
  						onFatalError(new Exception("Could not free slot " + slotId));
  					}
  
  					throw new SlotAllocationException("Could not add job to job leader service.", e);
  				}
  			}
  		} catch (TaskManagerException taskManagerException) {
  			return FutureUtils.completedExceptionally(taskManagerException);
  		}
  
  		return CompletableFuture.completedFuture(Acknowledge.get());
  	}