@Override
    public Map<ExecutionAttemptID, ExecutionSlotAssignment> allocateSlotsFor(
            List<ExecutionAttemptID> executionAttemptIds) {

        final Map<ExecutionVertexID, ExecutionAttemptID> vertexIdToExecutionId = new HashMap<>();
        executionAttemptIds.forEach(
                executionId ->
                        vertexIdToExecutionId.put(executionId.getExecutionVertexId(), executionId));

        checkState(
                vertexIdToExecutionId.size() == executionAttemptIds.size(),
                "SlotSharingExecutionSlotAllocator does not support one execution vertex to have multiple concurrent executions");

        final List<ExecutionVertexID> vertexIds =
                executionAttemptIds.stream()
                        .map(ExecutionAttemptID::getExecutionVertexId)
                        .collect(Collectors.toList());

        return allocateSlotsForVertices(vertexIds).stream()
                .collect(
                        Collectors.toMap(
                                vertexAssignment ->
                                        vertexIdToExecutionId.get(
                                                vertexAssignment.getExecutionVertexId()),
                                vertexAssignment ->
                                        new ExecutionSlotAssignment(
                                                vertexIdToExecutionId.get(
                                                        vertexAssignment.getExecutionVertexId()),
                                                vertexAssignment.getLogicalSlotFuture())));
    }
@Test
    void failLogicalSlotsIfPhysicalSlotIsFailed() {
        final TestingPhysicalSlotRequestBulkChecker bulkChecker =
                new TestingPhysicalSlotRequestBulkChecker();
        AllocationContext context =
                AllocationContext.newBuilder()
                        .addGroup(EV1, EV2)
                        .withBulkChecker(bulkChecker)
                        .withPhysicalSlotProvider(
                                TestingPhysicalSlotProvider.createWithFailingPhysicalSlotCreation(
                                        new FlinkException("test failure")))
                        .build();

        final Map<ExecutionAttemptID, ExecutionSlotAssignment> allocatedSlots =
                context.allocateSlotsFor(EV1, EV2);

        for (ExecutionSlotAssignment allocatedSlot : allocatedSlots.values()) {
            assertThat(allocatedSlot.getLogicalSlotFuture()).isCompletedExceptionally();
        }

        assertThat(bulkChecker.getBulk().getPendingRequests()).isEmpty();

        final Set<SlotRequestId> requests = context.getSlotProvider().getRequests().keySet();
        assertThat(context.getSlotProvider().getCancellations().keySet()).isEqualTo(requests);
    }