@Override
    public Map<ExecutionAttemptID, ExecutionSlotAssignment> allocateSlotsFor(
            List<ExecutionAttemptID> executionAttemptIds) {
        Map<ExecutionAttemptID, ExecutionSlotAssignment> result = new HashMap<>();

        Map<SlotRequestId, ExecutionAttemptID> remainingExecutionsToSlotRequest =
                new HashMap<>(executionAttemptIds.size());
        List<PhysicalSlotRequest> physicalSlotRequests =
                new ArrayList<>(executionAttemptIds.size());

        for (ExecutionAttemptID executionAttemptId : executionAttemptIds) {
            if (requestedPhysicalSlots.containsKeyA(executionAttemptId)) {
                result.put(
                        executionAttemptId,
                        new ExecutionSlotAssignment(
                                executionAttemptId,
                                requestedPhysicalSlots.getValueByKeyA(executionAttemptId)));
            } else {
                final SlotRequestId slotRequestId = new SlotRequestId();
                final ResourceProfile resourceProfile =
                        resourceProfileRetriever.apply(executionAttemptId);
                Collection<TaskManagerLocation> preferredLocations =
                        preferredLocationsRetriever.getPreferredLocations(
                                executionAttemptId.getExecutionVertexId(), Collections.emptySet());
                final SlotProfile slotProfile =
                        SlotProfile.priorAllocation(
                                resourceProfile,
                                resourceProfile,
                                preferredLocations,
                                Collections.emptyList(),
                                Collections.emptySet());
                final PhysicalSlotRequest request =
                        new PhysicalSlotRequest(
                                slotRequestId, slotProfile, slotWillBeOccupiedIndefinitely);
                physicalSlotRequests.add(request);
                remainingExecutionsToSlotRequest.put(slotRequestId, executionAttemptId);
            }
        }

        result.putAll(
                allocatePhysicalSlotsFor(remainingExecutionsToSlotRequest, physicalSlotRequests));
        return result;
    }
@Test
    void testPhysicalSlotReleasesLogicalSlots() throws Exception {
        final AllocationContext context = new AllocationContext();
        final CompletableFuture<LogicalSlot> slotFuture =
                context.allocateSlotsFor(EXECUTION_ATTEMPT_ID);
        final TestingPayload payload = new TestingPayload();
        slotFuture.thenAccept(logicalSlot -> logicalSlot.tryAssignPayload(payload));
        final SlotRequestId slotRequestId =
                context.getSlotProvider().getFirstRequestOrFail().getSlotRequestId();
        final TestingPhysicalSlot physicalSlot =
                context.getSlotProvider().getFirstResponseOrFail().get();

        assertThat(payload.getTerminalStateFuture()).isNotDone();
        assertThat(physicalSlot.getPayload()).isNotNull();

        physicalSlot.getPayload().release(new Throwable());
        assertThat(payload.getTerminalStateFuture()).isDone();
        assertThat(context.getSlotProvider().getCancellations()).containsKey(slotRequestId);

        context.allocateSlotsFor(EXECUTION_ATTEMPT_ID);
        // 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);
    }