class SlotSharingExecutionSlotAllocator implements ExecutionSlotAllocator {

	@Override
	public List<SlotExecutionVertexAssignment> allocateSlotsFor(
			List<ExecutionVertexSchedulingRequirements> executionVertexSchedulingRequirements) {
		List<ExecutionVertexID> executionVertexIds = executionVertexSchedulingRequirements
			.stream()
			.map(ExecutionVertexSchedulingRequirements::getExecutionVertexId)
			.collect(Collectors.toList());

		SharedSlotProfileRetriever sharedSlotProfileRetriever = sharedSlotProfileRetrieverFactory
			.createFromBulk(new HashSet<>(executionVertexIds));
		Map<ExecutionSlotSharingGroup, List<ExecutionVertexID>> executionsByGroup = executionVertexIds
			.stream()
			.collect(Collectors.groupingBy(slotSharingStrategy::getExecutionSlotSharingGroup));
		Map<ExecutionVertexID, SlotExecutionVertexAssignment> assignments =
			allocateLogicalSlotsFromSharedSlots(sharedSlotProfileRetriever, executionsByGroup);

		bulkChecker.schedulePendingRequestBulkTimeoutCheck(createBulk(executionsByGroup), allocationTimeout);

		return executionVertexIds.stream().map(assignments::get).collect(Collectors.toList());
	}

	  SlotSharingExecutionSlotAllocator(
			PhysicalSlotProvider slotProvider,
			boolean slotWillBeOccupiedIndefinitely,
			SlotSharingStrategy slotSharingStrategy,
			SharedSlotProfileRetrieverFactory sharedSlotProfileRetrieverFactory,
			PhysicalSlotRequestBulkChecker bulkChecker,
			Time allocationTimeout,
			Function<ExecutionVertexID, ResourceProfile> resourceProfileRetriever);

}