class DefaultExecutionSlotAllocator implements ExecutionSlotAllocator {

	@Override
	public Collection<SlotExecutionVertexAssignment> allocateSlotsFor(
			Collection<ExecutionVertexSchedulingRequirements> executionVertexSchedulingRequirements) {

		List<SlotExecutionVertexAssignment> slotExecutionVertexAssignments =
				new ArrayList<>(executionVertexSchedulingRequirements.size());

		Set<AllocationID> allPreviousAllocationIds = computeAllPriorAllocationIds(executionVertexSchedulingRequirements);

		for (ExecutionVertexSchedulingRequirements schedulingRequirements : executionVertexSchedulingRequirements) {
			final ExecutionVertexID executionVertexId = schedulingRequirements.getExecutionVertexId();
			final SlotRequestId slotRequestId = new SlotRequestId();
			final SlotSharingGroupId slotSharingGroupId = schedulingRequirements.getSlotSharingGroupId();

			LOG.debug("Allocate slot with id {} for execution {}", slotRequestId, executionVertexId);

			CompletableFuture<LogicalSlot> slotFuture = calculatePreferredLocations(
					executionVertexId,
					schedulingRequirements.getPreferredLocations(),
					inputsLocationsRetriever).thenCompose(
							(Collection<TaskManagerLocation> preferredLocations) ->
									slotProvider.allocateSlot(
											slotRequestId,
											new ScheduledUnit(
													executionVertexId.getJobVertexId(),
													slotSharingGroupId,
													schedulingRequirements.getCoLocationConstraint()),
											new SlotProfile(
													schedulingRequirements.getResourceProfile(),
													preferredLocations,
													Arrays.asList(schedulingRequirements.getPreviousAllocationId()),
													allPreviousAllocationIds),
											true,
											allocationTimeout));

			SlotExecutionVertexAssignment slotExecutionVertexAssignment =
					new SlotExecutionVertexAssignment(executionVertexId, slotFuture);
			// add to map first to avoid the future completed before added.
			pendingSlotAssignments.put(executionVertexId, slotExecutionVertexAssignment);

			slotFuture.whenComplete(
					(ignored, throwable) -> {
						pendingSlotAssignments.remove(executionVertexId);
						if (throwable != null) {
							slotProvider.cancelSlotRequest(slotRequestId, slotSharingGroupId, throwable);
						}
					});

			slotExecutionVertexAssignments.add(slotExecutionVertexAssignment);
		}

		return slotExecutionVertexAssignments;
	}

	public  DefaultExecutionSlotAllocator(
			SlotProvider slotProvider,
			InputsLocationsRetriever inputsLocationsRetriever,
			Time allocationTimeout);

}