org.springframework.batch.core.JobExecutionException：分区处理程序在 Spring Batch 中返回了一个不成功的步骤答案

【问题标题】：org.springframework.batch.core.JobExecutionException: Partition handler returned an unsuccessful step in Spring Batchorg.springframework.batch.core.JobExecutionException：分区处理程序在 Spring Batch 中返回了一个不成功的步骤
【发布时间】：2020-05-16 15:32:03
【问题描述】：

我正在通过 Spring Boot 使用 Spring Batch。在此示例中，我实现了自定义分区，因为 Employee Table 具有 UUID 值作为 PK。

Error:
org.springframework.transaction.CannotCreateTransactionException: Could not open JDBC Connection for transaction; nested exception is java.sql.SQLTransientConnectionException: HikariPool-1 - Connection is not available, request timed out after 30002ms.
    at org.springframework.jdbc.datasource.DataSourceTransactionManager.doBegin(DataSourceTransactionManager.java:307) ~[spring-jdbc-5.2.4.RELEASE.jar:5.2.4.RELEASE]
    at org.springframework.transaction.support.AbstractPlatformTransactionManager.startTransaction(AbstractPlatformTransactionManager.java:400) ~[spring-tx-5.2.4.RELEASE.jar:5.2.4.RELEASE]
    at org.springframework.transaction.support.AbstractPlatformTransactionManager.getTransaction(AbstractPlatformTransactionManager.java:373) ~[spring-tx-5.2.4.RELEASE.jar:5.2.4.RELEASE]
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.8.0_171]
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[na:1.8.0_171]
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:1.8.0_171]
    at java.lang.reflect.Method.invoke(Method.java:498) ~[na:1.8.0_171]
    at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:344) ~[spring-aop-5.2.4.RELEASE.jar:5.2.4.RELEASE]
    at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:198) ~[spring-aop-5.2.4.RELEASE.jar:5.2.4.RELEASE]
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163) ~[spring-aop-5.2.4.RELEASE.jar:5.2.4.RELEASE]
    at org.springframework.batch.core.configuration.annotation.SimpleBatchConfiguration$PassthruAdvice.invoke(SimpleBatchConfiguration.java:127) ~[spring-batch-core-4.2.1.RELEASE.jar:4.2.1.RELEASE]
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186) ~[spring-aop-5.2.4.RELEASE.jar:5.2.4.RELEASE]
    at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:212) ~[spring-aop-5.2.4.RELEASE.jar:5.2.4.RELEASE]
    at com.sun.proxy.$Proxy74.getTransaction(Unknown Source) ~[na:na]
    at org.springframework.transaction.support.TransactionTemplate.execute(TransactionTemplate.java:137) ~[spring-tx-5.2.4.RELEASE.jar:5.2.4.RELEASE]
    at org.springframework.batch.core.step.tasklet.TaskletStep$2.doInChunkContext(TaskletStep.java:273) ~[spring-batch-core-4.2.1.RELEASE.jar:4.2.1.RELEASE]
    at org.springframework.batch.core.scope.context.StepContextRepeatCallback.doInIteration(StepContextRepeatCallback.java:82) ~[spring-batch-core-4.2.1.RELEASE.jar:4.2.1.RELEASE]
    at org.springframework.batch.repeat.support.RepeatTemplate.getNextResult(RepeatTemplate.java:375) ~[spring-batch-infrastructure-4.2.1.RELEASE.jar:4.2.1.RELEASE]
    at org.springframework.batch.repeat.support.RepeatTemplate.executeInternal(RepeatTemplate.java:215) ~[spring-batch-infrastructure-4.2.1.RELEASE.jar:4.2.1.RELEASE]
    at org.springframework.batch.repeat.support.RepeatTemplate.iterate(RepeatTemplate.java:145) ~[spring-batch-infrastructure-4.2.1.RELEASE.jar:4.2.1.RELEASE]
    at org.springframework.batch.core.step.tasklet.TaskletStep.doExecute(TaskletStep.java:258) ~[spring-batch-core-4.2.1.RELEASE.jar:4.2.1.RELEASE]
    at org.springframework.batch.core.step.AbstractStep.execute(AbstractStep.java:208) ~[spring-batch-core-4.2.1.RELEASE.jar:4.2.1.RELEASE]
    at org.springframework.batch.core.partition.support.TaskExecutorPartitionHandler$1.call(TaskExecutorPartitionHandler.java:138) [spring-batch-core-4.2.1.RELEASE.jar:4.2.1.RELEASE]
    at org.springframework.batch.core.partition.support.TaskExecutorPartitionHandler$1.call(TaskExecutorPartitionHandler.java:135) [spring-batch-core-4.2.1.RELEASE.jar:4.2.1.RELEASE]
    at java.util.concurrent.FutureTask.run(FutureTask.java:266) [na:1.8.0_171]
    at java.lang.Thread.run(Thread.java:748) [na:1.8.0_171]
Caused by: java.sql.SQLTransientConnectionException: HikariPool-1 - Connection is not available, request timed out after 30002ms.
    at com.zaxxer.hikari.pool.HikariPool.createTimeoutException(HikariPool.java:689) ~[HikariCP-3.4.2.jar:na]
    at com.zaxxer.hikari.pool.HikariPool.getConnection(HikariPool.java:196) ~[HikariCP-3.4.2.jar:na]
    at com.zaxxer.hikari.pool.HikariPool.getConnection(HikariPool.java:161) ~[HikariCP-3.4.2.jar:na]
    at com.zaxxer.hikari.HikariDataSource.getConnection(HikariDataSource.java:128) ~[HikariCP-3.4.2.jar:na]
    at org.springframework.jdbc.datasource.DataSourceTransactionManager.doBegin(DataSourceTransactionManager.java:263) ~[spring-jdbc-5.2.4.RELEASE.jar:5.2.4.RELEASE]
    ... 25 common frames omitted

2020-05-16 20:55:22.675  INFO 31056 --- [cTaskExecutor-7] o.s.batch.core.step.AbstractStep         : Step: [slaveStep:Thread--9] executed in 32s370ms
###################### 80
2020-05-16 20:55:22.787  INFO 31056 --- [TaskExecutor-10] org.mongodb.driver.connection            : Opened connection [connectionId{localValue:2, serverValue:70}] to localhost:27017
2020-05-16 20:55:22.815  INFO 31056 --- [TaskExecutor-10] o.s.batch.core.step.AbstractStep         : Step: [slaveStep:Thread--8] executed in 32s510ms
###################### 50
###################### 70
2020-05-16 20:55:22.839  INFO 31056 --- [cTaskExecutor-1] o.s.batch.core.step.AbstractStep         : Step: [slaveStep:Thread--5] executed in 32s539ms
###################### 20
###################### 40
2020-05-16 20:55:22.845  INFO 31056 --- [cTaskExecutor-4] o.s.batch.core.step.AbstractStep         : Step: [slaveStep:Thread--7] executed in 32s540ms
###################### 10
2020-05-16 20:55:22.851  INFO 31056 --- [cTaskExecutor-9] o.s.batch.core.step.AbstractStep         : Step: [slaveStep:Thread--2] executed in 32s546ms
###################### 30
2020-05-16 20:55:22.856  INFO 31056 --- [cTaskExecutor-3] o.s.batch.core.step.AbstractStep         : Step: [slaveStep:Thread--1] executed in 32s556ms
###################### 100
###################### 60
2020-05-16 20:55:22.860  INFO 31056 --- [cTaskExecutor-6] org.mongodb.driver.connection            : Opened connection [connectionId{localValue:3, serverValue:71}] to localhost:27017
2020-05-16 20:55:22.864  INFO 31056 --- [cTaskExecutor-5] o.s.batch.core.step.AbstractStep         : Step: [slaveStep:Thread--3] executed in 32s559ms
2020-05-16 20:55:22.867  INFO 31056 --- [cTaskExecutor-6] o.s.batch.core.step.AbstractStep         : Step: [slaveStep:Thread--4] executed in 32s565ms
2020-05-16 20:55:22.874  INFO 31056 --- [cTaskExecutor-8] o.s.batch.core.step.AbstractStep         : Step: [slaveStep:Thread--6] executed in 32s569ms
2020-05-16 20:55:22.883  INFO 31056 --- [cTaskExecutor-2] o.s.batch.core.step.AbstractStep         : Step: [slaveStep:Thread--10] executed in 32s583ms
2020-05-16 20:55:22.885 ERROR 31056 --- [           main] o.s.batch.core.step.AbstractStep         : Encountered an error executing step contactStepOne in job readContactJob

org.springframework.batch.core.JobExecutionException: Partition handler returned an unsuccessful step
    at org.springframework.batch.core.partition.support.PartitionStep.doExecute(PartitionStep.java:112) ~[spring-batch-core-4.2.1.RELEASE.jar:4.2.1.RELEASE]
    at org.springframework.batch.core.step.AbstractStep.execute(AbstractStep.java:208) ~[spring-batch-core-4.2.1.RELEASE.jar:4.2.1.RELEASE]
    at org.springframework.batch.core.job.SimpleStepHandler.handleStep(SimpleStepHandler.java:148) [spring-batch-core-4.2.1.RELEASE.jar:4.2.1.RELEASE]
    at org.springframework.batch.core.job.AbstractJob.handleStep(AbstractJob.java:410) [spring-batch-core-4.2.1.RELEASE.jar:4.2.1.RELEASE]
    at org.springframework.batch.core.job.SimpleJob.doExecute(SimpleJob.java:136) [spring-batch-core-4.2.1.RELEASE.jar:4.2.1.RELEASE]
    at org.springframework.batch.core.job.AbstractJob.execute(AbstractJob.java:319) [spring-batch-core-4.2.1.RELEASE.jar:4.2.1.RELEASE]
    at org.springframework.batch.core.launch.support.SimpleJobLauncher$1.run(SimpleJobLauncher.java:147) [spring-batch-core-4.2.1.RELEASE.jar:4.2.1.RELEASE]
    at org.springframework.core.task.SyncTaskExecutor.execute(SyncTaskExecutor.java:50) [spring-core-5.2.4.RELEASE.jar:5.2.4.RELEASE]
    at org.springframework.batch.core.launch.support.SimpleJobLauncher.run(SimpleJobLauncher.java:140) [spring-batch-core-4.2.1.RELEASE.jar:4.2.1.RELEASE]
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.8.0_171]
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[na:1.8.0_171]
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:1.8.0_171]
    at java.lang.reflect.Method.invoke(Method.java:498) ~[na:1.8.0_171]
    at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:344) [spring-aop-5.2.4.RELEASE.jar:5.2.4.RELEASE]
    at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:198) [spring-aop-5.2.4.RELEASE.jar:5.2.4.RELEASE]
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163) [spring-aop-5.2.4.RELEASE.jar:5.2.4.RELEASE]
    at org.springframework.batch.core.configuration.annotation.SimpleBatchConfiguration$PassthruAdvice.invoke(SimpleBatchConfiguration.java:127) [spring-batch-core-4.2.1.RELEASE.jar:4.2.1.RELEASE]
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186) [spring-aop-5.2.4.RELEASE.jar:5.2.4.RELEASE]
    at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:212) [spring-aop-5.2.4.RELEASE.jar:5.2.4.RELEASE]
    at com.sun.proxy.$Proxy71.run(Unknown Source) [na:na]
    at com.example.PostgresMongoContactsMigrationApplication.run(PostgresMongoContactsMigrationApplication.java:49) [classes/:na]
    at org.springframework.boot.SpringApplication.callRunner(SpringApplication.java:784) [spring-boot-2.2.5.RELEASE.jar:2.2.5.RELEASE]
    at org.springframework.boot.SpringApplication.callRunners(SpringApplication.java:768) [spring-boot-2.2.5.RELEASE.jar:2.2.5.RELEASE]
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:322) [spring-boot-2.2.5.RELEASE.jar:2.2.5.RELEASE]
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:1226) [spring-boot-2.2.5.RELEASE.jar:2.2.5.RELEASE]
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:1215) [spring-boot-2.2.5.RELEASE.jar:2.2.5.RELEASE]
    at com.example.PostgresMongoContactsMigrationApplication.main(PostgresMongoContactsMigrationApplication.java:38) [classes/:na]

EmployeePartitions.java

public class EmployeePartitions implements Partitioner {

    @Override
    public Map<String, ExecutionContext> partition(int gridSize) {
        Map<String, ExecutionContext> partitionMap = new HashMap<>();

        int limit = 10;
        int offset = 0;
        int increment = 10;
        int counter = 0;
        for (int i = 0; i < 100; i += increment) {
            counter++;
            if (i == 0)
                offset = i;
            else
                offset = i + 1;
            limit = i + increment;

            System.out.println("OFFSET " + offset + ", LIMIT = " + limit);

            ExecutionContext context = new ExecutionContext();
            context.put("limit", limit);
            context.put("offset", offset);

            partitionMap.put("Thread--" + counter, context);
            System.out.println(partitionMap);
        }

        System.out.println(partitionMap.size());
        return partitionMap;
    }
}

EmployeeJob.java

@Configuration
public class EmployeeJob {
    private static final Integer chunkSize = 200;

    @Autowired
    private DataSource dataSource;

    @Autowired
    private Environment env;

    @Autowired
    private JobBuilderFactory jobBuilderFactory;

    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Bean
    public EmployeesPartitions EmployeesPartition() {
        return new EmployeesPartitions();
    }


    @Bean("readEmployeeJob")
    public Job readEmployeeJob() throws Exception {
        return jobBuilderFactory.get("readEmployeeJob")
                .incrementer(new RunIdIncrementer())
                .start(EmployeeStepOne())
                .build();
    }

    @Bean
    public Step EmployeeStepOne() throws Exception {        
        return stepBuilderFactory.get("EmployeeStepOne")
                .partitioner(slaveStep().getName(), EmployeesPartition())
                .step(slaveStep())
                .gridSize(1)
                .taskExecutor(new SimpleAsyncTaskExecutor())
                .build();
    }


    // slave step
    @Bean
    public Step slaveStep() throws Exception {
        return stepBuilderFactory.get("slaveStep")
                .<EmployeesDTO, EmployeesDTO>chunk(chunkSize)
                .reader(EmployeeReader(null, null))
                .writer(EmployeeWriter())
                .build();
    }

    @Bean(destroyMethod = "")
    @StepScope
    public JdbcCursorItemReader<EmployeesDTO> EmployeeReader(
            @Value("#{stepExecutionContext['limit']}") Long limit,
            @Value("#{stepExecutionContext['offset']}") Long offset) throws Exception {
        JdbcCursorItemReader<EmployeesDTO> reader = new JdbcCursorItemReader<>();
        reader.setDataSource(this.dataSource);
        reader.setSql("select * from scv_owner.Employees order by Employee_id limit "+ limit +" offset "+offset);
        reader.setRowMapper(new EmployeeRowMapper());
        reader.afterPropertiesSet();
        return reader;
    }

    @Bean
    public ItemProcessor<EmployeesDTO, EmployeesDTO> EmployeeProcessor() {
        return new EmployeesProcessor();
    }

    @Bean
    public ItemWriter<EmployeesDTO> EmployeeWriter() {
        return new EmployeeWriter();
    }
}

【问题讨论】：

Could not open JDBC Connection for transaction; Connection is not available, request timed out after 30002ms.。看起来您的一个分区无法打开与数据库的连接，您的池中是否有足够的连接？您的数据库连接池的大小是多少？连接超时参数在这里也很重要，因为其中一个分区可以在超时之前完成并将其连接释放回可供另一个分区使用的池。
@MahmoudBenHassine - 我没有明确设置连接池大小，默认为 10，最好的大小是多少？此外，在对其他作业使用多线程步骤时，我也遇到了同样的问题。 connection timeout parameter 我需要设置什么？
@MahmoudBenHassine - 我真的被 n+1 期的 Spring Batch 困住了。可以说，DB 为单个员工提供了 3 个不同的电子邮件对象（电子邮件表不同），我需要使用 Join 进行设置，现在的问题是我将在哪里以及如何将这 3 个不同的电子邮件对象嵌入到单个员工中？
请继续关注这里讨论的问题，即连接超时。我为此添加了一个答案。关于电子邮件的问题，看起来你在这里问了关于 3 电子邮件问题的问题：stackoverflow.com/a/61885183/5019386（与用户user4567570），我也回答了。

标签： spring spring-batch

【解决方案1】：

最好的尺寸是多少？

没有最佳尺寸，这取决于用例。为了获得最大吞吐量，您应该设置池大小，例如如果您想要最大吞吐量，每个分区可以有自己的连接，否则根据分区完成所需的时间来选择超时。

例如，如果您有 10 个分区并且池大小为 5，则 5 个分区将工作，另外 5 个分区将等待。现在，如果一个分区最多需要 1 分钟才能完成，则可以将超时设置为 1 分钟（或者可能更长一点），这样等待的分区就不会超时，也就是在超时之前将连接返回到池中。

【讨论】：