docs: boolean offerSlot(
  			final TaskManagerLocation taskManagerLocation,
  			final TaskManagerGateway taskManagerGateway,
  			final SlotOffer slotOffer) {
  
  		componentMainThreadExecutor.assertRunningInMainThread();
  
  		// check if this TaskManager is valid
  		final ResourceID resourceID = taskManagerLocation.getResourceID();
  		final AllocationID allocationID = slotOffer.getAllocationId();
  
  		if (!registeredTaskManagers.contains(resourceID)) {
  			log.debug("Received outdated slot offering [{}] from unregistered TaskManager: {}",
  					slotOffer.getAllocationId(), taskManagerLocation);
  			return false;
  		}
  
  		// check whether we have already using this slot
  		AllocatedSlot existingSlot;
  		if ((existingSlot = allocatedSlots.get(allocationID)) != null ||
  			(existingSlot = availableSlots.get(allocationID)) != null) {
  
  			// we need to figure out if this is a repeated offer for the exact same slot,
  			// or another offer that comes from a different TaskManager after the ResourceManager
  			// re-tried the request
  
  			// we write this in terms of comparing slot IDs, because the Slot IDs are the identifiers of
  			// the actual slots on the TaskManagers
  			// Note: The slotOffer should have the SlotID
  			final SlotID existingSlotId = existingSlot.getSlotId();
  			final SlotID newSlotId = new SlotID(taskManagerLocation.getResourceID(), slotOffer.getSlotIndex());
  
  			if (existingSlotId.equals(newSlotId)) {
  				log.info("Received repeated offer for slot [{}]. Ignoring.", allocationID);
  
  				// return true here so that the sender will get a positive acknowledgement to the retry
  				// and mark the offering as a success
  				return true;
  			} else {
  				// the allocation has been fulfilled by another slot, reject the offer so the task executor
  				// will offer the slot to the resource manager
  				return false;
  			}
  		}
  
  		final AllocatedSlot allocatedSlot = new AllocatedSlot(
  			allocationID,
  			taskManagerLocation,
  			slotOffer.getSlotIndex(),
  			slotOffer.getResourceProfile(),
  			taskManagerGateway);
  
  		// check whether we have request waiting for this slot
  		PendingRequest pendingRequest = pendingRequests.removeKeyB(allocationID);
  		if (pendingRequest != null) {
  			// we were waiting for this!
  			allocatedSlots.add(pendingRequest.getSlotRequestId(), allocatedSlot);
  
  			if (!pendingRequest.getAllocatedSlotFuture().complete(allocatedSlot)) {
  				// we could not complete the pending slot future --> try to fulfill another pending request
  				allocatedSlots.remove(pendingRequest.getSlotRequestId());
  				tryFulfillSlotRequestOrMakeAvailable(allocatedSlot);
  			} else {
  				log.debug("Fulfilled slot request [{}] with allocated slot [{}].", pendingRequest.getSlotRequestId(), allocationID);
  			}
  		}
  		else {
  			// we were actually not waiting for this:
  			//   - could be that this request had been fulfilled
  			//   - we are receiving the slots from TaskManagers after becoming leaders
  			tryFulfillSlotRequestOrMakeAvailable(allocatedSlot);
  		}
  
  		// we accepted the request in any case. slot will be released after it idled for
  		// too long and timed out
  		return true;
  	}