JobManager { public long runSubJobForNodes(String jobName, List<JobParameters> jobParametersList, String clusterName, ClusterStatus successStatus, ClusterStatus failStatus) throws Exception { return createAndLaunchJobWithSubJob(clusterName, jobName, jobParametersList, successStatus, failStatus); }  long runJob(String jobName, JobParameters param); long runSubJobForNodes(String jobName,
         List<JobParameters> jobParametersList, String clusterName,
         ClusterStatus successStatus,
         ClusterStatus failStatus); long runJobWithSubJob(String jobName, JobParameters param,
         String subJobName, ClusterStatus successStatus,
         ClusterStatus failStatus); boolean stopJobExecution(long jobExecutionId); long restartJobExecution(long jobExecutionId); TaskRead getJobExecutionStatus(long jobExecutionId); List<TaskRead> getLatestTaskForExistedClusters(); TaskRead waitJobExecution(long jobExecutionId, long timeoutMs); JobRepository getJobRepository(); void setJobRepository(JobRepository jobRepository); JobLauncher getJobLauncher(); void setJobLauncher(JobLauncher jobLauncher); JobExplorer getJobExplorer(); void setJobExplorer(JobExplorer jobExplorer); JobOperator getJobOperator(); JobParametersExtractor getJobParametersExtractor(); void setJobParametersExtractor(
         JobParametersExtractor jobParametersExtractor); void setJobOperator(JobOperator jobOperator); JobRegistry getJobRegistry(); void setJobRegistry(JobRegistry jobRegistry); JobExecutionStatusHolder getJobExecutionStatusHolder(); void setJobExecutionStatusHolder(
         JobExecutionStatusHolder jobExecutionStatusHolder); JobExecutionStatusHolder getMainJobExecutionStatusHolder(); void setMainJobExecutionStatusHolder(
         JobExecutionStatusHolder subJobExecutionStatusHolder); IClusterEntityManager getClusterEntityMgr(); void setClusterEntityMgr(IClusterEntityManager clusterEntityMgr); JobExecutionListener getMainJobExecutionListener(); void setMainJobExecutionListener(
         JobExecutionListener mainJobExecutionListener);  }
@Test public void testRunSubJobForNodes() throws Exception { ApplicationContext context = new ClassPathXmlApplicationContext("spring/*.xml"); JobManager jobManager = (JobManager) context.getBean("jobManager"); List<JobParameters> jobParametersList = new ArrayList<JobParameters>(); JobParametersBuilder parametersBuilder = new JobParametersBuilder(); String clusterName = "test-cluster"; JobParameters nodeParameters = parametersBuilder .addString(JobConstants.SUB_JOB_NODE_NAME, "node1") .addString("cluster.name", clusterName).toJobParameters(); jobParametersList.add(nodeParameters); nodeParameters = parametersBuilder .addString(JobConstants.SUB_JOB_NODE_NAME, "node2") .addString("cluster.name", clusterName).toJobParameters(); jobParametersList.add(nodeParameters); nodeParameters = parametersBuilder .addString(JobConstants.SUB_JOB_NODE_NAME, "node3") .addString("cluster.name", clusterName).toJobParameters(); jobParametersList.add(nodeParameters); String subJobName = "simpleJob"; long jobExecutionId = jobManager.runSubJobForNodes(subJobName, jobParametersList, clusterName, ClusterStatus.RUNNING, ClusterStatus.ERROR); while (true) { Thread.sleep(50); TaskRead tr = jobManager.getJobExecutionStatus(jobExecutionId); logger.info("======= main job progress = " + tr.getProgress()); if (TaskRead.Status.COMPLETED.equals(tr.getStatus()) || TaskRead.Status.FAILED.equals(tr.getStatus())) { logger.info("===========COMPLETED/FAILED============"); for (TaskRead.NodeStatus success : tr.getSucceedNodes()) { logger.info("success node: " + success); } for (TaskRead.NodeStatus fail : tr.getFailNodes()) { logger.info("fail node: " + fail); } break; } } }