@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 testRequirementDecreaseTriggersScaleDown() throws Exception {
        final JobGraph jobGraph = createJobGraph();

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

        final AdaptiveScheduler scheduler =
                createSchedulerWithNoTimeouts(jobGraph, declarativeSlotPool);

        final SubmissionBufferingTaskManagerGateway taskManagerGateway =
                createSubmissionBufferingTaskManagerGateway(PARALLELISM, scheduler);

        startJobWithSlotsMatchingParallelism(
                scheduler, declarativeSlotPool, taskManagerGateway, PARALLELISM);
        awaitJobReachingParallelism(taskManagerGateway, scheduler, PARALLELISM);

        int scaledDownParallelism = PARALLELISM - 1;
        JobResourceRequirements newJobResourceRequirements =
                createRequirementsWithUpperParallelism(scaledDownParallelism);
        singleThreadMainThreadExecutor.execute(
                () -> scheduler.updateJobResourceRequirements(newJobResourceRequirements));

        awaitJobReachingParallelism(taskManagerGateway, scheduler, scaledDownParallelism);
    }