@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) { } 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) { } 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)); }
public synchronized void cleanup(Collection<Long> removedPlanIds) { for (long removedPlanId : removedPlanIds) { clean(removedPlanId); } }
WorkflowTracker { public synchronized void cleanup(Collection<Long> removedPlanIds) { for (long removedPlanId : removedPlanIds) { clean(removedPlanId); } } }
WorkflowTracker { public synchronized void cleanup(Collection<Long> removedPlanIds) { for (long removedPlanId : removedPlanIds) { clean(removedPlanId); } } WorkflowTracker(JobMaster jobMaster); }
WorkflowTracker { public synchronized void cleanup(Collection<Long> removedPlanIds) { for (long removedPlanId : removedPlanIds) { clean(removedPlanId); } } WorkflowTracker(JobMaster jobMaster); synchronized void run(WorkflowConfig workflowConfig, long jobId); synchronized boolean cancel(long jobId); WorkflowInfo getStatus(long jobId, boolean verbose); Collection<WorkflowInfo> getAllInfo(); Collection<Long> list(); synchronized void cleanup(Collection<Long> removedPlanIds); void onPlanStatusChange(PlanInfo planInfo); }
WorkflowTracker { public synchronized void cleanup(Collection<Long> removedPlanIds) { for (long removedPlanId : removedPlanIds) { clean(removedPlanId); } } WorkflowTracker(JobMaster jobMaster); synchronized void run(WorkflowConfig workflowConfig, long jobId); synchronized boolean cancel(long jobId); WorkflowInfo getStatus(long jobId, boolean verbose); Collection<WorkflowInfo> getAllInfo(); Collection<Long> list(); synchronized void cleanup(Collection<Long> removedPlanIds); void onPlanStatusChange(PlanInfo planInfo);  }