public synchronized long run(JobConfig jobConfig)
      throws JobDoesNotExistException, ResourceExhaustedException {
    long jobId = getNewJobId();
    run(jobConfig, jobId);
    return jobId;
  }
@Test
  public void flowControl() throws Exception {
    try (MockedStatic<PlanCoordinator> mockStaticPlanCoordinator = mockPlanCoordinator()) {
      TestPlanConfig jobConfig = new TestPlanConfig("/test");
      for (long i = 0; i < TEST_JOB_MASTER_JOB_CAPACITY; i++) {
        mJobMaster.run(jobConfig);
      }
      try {
        mJobMaster.run(jobConfig);
        Assert.fail("should not be able to run more jobs than job master capacity");
      } catch (ResourceExhaustedException e) {
        Assert.assertEquals(ExceptionMessage.JOB_MASTER_FULL_CAPACITY
                .getMessage(Configuration.get(PropertyKey.JOB_MASTER_JOB_CAPACITY)),
            e.getMessage());
      }
    }
  }