class DefaultExecutionSlotAllocator extends AbstractExecutionSlotAllocator {

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

		validateSchedulingRequirements(executionVertexSchedulingRequirements);

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

		Set<AllocationID> allPreviousAllocationIds = computeAllPriorAllocationIds(executionVertexSchedulingRequirements);

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

			final SlotRequestId slotRequestId = new SlotRequestId();

			final CompletableFuture<LogicalSlot> slotFuture = allocateSlot(
				schedulingRequirements,
				slotRequestId,
				allPreviousAllocationIds);

			final SlotExecutionVertexAssignment slotExecutionVertexAssignment =
				createAndRegisterSlotExecutionVertexAssignment(
					executionVertexId,
					slotFuture,
					throwable -> slotProviderStrategy.cancelSlotRequest(slotRequestId, slotSharingGroupId, throwable));

			slotExecutionVertexAssignments.add(slotExecutionVertexAssignment);
		}

		return slotExecutionVertexAssignments;
	}

	  DefaultExecutionSlotAllocator(
			final SlotProviderStrategy slotProviderStrategy,
			final PreferredLocationsRetriever preferredLocationsRetriever);

}