@Override
    public JobResourceRequirements requestJobResourceRequirements() {
        final JobResourceRequirements.Builder builder = JobResourceRequirements.newBuilder();
        for (JobInformation.VertexInformation vertex : jobInformation.getVertices()) {
            builder.setParallelismForJobVertex(
                    vertex.getJobVertexID(), vertex.getMinParallelism(), vertex.getParallelism());
        }
        return builder.build();
    }
@Test
    void testRequestDefaultResourceRequirementsInReactiveMode() throws Exception {
        final JobGraph jobGraph = createJobGraph();
        final Configuration configuration = new Configuration();
        configuration.set(JobManagerOptions.SCHEDULER_MODE, SchedulerExecutionMode.REACTIVE);
        final AdaptiveScheduler scheduler =
                new AdaptiveSchedulerBuilder(
                                jobGraph, mainThreadExecutor, EXECUTOR_RESOURCE.getExecutor())
                        .setJobMasterConfiguration(configuration)
                        .build();
        assertThat(scheduler.requestJobResourceRequirements())
                .isEqualTo(
                        JobResourceRequirements.newBuilder()
                                .setParallelismForJobVertex(
                                        JOB_VERTEX.getID(),
                                        1,
                                        SchedulerBase.getDefaultMaxParallelism(JOB_VERTEX))
                                .build());
    }