Implementation of JobLauncher that is tolerant on failures and accept…

[bojanv55]

…s retry template to do the actual restart/retry.

[fmbenhassine]

Hi @bojanv55 ,

According to BATCH-266 and BATCH-1329, this feature is out of scope and should be left to the user to implement either with an AOP advise or by wrapping the JobLauncher/JobOperator with a retry template rather then the other way around (which will introduce a lot of code duplication).

Here is a quick example:

import java.util.Map;

import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.JobParametersBuilder;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;

@Configuration
@EnableBatchProcessing
public class MyJob {

	private final JobBuilderFactory jobs;

	private final StepBuilderFactory steps;

	@Autowired
	public MyJob(JobBuilderFactory jobs, StepBuilderFactory steps) {
		this.jobs = jobs;
		this.steps = steps;
	}

	@Bean
	public Step step() {
		return steps.get("step")
				.tasklet((contribution, chunkContext) -> {
					Map<String, Object> jobParameters = chunkContext.getStepContext().getJobParameters();
					String name = (String) jobParameters.get("name");
					System.out.println("Hello " + name);
					throw new Exception("Boom!");
				})
				.build();
	}

	@Bean
	public Job job() {
		return jobs.get("job")
				.start(step())
				.build();
	}

	public static void main(String[] args) throws Throwable {
		RetryTemplate retryTemplate = new RetryTemplate();
		retryTemplate.setRetryPolicy(new SimpleRetryPolicy(3));

		ApplicationContext applicationContext = new AnnotationConfigApplicationContext(MyJob.class);
		JobLauncher jobLauncher = applicationContext.getBean(JobLauncher.class);
		Job job = applicationContext.getBean(Job.class);
		JobParameters jobParameters = new JobParametersBuilder().addString("name", "world").toJobParameters();

		retryTemplate.execute(retryContext -> {
			JobExecution jobExecution = jobLauncher.run(job, jobParameters);
			if (!jobExecution.getAllFailureExceptions().isEmpty()){
                                System.out.println("Job failed, retrying..");
				throw jobExecution.getAllFailureExceptions().iterator().next();
			}
			return jobExecution;
		});
	}

}

This will print:

Hello world
Job failed, retrying..
Hello world
Job failed, retrying..
Hello world
Job failed, retrying..

and then fail.

Do you agree?

Kr,
Mahmoud

[fmbenhassine]

@bojanv55 Any update on this?

[bojanv55]

Sure, this is another option. You can close this MR then.

[CKFVL]

@benas Will the retrytemplate work with itemreader, itemprocessor and itemwriter too. Because I see retry doesn't happen when I changed your code to use itemreader and itemwriter.

package com.nielsen.watch.tam.ncl.etl.config;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.support.ListItemReader;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;

import java.util.Arrays;

@Configuration
@EnableBatchProcessing
public class MyJob {
  private final JobBuilderFactory jobs;
  private final StepBuilderFactory steps;

  @Autowired
  public MyJob(JobBuilderFactory jobs, StepBuilderFactory steps) {
    this.jobs = jobs;
    this.steps = steps;
  }

  @Bean
  public Step step() {
    return steps.get("step")
      .<Integer, Integer>chunk(2)
      .reader(itemReader())
      .writer(itemWriter())
      //.faultTolerant()
      //.retryLimit(5)
      //.retry(Exception.class)
      .build();
  }

  @Bean
  public ItemReader<Integer> itemReader() {
    return new ListItemReader<>(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10));
  }

  @Bean
  public ItemWriter<Integer> itemWriter() {
    return items -> {
      for (Integer item : items) {
        System.out.println("item = " + item);
        if (item.equals(7)) {
          throw new Exception("Sevens are sometime nasty, let's retry them");
        }
      }
    };
  }
  @Bean
  public Job job() {
    return jobs.get("job")
      .start(step())
      .build();
  }

  public static void main(String[] args) throws Throwable {
    RetryTemplate retryTemplate = new RetryTemplate();
    retryTemplate.setRetryPolicy(new SimpleRetryPolicy(3));
    ApplicationContext applicationContext = new AnnotationConfigApplicationContext(MyJob.class);
    JobLauncher jobLauncher = applicationContext.getBean(JobLauncher.class);
    Job job = applicationContext.getBean(Job.class);
    //JobParameters jobParameters = new JobParametersBuilder().addString("name", "world").toJobParameters();
    retryTemplate.execute(retryContext -> {
      JobExecution jobExecution = jobLauncher.run(job, new JobParameters());
      if (!jobExecution.getAllFailureExceptions().isEmpty()) {
        System.out.println("Job failed, retrying..");
        throw jobExecution.getAllFailureExceptions().iterator().next();
      }
      return jobExecution;
    });  
   }
}

[fmbenhassine]

@CKFVL

Will the retrytemplate work with itemreader, itemprocessor and itemwriter too

No, the retry template I used in the example is external to the job definition. I could have used plain java code or any other library (like retry4j, failsafe or guava-retrying) to retry the job after a failure, but since we are a Spring shop, I used a RetryTemplate 😄 .

When you configure a retry policy on a chunk-oriented step through the FaultTolerantStepBuilder as in your example, a special retry template (a BatchRetryTemplate) will be used behind the scene, which is obviously different from the one I used to retry the job after a failure.