@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 testPhysicalSlotReleaseLogicalSlots() throws ExecutionException, InterruptedException {
        AllocationContext context = AllocationContext.newBuilder().addGroup(EV1, EV2).build();
        Map<ExecutionAttemptID, ExecutionSlotAssignment> assignments =
                context.allocateSlotsFor(EV1, EV2);
        List<TestingPayload> payloads =
                assignments.values().stream()
                        .map(
                                assignment -> {
                                    TestingPayload payload = new TestingPayload();
                                    assignment
                                            .getLogicalSlotFuture()
                                            .thenAccept(
                                                    logicalSlot ->
                                                            logicalSlot.tryAssignPayload(payload));
                                    return payload;
                                })
                        .collect(Collectors.toList());
        SlotRequestId slotRequestId =
                context.getSlotProvider().getFirstRequestOrFail().getSlotRequestId();
        TestingPhysicalSlot physicalSlot = context.getSlotProvider().getFirstResponseOrFail().get();

        assertThat(payloads.stream().allMatch(payload -> payload.getTerminalStateFuture().isDone()))
                .isFalse();
        assertThat(physicalSlot.getPayload()).isNotNull();
        physicalSlot.getPayload().release(new Throwable());
        assertThat(payloads.stream().allMatch(payload -> payload.getTerminalStateFuture().isDone()))
                .isTrue();

        assertThat(context.getSlotProvider().getCancellations()).containsKey(slotRequestId);

        context.allocateSlotsFor(EV1, EV2);
        // there should be one more physical slot allocation, as the first allocation should be
        // removed after releasing all logical slots
        assertThat(context.getSlotProvider().getRequests()).hasSize(2);
    }