I am working on a Spring Batch application containing two different Job bean (representing two different jobs). Both these jobs must be performed by my application (at the moment it can be done both sequentially and in parallel. It is not so important at the moment).
In order to achieve this behavior I am trying to follow this documentation but I am finding several problems: https://newbedev.com/spring-batch-running-multiple-jobs-in-parallel
I will try to explain what is my situation and what are the the encountered problems:
First of all I have this configuration class where my two Jobs object (and the related steps) are declared:
@Configuration
public class UpdateInfoBatchConfig {
private static final String PROPERTY_REST_API_URL = "rest.api.url";
@Autowired
private NotaryListServiceAdapter notaryListServiceAdapter;
@Autowired
private JobBuilderFactory jobs;
@Autowired
private StepBuilderFactory steps;
@Autowired
private NotaryService notaryService;
@Bean("firstStepItemReader")
public ItemReader<NotaryDistrict> itemReader(Environment environment, RestTemplate restTemplate) throws IllegalStateException, URISyntaxException {
System.out.println("itemReader() START !!!");
return new RESTNotaryDistrictsReader();
}
@Bean("firstStepItemWriter")
public ItemWriter<NotaryDistrict> itemWriter() {
return new LoggingItemWriter();
}
@Bean("secondStepItemReader")
public ItemReader<NotaryDistrict> secondStepReader(Environment environment) throws IllegalStateException {
System.out.println("secondStepItemReader() creation !!!");
return new SecondStepItemReader();
}
@Bean("secondStepItemProcessor")
public ItemProcessor<NotaryDistrict, NotaryDistrict> secondStepItemProcessor() {
return new SecondStepItemProcessor();
}
@Bean("secondStepItemWriter")
public ItemWriter<NotaryDistrict> secondStepItemWriter() {
return new SecondStepItemWriter();
}
/**
************************************ UPDATE NOTARY DISTRICTS LIST JOB SECTION ********************************************
*/
/**
* Creates a bean that represents the first step of the batch.
* How it works:
* 1) Call an external API in order to retrieve notary districts list
* 2) Return notary district one by one to the second step
* @param reader a custom reader calling an external API
* @param writer
* @param stepBuilderFactory
* @return
*/
@Bean("firstStep")
public Step firstStep(@Qualifier("firstStepItemReader") ItemReader<NotaryDistrict> reader,
@Qualifier("firstStepItemWriter") ItemWriter<NotaryDistrict> writer,
StepBuilderFactory stepBuilderFactory) {
return stepBuilderFactory.get("updateNotaryDistrictsStep")
.<NotaryDistrict, NotaryDistrict>chunk(1)
.reader(reader)
.writer(writer)
.build();
}
@Bean("secondStep")
public Step secondStep(@Qualifier("secondStepItemReader") ItemReader<NotaryDistrict> secondStepItemReader,
@Qualifier("secondStepItemProcessor") ItemProcessor<NotaryDistrict, NotaryDistrict> secondStepItemProcessor,
@Qualifier("secondStepItemWriter") ItemWriter<NotaryDistrict> secondStepItemWriter,
StepBuilderFactory stepBuilderFactory) {
return stepBuilderFactory.get("secondStep")
.<NotaryDistrict, NotaryDistrict>chunk(1)
.reader(secondStepItemReader)
.processor(secondStepItemProcessor)
.writer(secondStepItemWriter)
.build();
}
@Bean("updateNotaryDistrictsJob")
public Job updateNotaryDistrictsJob(JobBuilderFactory jobBuilderFactory,
@Qualifier("firstStep") Step firstStep,
@Qualifier("secondStep") Step secondStep) {
return jobBuilderFactory.get("updateNotaryDistrictsJob")
.start(firstStep)
.next(secondStep)
//.next(playerSummarization())
.build();
}
@Bean
public ExecutionContext executionContext() {
return new ExecutionContext();
}
/**
************************************ UPDATE NOTARY LIST JOB SECTION ********************************************
*/
@Bean()
public ItemReaderAdapter serviceItemReader() {
ItemReaderAdapter reader = new ItemReaderAdapter();
reader.setTargetObject(notaryListServiceAdapter);
reader.setTargetMethod("nextNotaryElement");
return reader;
}
@Bean
public Step readNotaryListStep(){
return steps.get("readNotaryListStep").
<Integer,Integer>chunk(1)
.reader(serviceItemReader())
.processor(new NotaryDetailsEnrichProcessor(notaryService))
.writer(new ConsoleItemWriter())
.build();
}
@Bean("updateNotaryListInfoJob")
public Job updateNotaryListInfoJob(){
return jobs.get("updateNotaryListInfoJob")
.incrementer(new RunIdIncrementer())
.start(readNotaryListStep())
.build();
}
}
Then, in a first moment, I created this other SpringBatchExampleJobLauncher launcher class. This works fine and was initially used in order to start a single job (I suppose that I have to change the logic of this launcher class in order to execute two jobs instead a single one):
public class SpringBatchExampleJobLauncher {
private static final Logger LOGGER = LoggerFactory.getLogger(SpringBatchExampleJobLauncher.class);
private final Job job;
private final JobLauncher jobLauncher;
private ExecutionContext executionContext;
@Autowired
public SpringBatchExampleJobLauncher(@Qualifier("updateNotaryDistrictsJob") Job job,
JobLauncher jobLauncher,
ExecutionContext executionContext) {
this.job = job;
this.jobLauncher = jobLauncher;
this.executionContext = executionContext;
}
//@Scheduled(cron = "0 */3 * * * *")
@Scheduled(cron = "0/30 * * * * *")
public void runSpringBatchExampleJob() throws JobParametersInvalidException, JobExecutionAlreadyRunningException, JobRestartException, JobInstanceAlreadyCompleteException {
LOGGER.info("Spring Batch example job was started");
List<NotaryDistrict> notaryDistrictsList = new ArrayList<NotaryDistrict>();
executionContext.put("notaryDistrictsList", notaryDistrictsList);
jobLauncher.run(job, newExecution());
LOGGER.info("Spring Batch example job was stopped");
}
private JobParameters newExecution() {
Map<String, JobParameter> parameters = new HashMap<>();
JobParameter parameter = new JobParameter(new Date());
parameters.put("currentTime", parameter);
return new JobParameters(parameters);
}
}
As you can see this class is pretty simple: its constructor take a single specific job (identified by the @Qualifier that is defined into the previous configuration class), the JobLauncher in order to run dis job and the ExecutionContext.
Then it contains the runSpringBatchExampleJob() that run this single job every 30 seconds (as specified by the CRON exception).
Ok….so, in order to start both my 2 jobs I think that I need to change this SpringBatchExampleJobLauncher in a similar way as shown here: https://newbedev.com/spring-batch-running-multiple-jobs-in-parallel
So what I have done. First of all I added the ThreadPoolTaskExecutor and the JobLauncher beans definition into my UpdateInfoBatchConfig configuration class, that become this:
@Configuration
public class UpdateInfoBatchConfig {
private static final String PROPERTY_REST_API_URL = "rest.api.url";
@Autowired
private NotaryListServiceAdapter notaryListServiceAdapter;
@Autowired
private JobBuilderFactory jobs;
@Autowired
private StepBuilderFactory steps;
@Autowired
private NotaryService notaryService;
@Bean("firstStepItemReader")
public ItemReader<NotaryDistrict> itemReader(Environment environment, RestTemplate restTemplate) throws IllegalStateException, URISyntaxException {
System.out.println("itemReader() START !!!");
return new RESTNotaryDistrictsReader();
}
@Bean("firstStepItemWriter")
public ItemWriter<NotaryDistrict> itemWriter() {
return new LoggingItemWriter();
}
@Bean("secondStepItemReader")
public ItemReader<NotaryDistrict> secondStepReader(Environment environment) throws IllegalStateException {
System.out.println("secondStepItemReader() creation !!!");
return new SecondStepItemReader();
}
@Bean("secondStepItemProcessor")
public ItemProcessor<NotaryDistrict, NotaryDistrict> secondStepItemProcessor() {
return new SecondStepItemProcessor();
}
@Bean("secondStepItemWriter")
public ItemWriter<NotaryDistrict> secondStepItemWriter() {
return new SecondStepItemWriter();
}
/**
* Creates a bean that represents the first step of the batch.
* How it works:
* 1) Call an external API in order to retrieve notary districts list
* 2) Return notary district one by one to the second step
* @param reader a custom reader calling an external API
* @param writer
* @param stepBuilderFactory
* @return
*/
@Bean("firstStep")
public Step firstStep(@Qualifier("firstStepItemReader") ItemReader<NotaryDistrict> reader,
@Qualifier("firstStepItemWriter") ItemWriter<NotaryDistrict> writer,
StepBuilderFactory stepBuilderFactory) {
return stepBuilderFactory.get("updateNotaryDistrictsStep")
.<NotaryDistrict, NotaryDistrict>chunk(1)
.reader(reader)
.writer(writer)
.build();
}
@Bean("secondStep")
public Step secondStep(@Qualifier("secondStepItemReader") ItemReader<NotaryDistrict> secondStepItemReader,
@Qualifier("secondStepItemProcessor") ItemProcessor<NotaryDistrict, NotaryDistrict> secondStepItemProcessor,
@Qualifier("secondStepItemWriter") ItemWriter<NotaryDistrict> secondStepItemWriter,
StepBuilderFactory stepBuilderFactory) {
return stepBuilderFactory.get("secondStep")
.<NotaryDistrict, NotaryDistrict>chunk(1)
.reader(secondStepItemReader)
.processor(secondStepItemProcessor)
.writer(secondStepItemWriter)
.build();
}
@Bean("updateNotaryDistrictsJob")
public Job updateNotaryDistrictsJob(JobBuilderFactory jobBuilderFactory,
@Qualifier("firstStep") Step firstStep,
@Qualifier("secondStep") Step secondStep) {
return jobBuilderFactory.get("updateNotaryDistrictsJob")
.start(firstStep)
.next(secondStep)
//.next(playerSummarization())
.build();
}
@Bean
public ExecutionContext executionContext() {
return new ExecutionContext();
}
/**
************************************ UPDATE NOTARY LIST JOB ********************************************
*/
@Bean()
public ItemReaderAdapter serviceItemReader() {
ItemReaderAdapter reader = new ItemReaderAdapter();
reader.setTargetObject(notaryListServiceAdapter);
reader.setTargetMethod("nextNotaryElement");
return reader;
}
@Bean
public Step readNotaryListStep(){
return steps.get("readNotaryListStep").
<Integer,Integer>chunk(1)
.reader(serviceItemReader())
.processor(new NotaryDetailsEnrichProcessor(notaryService))
.writer(new ConsoleItemWriter())
.build();
}
@Bean("updateNotaryListInfoJob")
public Job updateNotaryListInfoJob(){
return jobs.get("updateNotaryListInfoJob")
.incrementer(new RunIdIncrementer())
.start(readNotaryListStep())
.build();
}
/**
************************************ MULTIPLE JOB CONFIGURATION ********************************************
*/
@Bean
public ThreadPoolTaskExecutor taskExecutor() {
ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
taskExecutor.setCorePoolSize(15);
taskExecutor.setMaxPoolSize(20);
taskExecutor.setQueueCapacity(30);
return taskExecutor;
}
@Bean
public JobLauncher jobLauncher(ThreadPoolTaskExecutor taskExecutor, JobRepository jobRepository){
SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
jobLauncher.setTaskExecutor(taskExecutor);
jobLauncher.setJobRepository(jobRepository);
return jobLauncher;
}
}
As you can see the last two bean are the ThreadPoolTaskExecutor and my JobLauncher beans.
Then I changed my SpringBatchExampleJobLauncher in order to use this launcher and perform both my jobs instead a single one, this is what I have done:
/**
* This bean schedules and runs our Spring Batch job.
*/
@Component
public class SpringBatchExampleJobLauncher {
private static final Logger LOGGER = LoggerFactory.getLogger(SpringBatchExampleJobLauncher.class);
@Autowired
private JobLauncher jobLauncher;
@Autowired
@Qualifier("updateNotaryDistrictsJob")
private Job updateNotaryDistrictsJob;
@Autowired
@Qualifier("updateNotaryListInfoJob")
private Job updateNotaryListInfoJob;
@Scheduled(cron = "0/30 * * * * *")
public void run1(){
Map<String, JobParameter> confMap = new HashMap<>();
confMap.put("time", new JobParameter(System.currentTimeMillis()));
JobParameters jobParameters = new JobParameters(confMap);
try {
jobLauncher.run(updateNotaryDistrictsJob, jobParameters);
}catch (Exception ex){
LOGGER.error(ex.getMessage());
}
}
@Scheduled(cron = "0/50 * * * * *")
public void run2(){
Map<String, JobParameter> confMap = new HashMap<>();
confMap.put("time", new JobParameter(System.currentTimeMillis()));
JobParameters jobParameters = new JobParameters(confMap);
try {
jobLauncher.run(updateNotaryListInfoJob, jobParameters);
}catch (Exception ex){
LOGGER.error(ex.getMessage());
}
}
}
As you can see I am now injecting the previous defined JobLauncher bean and my two jobs beans (defined into my configuration class). Then I defined the run1() and the run2() methods that should run both my injected jobs when the CRON expression is satisfied.
The problem is I am now obtaining the following error in my stracktrace and nothing is performed:
*************************** APPLICATION FAILED TO START *************************** Description: The bean 'jobLauncher', defined in class path resource [org/springframework/batch/core/configuration/annotation/SimpleBatchConfiguration.class], could not be registered. A bean with that name has already been defined in class path resource [com/notariato/updateInfo/UpdateInfoBatchConfig.class] and overriding is disabled. Action: Consider renaming one of the beans or enabling overriding by setting spring.main.allow-bean-definition-overriding=true
Basically, it seems to me that this error is saying to me that the JobLauncher bean that I am trying to inject into my launcher class is yet defined into my UpdateInfoBatchConfig configuration class. But it is exactly what I am expected because I define my bean into the configuration class and then I am injecting it into the launcher class to be used.
What is wrong? What am I missing? How can I try to solve this issue?
Advertisement
Answer
This is because you are defining a JobLauncher bean in the application context and Spring Batch is also defining that bean via @EnableBatchProcessing (see its Javadoc).
If you want to use a custom JobLauncher, you should provide a BatchConfigurer bean and override getJobLauncher. One way to do that is to make one of your configuration classes extend DefaultBatchConfigurer and override createJobLauncher(). This is explained in more details in the docs here.