docs: private void offerSlotsToJobManager(final JobID jobId) {
  		final JobManagerConnection jobManagerConnection = jobManagerTable.get(jobId);
  
  		if (jobManagerConnection == null) {
  			log.debug("There is no job manager connection to the leader of job {}.", jobId);
  		} else {
  			if (taskSlotTable.hasAllocatedSlots(jobId)) {
  				log.info("Offer reserved slots to the leader of job {}.", jobId);
  
  				final JobMasterGateway jobMasterGateway = jobManagerConnection.getJobManagerGateway();
  
  				final Iterator<TaskSlot> reservedSlotsIterator = taskSlotTable.getAllocatedSlots(jobId);
  				final JobMasterId jobMasterId = jobManagerConnection.getJobMasterId();
  
  				final Collection<SlotOffer> reservedSlots = new HashSet<>(2);
  
  				while (reservedSlotsIterator.hasNext()) {
  					SlotOffer offer = reservedSlotsIterator.next().generateSlotOffer();
  					reservedSlots.add(offer);
  				}
  
  				CompletableFuture<Collection<SlotOffer>> acceptedSlotsFuture = jobMasterGateway.offerSlots(
  					getResourceID(),
  					reservedSlots,
  					taskManagerConfiguration.getTimeout());
  
  				acceptedSlotsFuture.whenCompleteAsync(
  					(Iterable<SlotOffer> acceptedSlots, Throwable throwable) -> {
  						if (throwable != null) {
  							if (throwable instanceof TimeoutException) {
  								log.info("Slot offering to JobManager did not finish in time. Retrying the slot offering.");
  								// We ran into a timeout. Try again.
  								offerSlotsToJobManager(jobId);
  							} else {
  								log.warn("Slot offering to JobManager failed. Freeing the slots " +
  									"and returning them to the ResourceManager.", throwable);
  
  								// We encountered an exception. Free the slots and return them to the RM.
  								for (SlotOffer reservedSlot: reservedSlots) {
  									freeSlotInternal(reservedSlot.getAllocationId(), throwable);
  								}
  							}
  						} else {
  							// check if the response is still valid
  							if (isJobManagerConnectionValid(jobId, jobMasterId)) {
  								// mark accepted slots active
  								for (SlotOffer acceptedSlot : acceptedSlots) {
  									try {
  										if (!taskSlotTable.markSlotActive(acceptedSlot.getAllocationId())) {
  											// the slot is either free or releasing at the moment
  											final String message = "Could not mark slot " + jobId + " active.";
  											log.debug(message);
  											jobMasterGateway.failSlot(
  												getResourceID(),
  												acceptedSlot.getAllocationId(),
  												new FlinkException(message));
  										}
  									} catch (SlotNotFoundException e) {
  										final String message = "Could not mark slot " + jobId + " active.";
  										jobMasterGateway.failSlot(
  											getResourceID(),
  											acceptedSlot.getAllocationId(),
  											new FlinkException(message));
  									}
  
  									reservedSlots.remove(acceptedSlot);
  								}
  
  								final Exception e = new Exception("The slot was rejected by the JobManager.");
  
  								for (SlotOffer rejectedSlot : reservedSlots) {
  									freeSlotInternal(rejectedSlot.getAllocationId(), e);
  								}
  							} else {
  								// discard the response since there is a new leader for the job
  								log.debug("Discard offer slot response since there is a new leader " +
  									"for the job {}.", jobId);
  							}
  						}
  					},
  					getMainThreadExecutor());
  
  			} else {
  				log.debug("There are no unassigned slots for the job {}.", jobId);
  			}
  		}
  	}