@Bean public Step step1() { return stepBuilderFactory.get("step1") .tasklet(new Tasklet() { @Override public RepeatStatus execute(StepContribution stepContribution, ChunkContext chunkContext) throws Exception { // get path of file in src/main/resources Path xmlDocPath = Paths.get(getFilePath()); // process the file to json String json = processXML2JSON(xmlDocPath); // insert json into mongodb insertToMongo(json); return RepeatStatus.FINISHED; } }).build(); }
public Tasklet getTasklet(BatchRunConfDto conf, String typeId, String scritpFile) { String cmd = Paths.get(conf.getBasePath(), scritpFile).toString(); switch (typeId) { case CMD_TYPE: cmd = "cmd /c " + cmd; return new ExecTasklet(cmd, execService, conf); case SHELL_TYPE: cmd = "sh -x " + cmd; return new ExecTasklet(cmd, execService, conf); case JAR_TYPE: cmd = "java -jar " + cmd; return new ExecTasklet(cmd, execService, conf); case BINARY_TYPE: return new ExecTasklet(cmd, execService, conf); case PROC_TYPE: return new ProcTasklet(scritpFile, jdbcTemplate); } return null; }
@Before public void init() throws Exception { this.context.register(BatchConfiguration.class); this.context.refresh(); JobRepository jobRepository = this.context.getBean(JobRepository.class); this.jobLauncher = this.context.getBean(JobLauncher.class); this.jobs = new JobBuilderFactory(jobRepository); PlatformTransactionManager transactionManager = this.context .getBean(PlatformTransactionManager.class); this.steps = new StepBuilderFactory(jobRepository, transactionManager); this.step = this.steps.get("step").tasklet(new Tasklet() { @Override public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { return null; } }).build(); this.job = this.jobs.get("job").start(this.step).build(); this.jobExplorer = this.context.getBean(JobExplorer.class); this.runner = new JobLauncherCommandLineRunner(this.jobLauncher, this.jobExplorer); this.context.getBean(BatchConfiguration.class).clear(); }
@Test public void retryFailedExecutionOnNonRestartableJob() throws Exception { this.job = this.jobs.get("job").preventRestart() .start(this.steps.get("step").tasklet(new Tasklet() { @Override public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { throw new RuntimeException("Planned"); } }).build()).incrementer(new RunIdIncrementer()).build(); this.runner.execute(this.job, new JobParameters()); this.runner.execute(this.job, new JobParameters()); // A failed job that is not restartable does not re-use the job params of // the last execution, but creates a new job instance when running it again. assertThat(this.jobExplorer.getJobInstances("job", 0, 100)).hasSize(2); }
@Test public void retryFailedExecutionWithNonIdentifyingParameters() throws Exception { this.job = this.jobs.get("job") .start(this.steps.get("step").tasklet(new Tasklet() { @Override public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { throw new RuntimeException("Planned"); } }).build()).incrementer(new RunIdIncrementer()).build(); JobParameters jobParameters = new JobParametersBuilder().addLong("id", 1L, false) .addLong("foo", 2L, false).toJobParameters(); this.runner.execute(this.job, jobParameters); this.runner.execute(this.job, jobParameters); assertThat(this.jobExplorer.getJobInstances("job", 0, 100)).hasSize(1); }
@Test public void retryFailedExecutionOnNonRestartableJob() throws Exception { this.job = this.jobs.get("job").preventRestart() .start(this.steps.get("step").tasklet(new Tasklet() { @Override public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { throw new RuntimeException("Planned"); } }).build()).incrementer(new RunIdIncrementer()).build(); this.runner.execute(this.job, new JobParameters()); this.runner.execute(this.job, new JobParameters()); // A failed job that is not restartable does not re-use the job params of // the last execution, but creates a new job instance when running it again. assertEquals(2, this.jobExplorer.getJobInstances("job", 0, 100).size()); }
@Test public void retryFailedExecutionWithNonIdentifyingParameters() throws Exception { this.job = this.jobs.get("job") .start(this.steps.get("step").tasklet(new Tasklet() { @Override public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { throw new RuntimeException("Planned"); } }).build()).incrementer(new RunIdIncrementer()).build(); JobParameters jobParameters = new JobParametersBuilder().addLong("id", 1L, false) .addLong("foo", 2L, false).toJobParameters(); this.runner.execute(this.job, jobParameters); this.runner.execute(this.job, jobParameters); assertEquals(1, this.jobExplorer.getJobInstances("job", 0, 100).size()); }
@Bean public Job importProductsJob(Tasklet decompressTasklet, ItemReader<Product> reader) { Step decompress = stepBuilders.get("decompress") .tasklet(decompressTasklet) .repository(jobRepository) .transactionManager(transactionManager) .build(); Step readWriteProducts = stepBuilders.get("readWriteProducts") .<Product, Product>chunk(3) .reader(reader) .writer(writer()) .faultTolerant() .skipLimit(5) .skip(FlatFileParseException.class) .build(); return jobBuilders.get("importProductsJob") .repository(jobRepository) .listener(loggerListener) .start(decompress) .next(readWriteProducts) .build(); }
private Step createTaskletStepWithListener(final String taskName, StepExecutionListener stepExecutionListener) { return this.steps.get(taskName) .tasklet(new Tasklet() { @Override public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { return RepeatStatus.FINISHED; } }) .transactionAttribute(getTransactionAttribute()) .listener(stepExecutionListener) .build(); }
private Step createTaskletStep(final String taskName) { return this.steps.get(taskName) .tasklet(new Tasklet() { @Override public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { return RepeatStatus.FINISHED; } }) .transactionAttribute(getTransactionAttribute()) .build(); }
public Step step2(){ return stepBuilderFactory.get("step2") .tasklet(new Tasklet(){ @Override public RepeatStatus execute(StepContribution stepContribution, ChunkContext chunkContext) throws Exception{ // checks if our collection exists Boolean doesexist = mongoTemplate.collectionExists("foo"); System.out.println("Status of collection returns :::::::::::::::::::::" + doesexist); // show all DBObjects in foo collection DBCursor alldocs = mongoTemplate.getCollection("foo").find(); List<DBObject> dbarray = alldocs.toArray(); System.out.println("list of db objects returns:::::::::::::::::::::" + dbarray); // execute the three methods we defined for querying the foo collection String result = doCollect(); String resultTwo = doCollectTwo(); String resultThree = doCollectThree(); System.out.println(" RESULT:::::::::::::::::::::" + result); System.out.println(" RESULT:::::::::::::::::::::" + resultTwo); System.out.println(" RESULT:::::::::::::::::::::" + resultThree); return RepeatStatus.FINISHED; } }).build(); }
@Bean public Step optionalStep() { return stepBuilderFactory.get("optionalStep") .tasklet(new Tasklet() { @Override public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { logger.info("IN OPTIONAL STEP ------------------------ */"); return RepeatStatus.FINISHED; } }) .build(); }
@Bean public Step githubStep1() throws Exception { return stepBuilderFactory.get("githubStep1") .tasklet(new Tasklet() { public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) { // GitHubDTO gitHubDTO = githubJobUI.getDummyStats(); GitHubDTO gitHubDTO = githubJobUI.getGitHubStats(); long currentStatId = githubJobUI.getCurrentGithubId(); gitHubDTO.setStatId(currentStatId); gitHubDTO.setStatDate(new Date()); githubJobUI.saveGithubStats(gitHubDTO); chunkContext .getStepContext() .getStepExecution() .getJobExecution() .getExecutionContext() .put("statId", currentStatId); logger.info("Working with GitHubDTO: " + gitHubDTO.toString()); return RepeatStatus.FINISHED; } }) .listener(githubPromotionListener()) .build(); }
@Bean public Tasklet tasklet() { return (contribution, chunkContext) -> { log.info("Executing tasklet step"); return RepeatStatus.FINISHED; }; }
@Test public void retryFailedExecution() throws Exception { this.job = this.jobs.get("job") .start(this.steps.get("step").tasklet(new Tasklet() { @Override public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { throw new RuntimeException("Planned"); } }).build()).incrementer(new RunIdIncrementer()).build(); this.runner.execute(this.job, new JobParameters()); this.runner.execute(this.job, new JobParameters()); assertThat(this.jobExplorer.getJobInstances("job", 0, 100)).hasSize(1); }
@Bean protected Tasklet tasklet() { return new Tasklet() { @Override public RepeatStatus execute(StepContribution contribution, ChunkContext context) { return RepeatStatus.FINISHED; } }; }
@Bean Tasklet tasklet() { return (a, b) -> { System.out.println("Hello, Spring Batch!!!"); return RepeatStatus.FINISHED; }; }
@Bean public Job job() throws Exception { Tasklet tasklet = (contribution, chunkContext) -> { Logger logger = LoggerFactory.getLogger(SimpleBatch.class); logger.info("Hello, Spring Batch!!"); return RepeatStatus.FINISHED; }; Step step = steps.get("step").tasklet(tasklet).build(); return jobs.get("job").start(step).build(); }
@Test public void retryFailedExecution() throws Exception { this.job = this.jobs.get("job") .start(this.steps.get("step").tasklet(new Tasklet() { @Override public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { throw new RuntimeException("Planned"); } }).build()).incrementer(new RunIdIncrementer()).build(); this.runner.execute(this.job, new JobParameters()); this.runner.execute(this.job, new JobParameters()); assertEquals(1, this.jobExplorer.getJobInstances("job", 0, 100).size()); }
@Bean Tasklet hiveInfluencersTasklet(HiveClientFactory hiveClientFactory) throws Exception { HiveTasklet hiveTasklet = new HiveTasklet(); hiveTasklet.setHiveClientFactory(hiveClientFactory); hiveTasklet.setScripts(hiveScripts()); return hiveTasklet; }
@Bean Tasklet resultsTasklet(@Qualifier("exportDataSource") final DataSource exportDataSource) { return new Tasklet() { @Override public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { JdbcTemplate jdbcTemplate = new JdbcTemplate(exportDataSource); System.out.println("Results:"); List<Map<String, Object>> results = jdbcTemplate.queryForList("select * from twitter_influencers"); for (Map<String, Object> r : results) { System.out.println(" " + r); } return RepeatStatus.FINISHED; } }; }
@Bean public Step step1() { return stepBuilderFactory.get("step1") .tasklet(new Tasklet() { public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) { return null; } }) .build(); }
@Bean public Step step1() { return this.stepBuilderFactory.get("step1") .tasklet(new Tasklet() { @Override public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { System.out.println("Tasklet has run"); return RepeatStatus.FINISHED; } }).build(); }
@Bean @StepScope public Tasklet workerTasklet( final @Value("#{stepExecutionContext['partitionNumber']}")Integer partitionNumber) { return new Tasklet() { @Override public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { System.out.println("This tasklet ran partition: " + partitionNumber); return RepeatStatus.FINISHED; } }; }
@Bean public Job job1() { return jobBuilderFactory.get("job1") .start(stepBuilderFactory.get("job1step1") .tasklet(new Tasklet() { @Override public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { logger.info("Job1 was run"); return RepeatStatus.FINISHED; } }) .build()) .build(); }
@Bean public Job job2() { return jobBuilderFactory.get("job2") .start(stepBuilderFactory.get("job2step1") .tasklet(new Tasklet() { @Override public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { logger.info("Job2 was run"); return RepeatStatus.FINISHED; } }) .build()) .build(); }
@Override public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException { registerJobExecutionEventListener(bean); if (bean instanceof AbstractStep) { registerStepExecutionEventListener(bean); if (bean instanceof TaskletStep) { TaskletStep taskletStep = (TaskletStep) bean; Tasklet tasklet = taskletStep.getTasklet(); registerChunkEventsListener(bean); if (tasklet instanceof ChunkOrientedTasklet) { Field chunkProviderField = ReflectionUtils.findField(ChunkOrientedTasklet.class, "chunkProvider"); ReflectionUtils.makeAccessible(chunkProviderField); SimpleChunkProvider chunkProvider = (SimpleChunkProvider) ReflectionUtils.getField(chunkProviderField, tasklet); Field chunkProcessorField = ReflectionUtils.findField(ChunkOrientedTasklet.class, "chunkProcessor"); ReflectionUtils.makeAccessible(chunkProcessorField); SimpleChunkProcessor chunkProcessor = (SimpleChunkProcessor) ReflectionUtils.getField(chunkProcessorField, tasklet); registerItemReadEvents(chunkProvider); registerSkipEvents(chunkProvider); registerItemProcessEvents(chunkProcessor); registerItemWriteEvents(chunkProcessor); registerSkipEvents(chunkProcessor); } } } return bean; }
@Bean public Step step1() { return stepBuilderFactory.get("step1").tasklet(new Tasklet() { @Override public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { System.out.println("Executed"); return RepeatStatus.FINISHED; } }).build(); }
@Bean public Job job() { return jobBuilderFactory.get("job") .start(stepBuilderFactory.get("step1").tasklet(new Tasklet() { @Override public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { System.out.println("Executed"); return RepeatStatus.FINISHED; } }).build()) .build(); }
@Bean public FactoryBean<Job> job() { return new FactoryBean<Job>() { @Override public Job getObject() throws Exception { return jobBuilderFactory.get("job") .start(stepBuilderFactory.get("step1").tasklet(new Tasklet() { @Override public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { System.out.println("Executed"); return RepeatStatus.FINISHED; } }).build()) .build(); } @Override public Class<?> getObjectType() { return Job.class; } @Override public boolean isSingleton() { return true; } }; }
@Bean public Job job1() { return jobBuilderFactory.get("job1") .start(stepBuilderFactory.get("job1step1").tasklet(new Tasklet() { @Override public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { System.out.println("Executed job1"); return RepeatStatus.FINISHED; } }).build()) .build(); }