HadoopWorkflowUtils { public static void createAndRegisterSpringBatchJob(ApplicationContext context, final WorkflowArtifacts artifacts) throws SpringHadoopAdminWorkflowException { if (context == null) { logger.warn("root applicaton context is null"); throw new SpringHadoopAdminWorkflowException("root applicaton context is null"); } String workflowDescriptor = getWorkflowDescriptor(artifacts); logger.info("create spring batch job:" + workflowDescriptor + ", classloader:" + artifacts.getWorkflowClassLoader()); final FileSystemXmlApplicationContext ctx = new FileSystemXmlApplicationContext(); ctx.setClassLoader(artifacts.getWorkflowClassLoader()); ctx.setConfigLocation(HadoopWorkflowUtils.fileURLPrefix + workflowDescriptor); try { SimpleJob job = new SimpleJob(generateSpringBatchJobName(artifacts)); TaskletStep step = new TaskletStep(springHadoopTaskName); SimpleSpringHadoopTasklet tasklet = new SimpleSpringHadoopTasklet(); tasklet.setContext(ctx); step.setTasklet(tasklet); JobRepository jobRepository = context.getBean(JobRepository.class); DataSourceTransactionManager transactionManager = context.getBean("transactionManager", DataSourceTransactionManager.class); step.setTransactionManager(transactionManager); step.setJobRepository(jobRepository); step.afterPropertiesSet(); job.addStep(step); JobRegistry jobRegistry = context.getBean("jobRegistry", JobRegistry.class); job.setJobRepository(jobRepository); job.afterPropertiesSet(); JobFactory jobFactory = new ReferenceJobFactory(job); jobRegistry.register(jobFactory); } catch (Exception e) { logger.warn("create and register Spring Batch Job failed"); throw new SpringHadoopAdminWorkflowException("create and register Spring Batch Job failed", e); } }  static WorkflowArtifacts getWorkflowArtifacts(File workflowArtifactFolder); static String[] getWorkflowDescriptor(File workflowArtifactFolder); static boolean isValidWorkflow(List<FileInfo> files); static URL[] getWorkflowLibararies(File workflowArtifactFolder); static ClassLoader getWorkflowClassLoader(File workflowArtifactFolder); static boolean isSpringBatchJob(ApplicationContext context, WorkflowArtifacts artifacts); static void createAndRegisterSpringBatchJob(ApplicationContext context, final WorkflowArtifacts artifacts); static String generateSpringBatchJobName(final WorkflowArtifacts artifacts); static void unregisterSpringBatchJob(ApplicationContext context, String springBatchJobName); static String getWorkflowDescriptor(final WorkflowArtifacts artifacts); static void processAndRegisterWorkflow(File parentFolder, ApplicationContext context); static WorkflowArtifacts processUploadedFile(final File parentFolder); static void registerSpringBatchJob(WorkflowArtifacts artifacts, ApplicationContext context); static String fileURLPrefix; static String springHadoopJobPrefix; static String springHadoopTaskName; }
@Test public void testCreateAndRegisterSpringBatchJob() throws SpringHadoopAdminWorkflowException { File folder = new File( "src/test/resources/org/springframework/data/hadoop/admin/util/testWithoutSpringBatchJob"); WorkflowArtifacts artifacts = HadoopWorkflowUtils.getWorkflowArtifacts(folder); ApplicationContext context = new ClassPathXmlApplicationContext(new String[] { "classpath:org/springframework/data/hadoop/admin/env-context.xml", "classpath:org/springframework/data/hadoop/admin/data-source-context.xml", "classpath:org/springframework/data/hadoop/admin/execution-context.xml", }); HadoopWorkflowUtils.createAndRegisterSpringBatchJob(context, artifacts); JobRegistry jobRegistry = context.getBean("jobRegistry", JobRegistry.class); Collection<String> jobNames = jobRegistry.getJobNames(); Assert.assertEquals(1, jobNames.size()); Assert.assertTrue(jobNames.contains("spring-hadoop-job-context.xml")); }