docs: private void allocateSlot(TaskManagerSlot taskManagerSlot, PendingSlotRequest pendingSlotRequest) {
  		Preconditions.checkState(taskManagerSlot.getState() == TaskManagerSlot.State.FREE);
  
  		TaskExecutorConnection taskExecutorConnection = taskManagerSlot.getTaskManagerConnection();
  		TaskExecutorGateway gateway = taskExecutorConnection.getTaskExecutorGateway();
  
  		final CompletableFuture<Acknowledge> completableFuture = new CompletableFuture<>();
  		final AllocationID allocationId = pendingSlotRequest.getAllocationId();
  		final SlotID slotId = taskManagerSlot.getSlotId();
  		final InstanceID instanceID = taskManagerSlot.getInstanceId();
  
  		taskManagerSlot.assignPendingSlotRequest(pendingSlotRequest);
  		pendingSlotRequest.setRequestFuture(completableFuture);
  
  		returnPendingTaskManagerSlotIfAssigned(pendingSlotRequest);
  
  		TaskManagerRegistration taskManagerRegistration = taskManagerRegistrations.get(instanceID);
  
  		if (taskManagerRegistration == null) {
  			throw new IllegalStateException("Could not find a registered task manager for instance id " +
  				instanceID + '.');
  		}
  
  		taskManagerRegistration.markUsed();
  
  		// RPC call to the task manager
  		CompletableFuture<Acknowledge> requestFuture = gateway.requestSlot(
  			slotId,
  			pendingSlotRequest.getJobId(),
  			allocationId,
  			pendingSlotRequest.getTargetAddress(),
  			resourceManagerId,
  			taskManagerRequestTimeout);
  
  		requestFuture.whenComplete(
  			(Acknowledge acknowledge, Throwable throwable) -> {
  				if (acknowledge != null) {
  					completableFuture.complete(acknowledge);
  				} else {
  					completableFuture.completeExceptionally(throwable);
  				}
  			});
  
  		completableFuture.whenCompleteAsync(
  			(Acknowledge acknowledge, Throwable throwable) -> {
  				try {
  					if (acknowledge != null) {
  						updateSlot(slotId, allocationId, pendingSlotRequest.getJobId());
  					} else {
  						if (throwable instanceof SlotOccupiedException) {
  							SlotOccupiedException exception = (SlotOccupiedException) throwable;
  							updateSlot(slotId, exception.getAllocationId(), exception.getJobId());
  						} else {
  							removeSlotRequestFromSlot(slotId, allocationId);
  						}
  
  						if (!(throwable instanceof CancellationException)) {
  							handleFailedSlotRequest(slotId, allocationId, throwable);
  						} else {
  							LOG.debug("Slot allocation request {} has been cancelled.", allocationId, throwable);
  						}
  					}
  				} catch (Exception e) {
  					LOG.error("Error while completing the slot allocation.", e);
  				}
  			},
  			mainThreadExecutor);
  	}