@Override
    public CompletableFuture<JobStatus> getJobTerminationFuture() {
        return jobTerminationFuture;
    }
@Test
    void testInitialRequirementLowerBoundBeyondAvailableSlotsCausesImmediateFailure()
            throws Exception {
        final JobGraph jobGraph = createJobGraph();

        final DefaultDeclarativeSlotPool declarativeSlotPool =
                createDeclarativeSlotPool(jobGraph.getJobID());

        final int availableSlots = 1;
        JobResourceRequirements initialJobResourceRequirements =
                createRequirementsWithEqualLowerAndUpperParallelism(PARALLELISM);

        final AdaptiveScheduler scheduler =
                prepareSchedulerWithNoTimeouts(jobGraph, declarativeSlotPool)
                        .withConfigurationOverride(
                                conf -> {
                                    conf.set(
                                            JobManagerOptions.RESOURCE_WAIT_TIMEOUT,
                                            Duration.ofMillis(1));
                                    return conf;
                                })
                        .setJobResourceRequirements(initialJobResourceRequirements)
                        .build();

        final SubmissionBufferingTaskManagerGateway taskManagerGateway =
                createSubmissionBufferingTaskManagerGateway(PARALLELISM, scheduler);

        startJobWithSlotsMatchingParallelism(
                scheduler, declarativeSlotPool, taskManagerGateway, availableSlots);

        // the job will fail because not enough slots are available
        FlinkAssertions.assertThatFuture(scheduler.getJobTerminationFuture())
                .eventuallySucceeds()
                .isEqualTo(JobStatus.FAILED);
        // no task was ever submitted because we failed immediately
        assertThat(taskManagerGateway.submittedTasks).isEmpty();
    }