@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 testFailLogicalSlotIfPhysicalSlotIsFails() {
        final AllocationContext context =
                new AllocationContext(
                        TestingPhysicalSlotProvider.createWithFailingPhysicalSlotCreation(
                                new FlinkException("test failure")),
                        false);
        final CompletableFuture<LogicalSlot> slotFuture =
                context.allocateSlotsFor(EXECUTION_ATTEMPT_ID);

        assertThat(slotFuture).isCompletedExceptionally();
        assertThat(context.getSlotProvider().getCancellations().keySet())
                .isEqualTo(context.getSlotProvider().getRequests().keySet());
    }