PlanStateManager { public PlanState.State waitUntilFinish() { return waitUntilFinish(Long.MAX_VALUE, TimeUnit.MILLISECONDS); } @Inject private  PlanStateManager(@Parameter(JobConf.DAGDirectory.class) final String dagDirectory); static PlanStateManager newInstance(final String dagDirectory); void setMetricStore(final MetricStore metricStore); synchronized void updatePlan(final PhysicalPlan physicalPlanToUpdate,
                                      final int maxScheduleAttemptToSet); synchronized List<String> getTaskAttemptsToSchedule(final String stageId); synchronized Set<String> getAllTaskAttemptsOfStage(final String stageId); synchronized Map<String, Long> getExecutingTaskToRunningTimeMs(final String stageId); synchronized List<Long> getCompletedTaskTimeListMs(final String stageId); synchronized boolean setNumOfClones(final String stageId, final int taskIndex, final int numOfClones); synchronized void onTaskStateChanged(final String taskId, final TaskState.State newTaskState); PlanState.State waitUntilFinish(); PlanState.State waitUntilFinish(final long timeout, final TimeUnit unit); @VisibleForTesting synchronized Map<String, TaskState.State> getAllTaskAttemptIdsToItsState(); synchronized boolean isPlanDone(); synchronized String getPlanId(); synchronized PlanState.State getPlanState(); synchronized StageState.State getStageState(final String stageId); synchronized TaskState.State getTaskState(final String taskId); synchronized PhysicalPlan getPhysicalPlan(); int getMaxScheduleAttempt(); synchronized boolean isInitialized(); void storeJSON(final String suffix); @Override synchronized String toString();  }
@Test(timeout = 4000) public void testWaitUntilFinish() throws Exception { final PhysicalPlan physicalPlan = TestPlanGenerator.generatePhysicalPlan(TestPlanGenerator.PlanType.TwoVerticesJoined, false); planStateManager.updatePlan(physicalPlan, MAX_SCHEDULE_ATTEMPT); assertFalse(planStateManager.isPlanDone()); final PlanState.State executingState = planStateManager.waitUntilFinish(100, TimeUnit.MILLISECONDS); assertEquals(PlanState.State.EXECUTING, executingState); final List<String> tasks = physicalPlan.getStageDAG().getTopologicalSort().stream() .flatMap(stage -> planStateManager.getTaskAttemptsToSchedule(stage.getId()).stream()) .collect(Collectors.toList()); tasks.forEach(taskId -> planStateManager.onTaskStateChanged(taskId, TaskState.State.EXECUTING)); tasks.forEach(taskId -> planStateManager.onTaskStateChanged(taskId, TaskState.State.COMPLETE)); final PlanState.State completedState = planStateManager.waitUntilFinish(); assertEquals(PlanState.State.COMPLETE, completedState); }