class SlotPool extends RpcEndpoint implements SlotPoolGateway, AllocatedSlotActions {

	@Override
	public CompletableFuture<LogicalSlot> allocateSlot(
			SlotRequestId slotRequestId,
			ScheduledUnit task,
			SlotProfile slotProfile,
			boolean allowQueuedScheduling,
			Time allocationTimeout) {

		log.debug("Allocating slot with request {} for task execution {}", slotRequestId, task.getTaskToExecute());

		final SlotSharingGroupId slotSharingGroupId = task.getSlotSharingGroupId();

		if (slotSharingGroupId != null) {
			// allocate slot with slot sharing
			final SlotSharingManager multiTaskSlotManager = slotSharingManagers.computeIfAbsent(
				slotSharingGroupId,
				id -> new SlotSharingManager(
					id,
					this,
					providerAndOwner));

			final SlotSharingManager.MultiTaskSlotLocality multiTaskSlotLocality;

			try {
				if (task.getCoLocationConstraint() != null) {
					multiTaskSlotLocality = allocateCoLocatedMultiTaskSlot(
						task.getCoLocationConstraint(),
						multiTaskSlotManager,
						slotProfile,
						allowQueuedScheduling,
						allocationTimeout);
				} else {
					multiTaskSlotLocality = allocateMultiTaskSlot(
						task.getJobVertexId(),
						multiTaskSlotManager,
						slotProfile,
						allowQueuedScheduling,
						allocationTimeout);
				}
			} catch (NoResourceAvailableException noResourceException) {
				return FutureUtils.completedExceptionally(noResourceException);
			}

			// sanity check
			Preconditions.checkState(!multiTaskSlotLocality.getMultiTaskSlot().contains(task.getJobVertexId()));

			final SlotSharingManager.SingleTaskSlot leaf = multiTaskSlotLocality.getMultiTaskSlot().allocateSingleTaskSlot(
				slotRequestId,
				task.getJobVertexId(),
				multiTaskSlotLocality.getLocality());

			return leaf.getLogicalSlotFuture();
		} else {
			// request an allocated slot to assign a single logical slot to
			CompletableFuture<SlotAndLocality> slotAndLocalityFuture = requestAllocatedSlot(
				slotRequestId,
				slotProfile,
				allowQueuedScheduling,
				allocationTimeout);

			return slotAndLocalityFuture.thenApply(
				(SlotAndLocality slotAndLocality) -> {
					final AllocatedSlot allocatedSlot = slotAndLocality.getSlot();

					final SingleLogicalSlot singleTaskSlot = new SingleLogicalSlot(
						slotRequestId,
						allocatedSlot,
						null,
						slotAndLocality.getLocality(),
						providerAndOwner);

					if (allocatedSlot.tryAssignPayload(singleTaskSlot)) {
						return singleTaskSlot;
					} else {
						final FlinkException flinkException = new FlinkException("Could not assign payload to allocated slot " + allocatedSlot.getAllocationId() + '.');
						releaseSlot(slotRequestId, null, flinkException);
						throw new CompletionException(flinkException);
					}
				});
		}
	}

	@VisibleForTesting protected  SlotPool(RpcService rpcService, JobID jobId);
	public  SlotPool(
			RpcService rpcService,
			JobID jobId,
			Clock clock,
			Time rpcTimeout,
			Time idleSlotTimeout);

}