@Override
    public void updateJobResourceRequirements(JobResourceRequirements jobResourceRequirements) {
        if (settings.getExecutionMode() == SchedulerExecutionMode.REACTIVE) {
            throw new UnsupportedOperationException(
                    "Cannot change the parallelism of a job running in reactive mode.");
        }
        final Optional<VertexParallelismStore> maybeUpdateVertexParallelismStore =
                DefaultVertexParallelismStore.applyJobResourceRequirements(
                        jobInformation.getVertexParallelismStore(), jobResourceRequirements);
        if (maybeUpdateVertexParallelismStore.isPresent()) {
            this.jobInformation =
                    new JobGraphJobInformation(jobGraph, maybeUpdateVertexParallelismStore.get());
            declareDesiredResources();
            state.tryRun(
                    ResourceListener.class,
                    ResourceListener::onNewResourceRequirements,
                    "Current state does not react to desired parallelism changes.");
        }
    }
@Test
    void testRequirementLowerBoundDecreaseAfterResourceScarcityBelowAvailableSlots()
            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)
                        .setJobResourceRequirements(initialJobResourceRequirements)
                        .build();

        final SubmissionBufferingTaskManagerGateway taskManagerGateway =
                createSubmissionBufferingTaskManagerGateway(PARALLELISM, scheduler);

        startJobWithSlotsMatchingParallelism(
                scheduler, declarativeSlotPool, taskManagerGateway, availableSlots);

        // unlock job by decreasing the parallelism
        JobResourceRequirements newJobResourceRequirements =
                createRequirementsWithLowerAndUpperParallelism(availableSlots, PARALLELISM);
        singleThreadMainThreadExecutor.execute(
                () -> scheduler.updateJobResourceRequirements(newJobResourceRequirements));

        awaitJobReachingParallelism(taskManagerGateway, scheduler, availableSlots);
    }