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

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

        final AdaptiveScheduler scheduler =
                createSchedulerWithNoTimeouts(jobGraph, declarativeSlotPool);

        final int scaledUpParallelism = PARALLELISM * 2;

        final SubmissionBufferingTaskManagerGateway taskManagerGateway =
                createSubmissionBufferingTaskManagerGateway(scaledUpParallelism, scheduler);

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

        JobResourceRequirements newJobResourceRequirements =
                createRequirementsWithUpperParallelism(scaledUpParallelism);
        singleThreadMainThreadExecutor.execute(
                () -> {
                    // first update requirements as otherwise slots are rejected!
                    scheduler.updateJobResourceRequirements(newJobResourceRequirements);
                    offerSlots(
                            declarativeSlotPool,
                            createSlotOffersForResourceRequirements(
                                    ResourceCounter.withResource(
                                            ResourceProfile.UNKNOWN, PARALLELISM)),
                            taskManagerGateway);
                });

        awaitJobReachingParallelism(taskManagerGateway, scheduler, scaledUpParallelism);
    }