@Override
    public CompletableFuture<Acknowledge> updateJobResourceRequirements(
            JobResourceRequirements jobResourceRequirements) {
        schedulerNG.updateJobResourceRequirements(jobResourceRequirements);
        return CompletableFuture.completedFuture(Acknowledge.get());
    }
@Test
    public void testSuccessfulResourceRequirementsUpdate() throws Exception {
        final CompletableFuture<JobResourceRequirements> schedulerUpdateFuture =
                new CompletableFuture<>();
        final TestingSchedulerNG scheduler =
                TestingSchedulerNG.newBuilder()
                        .setUpdateJobResourceRequirementsConsumer(schedulerUpdateFuture::complete)
                        .build();
        try (final JobMaster jobMaster =
                new JobMasterBuilder(jobGraph, rpcService)
                        .withConfiguration(configuration)
                        .withHighAvailabilityServices(haServices)
                        .withSlotPoolServiceSchedulerFactory(
                                DefaultSlotPoolServiceSchedulerFactory.create(
                                        TestingSlotPoolServiceBuilder.newBuilder(),
                                        new TestingSchedulerNGFactory(scheduler)))
                        .createJobMaster()) {
            jobMaster.start();
            final JobMasterGateway jobMasterGateway =
                    jobMaster.getSelfGateway(JobMasterGateway.class);

            final JobResourceRequirements.Builder jobResourceRequirementsBuilder =
                    JobResourceRequirements.newBuilder();
            for (JobVertex jobVertex : jobGraph.getVertices()) {
                jobResourceRequirementsBuilder.setParallelismForJobVertex(jobVertex.getID(), 1, 2);
            }

            final JobResourceRequirements newRequirements = jobResourceRequirementsBuilder.build();
            final CompletableFuture<Acknowledge> jobMasterUpdateFuture =
                    jobMasterGateway.updateJobResourceRequirements(newRequirements);

            assertThatFuture(jobMasterUpdateFuture).eventuallySucceeds();
            assertThatFuture(schedulerUpdateFuture).eventuallySucceeds().isEqualTo(newRequirements);
        }
    }