code: private CompletableFuture<LogicalSlot> allocateSharedSlot(
  		SlotRequestId slotRequestId,
  		ScheduledUnit scheduledUnit,
  		SlotProfile slotProfile,
  		boolean allowQueuedScheduling,
  		Time allocationTimeout) {
  		// allocate slot with slot sharing
  		final SlotSharingManager multiTaskSlotManager = slotSharingManagers.computeIfAbsent(
  			scheduledUnit.getSlotSharingGroupId(),
  			id -> new SlotSharingManager(
  				id,
  				slotPool,
  				this));
  
  		final SlotSharingManager.MultiTaskSlotLocality multiTaskSlotLocality;
  		try {
  			if (scheduledUnit.getCoLocationConstraint() != null) {
  				multiTaskSlotLocality = allocateCoLocatedMultiTaskSlot(
  					scheduledUnit.getCoLocationConstraint(),
  					multiTaskSlotManager,
  					slotProfile,
  					allowQueuedScheduling,
  					allocationTimeout);
  			} else {
  				multiTaskSlotLocality = allocateMultiTaskSlot(
  					scheduledUnit.getJobVertexId(),
  					multiTaskSlotManager,
  					slotProfile,
  					allowQueuedScheduling,
  					allocationTimeout);
  			}
  		} catch (NoResourceAvailableException noResourceException) {
  			return FutureUtils.completedExceptionally(noResourceException);
  		}
  
  		// sanity check
  		Preconditions.checkState(!multiTaskSlotLocality.getMultiTaskSlot().contains(scheduledUnit.getJobVertexId()));
  
  		final SlotSharingManager.SingleTaskSlot leaf = multiTaskSlotLocality.getMultiTaskSlot().allocateSingleTaskSlot(
  			slotRequestId,
  			scheduledUnit.getJobVertexId(),
  			multiTaskSlotLocality.getLocality());
  		return leaf.getLogicalSlotFuture();
  	}
docs: ------------------------------- slot sharing code