How to launch Spring Batch Job Asynchronously

Question

I have followed the spring batch doc and couldn't get my job running Asynchronously.

So I am running the Job from a web container and the job will be triggered via a REST end point.

I wanted to get the JobInstance ID to pass it in response before completing the whole job. So they can check the status of the job later with the JobInstance ID instead of waiting. But I couldn't get it work. Below is the sample code I tried. Please let me know what am I missing or wrong.

BatchConfig to make Async JobLauncher

@Configuration
public class BatchConfig {

    @Autowired
    JobRepository jobRepository;


    @Bean
    public JobLauncher simpleJobLauncher() throws Exception {
        SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
        jobLauncher.setJobRepository(jobRepository);
        jobLauncher.setTaskExecutor(new SimpleAsyncTaskExecutor());
        jobLauncher.afterPropertiesSet();
        return jobLauncher;
    }
}

Controller

@Autowired
JobLauncher jobLauncher;

@RequestMapping(value="/trigger-job", method = RequestMethod.GET)
public Long workHard() throws Exception {
    JobParameters jobParameters = new JobParametersBuilder().
            addLong("time", System.currentTimeMillis())
            .toJobParameters();
    JobExecution jobExecution = jobLauncher.run(batchComponent.customJob("paramhere"), jobParameters);
    System.out.println(jobExecution.getJobInstance().getInstanceId());
    System.out.println("OK RESPONSE");
    return jobExecution.getJobInstance().getInstanceId();
}

And JobBuilder as component

@Component
public class BatchComponent {

    @Autowired
    private JobBuilderFactory jobBuilderFactory;

    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    public Job customJob(String someParam) throws Exception {
        return jobBuilderFactory.get("personProcessor")
                .incrementer(new RunIdIncrementer()).listener(listener())
                .flow(personPorcessStep(someParam)).end().build();
    }


    private Step personPorcessStep(String someParam) throws Exception {
        return stepBuilderFactory.get("personProcessStep").<PersonInput, PersonOutput>chunk(1)
                .reader(new PersonReader(someParam)).faultTolerant().
                        skipPolicy(new DataDuplicateSkipper()).processor(new PersonProcessor())
                .writer(new PersonWriter()).build();
    }


    private JobExecutionListener listener() {
        return new PersonJobCompletionListener();
    }

    private class PersonInput {
        String firstName;

        public PersonInput(String firstName) {
            this.firstName = firstName;
        }

        public String getFirstName() {
            return firstName;
        }

        public void setFirstName(String firstName) {
            this.firstName = firstName;
        }
    }

    private class PersonOutput {
        String firstName;

        public String getFirstName() {
            return firstName;
        }

        public void setFirstName(String firstName) {
            this.firstName = firstName;
        }
    }

    public class PersonReader implements ItemReader<PersonInput> {
        private List<PersonInput> items;
        private int count = 0;

        public PersonReader(String someParam) throws InterruptedException {
            Thread.sleep(10000L); //to simulate processing
            //manipulate and provide data in the read method
            //just for testing i have given some dummy example
            items = new ArrayList<PersonInput>();
            PersonInput pi = new PersonInput("john");
            items.add(pi);
        }

        @Override
        public PersonInput read() {
            if (count < items.size()) {
                return items.get(count++);
            }
            return null;
        }
    }


    public class DataDuplicateSkipper implements SkipPolicy {

        @Override
        public boolean shouldSkip(Throwable exception, int skipCount) throws SkipLimitExceededException {
            if (exception instanceof DataIntegrityViolationException) {
                return true;
            }
            return true;
        }
    }


    private class PersonProcessor implements ItemProcessor<PersonInput, PersonOutput> {

        @Override
        public PersonOutput process(PersonInput item) throws Exception {
            return null;
        }
    }

    private class PersonWriter implements org.springframework.batch.item.ItemWriter<PersonOutput> {
        @Override
        public void write(List<? extends PersonOutput> results) throws Exception {
            return;
        }
    }

    private class PersonJobCompletionListener implements JobExecutionListener {
        public PersonJobCompletionListener() {
        }

        @Override
        public void beforeJob(JobExecution jobExecution) {

        }

        @Override
        public void afterJob(JobExecution jobExecution) {
            System.out.println("JOB COMPLETED");
        }
    }
}

Main Function

@SpringBootApplication
@EnableBatchProcessing
@EnableScheduling
@EnableAsync
public class SpringBatchTestApplication {

    public static void main(String[] args) {
        SpringApplication.run(SpringBatchTestApplication.class, args);
    }
}

I am using annotation based configurations and use gradle with the below batch package.

compile('org.springframework.boot:spring-boot-starter-batch')

Please let me know if some more info needed. I couldn't find any example to run this common use case.

Thanks for you time.

Answer 1

Create a new java file named BatchParallelProcessingConfiguration.java and add the following code,

import javax.sql.DataSource;

import org.springframework.batch.core.configuration.BatchConfigurationException;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.support.DefaultBatchConfiguration;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.launch.support.TaskExecutorJobLauncher;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.batch.BatchAutoConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;

@Configuration
// @EnableBatchProcessing
public class BatchParallelProcessingConfiguration {

    @Bean
    public TaskExecutor taskExecutor(){
        return new SimpleAsyncTaskExecutor();
    }

    @Bean
    public JobLauncher jobLauncher(){
        TaskExecutorJobLauncher jobLauncher = new TaskExecutorJobLauncher();
        jobLauncher.setTaskExecutor(taskExecutor());
        return jobLauncher;
    }
}

In order to override the default jobLauncher bean provider by the spring boot we need to add the below property in application.properties.

spring.main.allow-bean-definition-overriding=true

And that's it now your job can be run asynchronously and parallelly.

Answer 2

I know that this is an old question but I post this answer anyway for future users.

After reviewing your code I can't tell why you have this problem, but I can suggest you to use a Qualifier annotation plus use the ThreadPoolTaskExecutor like so and see if it solve your problem.

You may also check this tutorial: Asynchronous Spring Batch Job Processing for more details. It will help you configure a spring batch job asynchronously. This tutorial was written by me.

@Configuration
public class BatchConfig {
 
 @Autowired
 private JobRepository jobRepository;
 
 @Bean
 public TaskExecutor threadPoolTaskExecutor(){
  
  ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setMaxPoolSize(12);
        executor.setCorePoolSize(8);
        executor.setQueueCapacity(15);
  
   return executor;
 }
 
 @Bean
    public JobLauncher asyncJobLauncher() throws Exception {
        SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
        
        jobLauncher.setJobRepository(jobRepository);
        jobLauncher.setTaskExecutor(threadPoolTaskExecutor());
        return jobLauncher;
 }
}

Answer 3

If you're using Lombok this might help you:

TLDR: Lombok @AllArgsConstructor doesn't seem to work well with @Qualifier annotation EDIT: if you have enable @Qualifier annotations in the lombok.config file to be able to use @Qualifier with @AllArgsConstructor like this:

lombok.copyableAnnotations += org.springframework.beans.factory.annotation.Qualifier

I know old question, however I had the exact same problem and none of the answers solved it.

I configured the async job launcher like this and added the qualifier to make sure this jobLauncher is injected:

 @Bean(name = "asyncJobLauncher")
 public JobLauncher simpleJobLauncher(JobRepository jobRepository) throws Exception {
        SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
        jobLauncher.setJobRepository(jobRepository);
        jobLauncher.setTaskExecutor(new SimpleAsyncTaskExecutor());
        jobLauncher.afterPropertiesSet();
        return jobLauncher;
    }

And injected it like this

@Qualifier("asyncJobLauncher")
private final JobLauncher jobLauncher;

I was using Lombok @AllArgsConstructor after changing it to autowire, the correct job launcher got injected and the job is now executed asynchronously:

@Autowired
@Qualifier("asyncJobLauncher")
private JobLauncher jobLauncher;

Also I didn't had to extend my configuration from DefaultBatchConfigurer

Answer 4

If I look at your code I see a couple of mistake. First of all your custom config is not loaded, because, if it was, the injection will fail for duplicate bean instance for the same interface.

There's a lot of magic in spring boot, but if you don't tell him to do some component scan, nothing will be loaded as espected.

The second problem that i can see is your BatchConfig class: it does not extends DefaultBatchConfigure, nor overrides getJobLauncher(), so even if the boot magic will load everything you'll get the default one. Here is a configuration that will work and it's compliant with the documentation @EnableBatchProcessing API

BatchConfig

@Configuration
@EnableBatchProcessing(modular = true)
@Slf4j
public class BatchConfig extends DefaultBatchConfigurer {

  @Override
  @Bean
  public JobLauncher getJobLauncher() {
    try {
      SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
      jobLauncher.setJobRepository(getJobRepository());
      jobLauncher.setTaskExecutor(new SimpleAsyncTaskExecutor());
      jobLauncher.afterPropertiesSet();
      return jobLauncher;

    } catch (Exception e) {
      log.error("Can't load SimpleJobLauncher with SimpleAsyncTaskExecutor: {} fallback on default", e);
      return super.getJobLauncher();
    }
  }
}

Main Function

@SpringBootApplication
@EnableScheduling
@EnableAsync
@ComponentScan(basePackageClasses = {BatchConfig.class})
public class SpringBatchTestApplication {

    public static void main(String[] args) {
        SpringApplication.run(SpringBatchTestApplication.class, args);
    }
}

Answer 5

According to spring documentation to return a response of the http request asynchronous it is required to use org.springframework.core.task.SimpleAsyncTaskExecutor.

Any implementation of the spring TaskExecutor interface can be used to control how jobs are asynchronously executed.

spring batch documentation

<bean id="jobLauncher"
  class="org.springframework.batch.core.launch.support.SimpleJobLauncher">
<property name="jobRepository" ref="jobRepository" />
<property name="taskExecutor">
    <bean class="org.springframework.core.task.SimpleAsyncTaskExecutor" />
</property>

Answer 6

Try this,In your Configuration You need to create customJobLauncher with SimpleAsyncTaskExecutor using the @Bean(name = "myJobLauncher") and same will be used @Qualifier in your controller.

@Bean(name = "myJobLauncher")
public JobLauncher simpleJobLauncher() throws Exception {
    SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
    jobLauncher.setJobRepository(jobRepository);
    jobLauncher.setTaskExecutor(new SimpleAsyncTaskExecutor());
    jobLauncher.afterPropertiesSet();
    return jobLauncher;
}

In your Controller

@Autowired
@Qualifier("myJobLauncher")
private JobLauncher jobLauncher;

Answer 7

Although you’ve your custom jobLauncher, you’re running the job using default jobLauncher provided by Spring. Could you please autowire simpleJobLauncher in your controller and give it a try?

Answer 8

JobExecution jobExecution = jobLauncher.run(batchComponent.customJob("paramhere"), jobParameters);. Joblauncher will wait after the Job has been completed before returning anything, that why your service is probably taking long to respond if that is your problem. If you want asynchronous capabilities, you might want to look at Spring's @EnableAsync & @Async.

@EnableAsync