Spring Batch with SimpleAsyncTaskExecutor is not saving to DB

Question

I have a simple Spring Boot application with an enpoint that invokes a Spring Batch Job asynchronously through the SimpleAsyncTaskExecutor configured in the JobLauncher bean.

The Spring Batch Job starts asynchronously and works fine, but nothing is saved to the database.

If I remove the SimpleAsyncTaskExecutor the data is saved.

This is my BatchConfigurer. Heere I configure the JobLauncher with a SimpleAsyncTaskExecutor . If I remove the line simpleJobLauncher.setTaskExecutor(new SimpleAsyncTaskExecutor()); // (1) the data is saved.

@Component
public class CustomBatchConfiguration implements BatchConfigurer {

private static final Log LOGGER = LogFactory.getLog(CustomBatchConfiguration.class);

@Autowired
private BatchProperties properties;

@Autowired
private DataSource dataSource;

@Autowired
private EntityManagerFactory entityManagerFactory;

private PlatformTransactionManager transactionManager;

private JobRepository jobRepository;

private JobLauncher jobLauncher;

private JobExplorer jobExplorer;

/**
 * Registers {@link JobRepository} bean.
 */
@Override
public JobRepository getJobRepository() {
    return this.jobRepository;
}

/**
 * Registers {@link PlatformTransactionManager} bean.
 */
@Override
public PlatformTransactionManager getTransactionManager() {
    return this.transactionManager;
}

/**
 * Registers {@link JobLauncher} bean.
 */
@Override
public JobLauncher getJobLauncher() {
    return this.jobLauncher;
}

/**
 * Registers {@link JobExplorer} bean. This bean is actually created in
 * {@link BatchConfig}.
 */
@Override
public JobExplorer getJobExplorer() throws Exception {
    return this.jobExplorer;
}

/**
 * Initializes Spring Batch components.
 */
@PostConstruct
public void initialize() {
    try {
        this.transactionManager = createTransactionManager();
        this.jobRepository = createJobRepository();
        this.jobLauncher = createJobLauncher();
        this.jobExplorer = createJobExplorer();
    } catch (Exception ex) {
        throw new IllegalStateException("Unable to initialize Spring Batch", ex);
    }
}

private JobExplorer createJobExplorer() throws Exception {
    JobExplorerFactoryBean jobExplorerFactoryBean = new JobExplorerFactoryBean();
    jobExplorerFactoryBean.setDataSource(this.dataSource);
    String tablePrefix = this.properties.getTablePrefix();

    if (StringUtils.hasText(tablePrefix)) {
        jobExplorerFactoryBean.setTablePrefix(tablePrefix);
    }

    jobExplorerFactoryBean.afterPropertiesSet();
    return jobExplorerFactoryBean.getObject();
}

private JobLauncher createJobLauncher() throws Exception {
    SimpleJobLauncher simpleJobLauncher = new SimpleJobLauncher();

    simpleJobLauncher.setJobRepository(getJobRepository());
    simpleJobLauncher.setTaskExecutor(new SimpleAsyncTaskExecutor()); // (1)
    simpleJobLauncher.afterPropertiesSet();

    return simpleJobLauncher;
}

private JobRepository createJobRepository() throws Exception {
    JobRepositoryFactoryBean jobRepositoryFactoryBean = new JobRepositoryFactoryBean();

    jobRepositoryFactoryBean.setDatabaseType("db2");
    jobRepositoryFactoryBean.setDataSource(this.dataSource);

    if (this.entityManagerFactory != null) {
        LOGGER.warn("JPA does not support custom isolation levels, so locks may not be taken when launching Jobs");
        jobRepositoryFactoryBean.setIsolationLevelForCreate("ISOLATION_DEFAULT");
    }

    String tablePrefix = this.properties.getTablePrefix();
    if (StringUtils.hasText(tablePrefix)) {
        jobRepositoryFactoryBean.setTablePrefix(tablePrefix);
    }

    jobRepositoryFactoryBean.setTransactionManager(getTransactionManager());
    jobRepositoryFactoryBean.afterPropertiesSet();

    return jobRepositoryFactoryBean.getObject();
}

private PlatformTransactionManager createTransactionManager() {
    return new DataSourceTransactionManager(dataSource);
}

}

And this is my BatchConfiguration.

@Autowired(required = true)
private MyItemReader myItemReader;

@Autowired(required = true)
private MyItemProcessor myItemProcessor;

@Autowired(required = true)
private MyItemWriter myItemWriter;


@Bean
public Step myStep(TaskExecutor taskExecutor) { 

    return stepBuilderFactory.get("myStepName")
            .<SomeWrapper, SomeWrapper>chunk(
                    1)
            .reader(myItemReader)
            .processor(myItemProcessor).writer(myItemWriter)

            .build();
}

@Bean(name = "myJob")
public Job myJob(Step myStep) {
    return jobBuilderFactory.get("myJobName").incrementer(new RunIdIncrementer())
            .flow(myStep).end().build();
}

I'm missing something?

Thanks in advance

alobaton alobaton · Accepted Answer · 2020-05-03T19:39:46

I did the following to solve my problem:

In CustomBatchConfiguration I changed the following

private PlatformTransactionManager createTransactionManager() {        
    return new DataSourceTransactionManager(dataSource);
}

to

private PlatformTransactionManager createTransactionManager() {
    if (this.entityManagerFactory != null) {
        return new JpaTransactionManager(this.entityManagerFactory);
    }

    return new DataSourceTransactionManager(this.dataSource);
}

I used the the spring JpaTransactionManager.

I change JobLauncher TaskExecutor from

simpleJobLauncher.setTaskExecutor(new SimpleAsyncTaskExecutor());

to

ThreadPoolTaskExecutor threadPoolExecutor = new ThreadPoolTaskExecutor();
threadPoolExecutor.setMaxPoolSize(springBatchJobThreadPollMaxSize);
threadPoolExecutor.afterPropertiesSet();

TaskExecutor taskExecutor = new DelegatingSecurityContextAsyncTaskExecutor(threadPoolExecutor);

Creating a more robust pool of threads and sharing the spring security context among them.

I used a Tasklet instead of Spring Batch chunk oriented solution. Here the details.

Spring Batch with SimpleAsyncTaskExecutor is not saving to DB

1 Answers