docs: private SlotSharingManager.MultiTaskSlotLocality allocateCoLocatedMultiTaskSlot(
  		CoLocationConstraint coLocationConstraint,
  		SlotSharingManager multiTaskSlotManager,
  		SlotProfile slotProfile,
  		boolean allowQueuedScheduling,
  		Time allocationTimeout) throws NoResourceAvailableException {
  		final SlotRequestId coLocationSlotRequestId = coLocationConstraint.getSlotRequestId();
  
  		if (coLocationSlotRequestId != null) {
  			// we have a slot assigned --> try to retrieve it
  			final SlotSharingManager.TaskSlot taskSlot = multiTaskSlotManager.getTaskSlot(coLocationSlotRequestId);
  
  			if (taskSlot != null) {
  				Preconditions.checkState(taskSlot instanceof SlotSharingManager.MultiTaskSlot);
  				return SlotSharingManager.MultiTaskSlotLocality.of(((SlotSharingManager.MultiTaskSlot) taskSlot), Locality.LOCAL);
  			} else {
  				// the slot may have been cancelled in the mean time
  				coLocationConstraint.setSlotRequestId(null);
  			}
  		}
  
  		if (coLocationConstraint.isAssigned()) {
  			// refine the preferred locations of the slot profile
  			slotProfile = new SlotProfile(
  				slotProfile.getResourceProfile(),
  				Collections.singleton(coLocationConstraint.getLocation()),
  				slotProfile.getPreferredAllocations());
  		}
  
  		// get a new multi task slot
  		SlotSharingManager.MultiTaskSlotLocality multiTaskSlotLocality = allocateMultiTaskSlot(
  			coLocationConstraint.getGroupId(),
  			multiTaskSlotManager,
  			slotProfile,
  			allowQueuedScheduling,
  			allocationTimeout);
  
  		// check whether we fulfill the co-location constraint
  		if (coLocationConstraint.isAssigned() && multiTaskSlotLocality.getLocality() != Locality.LOCAL) {
  			multiTaskSlotLocality.getMultiTaskSlot().release(
  				new FlinkException("Multi task slot is not local and, thus, does not fulfill the co-location constraint."));
  
  			throw new NoResourceAvailableException("Could not allocate a local multi task slot for the " +
  				"co location constraint " + coLocationConstraint + '.');
  		}
  
  		final SlotRequestId slotRequestId = new SlotRequestId();
  		final SlotSharingManager.MultiTaskSlot coLocationSlot =
  			multiTaskSlotLocality.getMultiTaskSlot().allocateMultiTaskSlot(
  				slotRequestId,
  				coLocationConstraint.getGroupId());
  
  		// mark the requested slot as co-located slot for other co-located tasks
  		coLocationConstraint.setSlotRequestId(slotRequestId);
  
  		// lock the co-location constraint once we have obtained the allocated slot
  		coLocationSlot.getSlotContextFuture().whenComplete(
  			(SlotContext slotContext, Throwable throwable) -> {
  				if (throwable == null) {
  					// check whether we are still assigned to the co-location constraint
  					if (Objects.equals(coLocationConstraint.getSlotRequestId(), slotRequestId)) {
  						coLocationConstraint.lockLocation(slotContext.getTaskManagerLocation());
  					} else {
  						log.debug("Failed to lock colocation constraint {} because assigned slot " +
  								"request {} differs from fulfilled slot request {}.",
  							coLocationConstraint.getGroupId(),
  							coLocationConstraint.getSlotRequestId(),
  							slotRequestId);
  					}
  				} else {
  					log.debug("Failed to lock colocation constraint {} because the slot " +
  							"allocation for slot request {} failed.",
  						coLocationConstraint.getGroupId(),
  						coLocationConstraint.getSlotRequestId(),
  						throwable);
  				}
  			});
  
  		return SlotSharingManager.MultiTaskSlotLocality.of(coLocationSlot, multiTaskSlotLocality.getLocality());
  	}