public synchronized void cleanup(Collection<Long> removedPlanIds) {
    for (long removedPlanId : removedPlanIds) {
      clean(removedPlanId);
    }
  }
@Test
  public void testCleanup() throws Exception {
    SleepJobConfig jobConfig = new SleepJobConfig(1);
    mPlanTracker.run(jobConfig, mCommandManager, mMockJobServerContext, mWorkers, 1);

    jobConfig = new SleepJobConfig(1);
    mPlanTracker.run(jobConfig, mCommandManager, mMockJobServerContext, mWorkers, 2);

    jobConfig = new SleepJobConfig(1);
    mPlanTracker.run(jobConfig, mCommandManager, mMockJobServerContext, mWorkers, 3);

    doAnswer(invocation -> {
      PlanConfig config = invocation.getArgument(0, PlanConfig.class);
      long jobId = invocation.getArgument(1, Long.class);

      mPlanTracker.run(config, mCommandManager, mMockJobServerContext, mWorkers, jobId);
      return null;
    }).when(mMockJobMaster).run(any(PlanConfig.class), any(Long.class));

    ArrayList<JobConfig> jobs = Lists.newArrayList();

    SleepJobConfig child1 = new SleepJobConfig(1);
    SleepJobConfig child2 = new SleepJobConfig(2);
    jobs.add(child1);
    jobs.add(child2);

    CompositeConfig config = new CompositeConfig(jobs, false);

    mWorkflowTracker.run(config, 0);

    try {
      mPlanTracker.run(new SleepJobConfig(1), mCommandManager, mMockJobServerContext, mWorkers, 4);
      fail();
    } catch (ResourceExhaustedException e) {
      // Should fail
    }

    mPlanTracker.coordinators().stream().filter(coordinator -> coordinator.getJobId() == 100)
        .findFirst().get().setJobAsFailed("TestError", "failed");

    mPlanTracker.run(new SleepJobConfig(1), mCommandManager, mMockJobServerContext, mWorkers, 4);

    assertNotNull(mWorkflowTracker.getStatus(0, true));

    try {
      mPlanTracker.run(new SleepJobConfig(1), mCommandManager, mMockJobServerContext, mWorkers, 5);
      fail();
    } catch (ResourceExhaustedException e) {
      // Should fail
    }

    mPlanTracker.coordinators().stream().filter(coordinator -> coordinator.getJobId() == 101)
        .findFirst().get().setJobAsFailed("TestError", "failed");

    mPlanTracker.run(new SleepJobConfig(1), mCommandManager, mMockJobServerContext, mWorkers, 5);

    assertNull(mWorkflowTracker.getStatus(100, true));
  }