How to Launch the Same Job in Parallel with Spring Batch?

Question

I am trying to launching the same Job in parallel, which the Job does reading with JdbcCurorItemReader from a database table and write it to a different table. The table contains one column and integer only. I created a JobLauncher with SimpleAsyncTaskExecutor. I am running the job with a for loop. The Job seems to start in parallel but I am getting various errors for the reader. such as "Unexpected cursor position change.", "Reader must be open before it can be read." and "Failed to initialize the reader". What should I do to solve this problem?

@Bean
public JobLauncher batchJobLauncher(SimpleAsyncTaskExecutor simpleAsyncTaskExecutor, JobRepository jobRepository){
        SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
        jobLauncher.setTaskExecutor(new SimpleAsyncTaskExecutor(){{
            setConcurrencyLimit(20);
        }});
        jobLauncher.setJobRepository(jobRepository);
        return jobLauncher;
    }

@Bean
public JdbcBatchItemWriter<Number> writer(DataSource dataSource){
        return  new JdbcBatchItemWriterBuilder<Number>()
            .itemPreparedStatementSetter(new ItemPreparedStatementSetter<Number>()
                {
                    @Override
                    public void setValues(Number number, PreparedStatement preparedStatement) throws SQLException
                    {
                        preparedStatement.setInt(1, Number.getNumber());
                    }
                })
                .dataSource(dataSource)
                .sql("INSERT INTO INTEGER_NUMBER2(NUMBER) VALUES (?)")
                .build();
    }

@Bean
public JdbcCursorItemReader<Number> reader(DataSource dataSource){
    return new JdbcCursorItemReaderBuilder<Number>()
                .dataSource(dataSource)
                .sql("SELECT INTEGER_NUMBER FROM NUMBER")
                .rowMapper(new NumberRowMapper())
                .saveState(false)
                .build();
    }

for(int i=0; i<5; i++){
   batchJobLauncher.run(job, new JobParametersBuilder().addLong("Num_Id", i)).toJobParameters());
}

Answer 1

The JdbcCursorItemReader is not thread-safe. According to your code, it is declared as a singleton bean and is shared between threads. What you can do is wrap it in a SynchronizedItemStreamReader or make it a step scoped bean.