class AbstractExecutionSlotAllocator implements ExecutionSlotAllocator {

	SlotExecutionVertexAssignment createAndRegisterSlotExecutionVertexAssignment(
			final ExecutionVertexID executionVertexId,
			final CompletableFuture<LogicalSlot> logicalSlotFuture,
			final Consumer<Throwable> slotRequestFailureHandler) {

		final SlotExecutionVertexAssignment slotExecutionVertexAssignment =
			new SlotExecutionVertexAssignment(executionVertexId, logicalSlotFuture);

		// add to map first in case the slot future is already completed
		pendingSlotAssignments.put(executionVertexId, slotExecutionVertexAssignment);

		logicalSlotFuture.whenComplete(
			(ignored, throwable) -> {
				pendingSlotAssignments.remove(executionVertexId);
				if (throwable != null) {
					slotRequestFailureHandler.accept(throwable);
				}
			});

		return slotExecutionVertexAssignment;
	}

	  AbstractExecutionSlotAllocator(final PreferredLocationsRetriever preferredLocationsRetriever);

}