Spring Batch 5.1.1 JdbcItemReader读取PostgreSQL失败，任务完成无数据迁移

问题场景 #

基于Spring Boot 3.2.5、Kotlin、PostgreSQL搭建的Spring Batch 5.1.1批处理任务，执行后日志显示任务完成，但data_batch表的实际数据未迁移至data_batch_etl表。执行日志如下：

Job: [SimpleJob: [name=Job1]] launched with the following parameters: [{}]
: Step already complete or not restartable, so no action to execute: StepExecution: id=3, version=7, name=Step1, status=COMPLETED, exitStatus=COMPLETED, readCount=0, filterCount=0, writeCount=0 readSkipCount=0, writeSkipCount=0, processSkipCount=0, commitCount=5, rollbackCount=0, exitDescription=
Job: [SimpleJob: [name=Job1]] completed with the following parameters: [{}] and the following status: [COMPLETED] in 19ms

原因分析 #

• 任务重复执行标记问题：日志明确提示Step already complete or not restartable，说明该Step已被Spring Batch标记为完成状态。Spring Batch会在元数据表（BATCH_JOB_EXECUTION、BATCH_STEP_EXECUTION等）中记录任务执行状态，当使用相同参数（此处为空参数{}）再次执行任务时，会判定任务已完成，不会触发读取、处理、写入逻辑，因此出现readCount=0的情况。
• 读取逻辑未触发：由于任务被判定为已完成，Step不会调用ItemReader执行查询，即便data_batch表有数据也无法被读取。

解决方法 #

1. 传入唯一任务参数 #

每次执行任务时传入不同的参数（如时间戳），让Spring Batch识别为新的任务实例，从而重新执行Step。例如启动时添加参数：

--job.parameters=runTime=$(date +%Y%m%d%H%M%S)

2. 配置Step允许重复执行 #

在Step定义中添加allowStartIfComplete(true)，允许Step即使已完成也可重新执行。修改step1的Bean定义：

@Bean
fun step1(jobRepository: JobRepository, taskExecutor: TaskExecutor): Step {
    return StepBuilder("Step1", jobRepository)
        .chunk<Data_Batch, Data_Batch_ETL>(3, transactionManager())
        .reader(itemReader())
        .processor(itemProcessor())
        .writer(itemWriter())
        .allowStartIfComplete(true) // 添加该行配置
        .build()
}

3. 清理任务元数据记录 #

手动删除BATCH_JOB_EXECUTION、BATCH_STEP_EXECUTION等元数据表中对应Job1和Step1的执行记录，下次执行时会重新初始化任务状态。注意：此方法会丢失历史执行记录，需谨慎操作。

核心配置代码 #

BatchConfig.kt #

@Configuration
class BatchConfig(){

    companion object {
        private val logger: Logger = LogManager.getLogger(BatchConfig::class.java)
    }

    @Autowired
    private lateinit var dataSource  : DataSource

    fun itemReader(): JdbcCursorItemReader<Data_Batch> {
        return JdbcCursorItemReaderBuilder<Data_Batch>()
            .dataSource(dataSource)
            .name("BatchReader")
            .sql("SELECT batch_code , batch_name FROM data_batch")
            .rowMapper(Batch_Mapper())
            .build()
    }

    fun itemProcessor(): Batch_Procesor {
        return Batch_Procesor()
    }

    fun itemWriter(): JdbcBatchItemWriter<Data_Batch_ETL> {
        return JdbcBatchItemWriterBuilder<Data_Batch_ETL>()
            .sql("INSERT INTO data_batch_etl (batch_name, batch_source) VALUES (:batch_name, :batch_source)")
            .dataSource(dataSource)
            .beanMapped()
            .build()
    }

    @Bean
    fun step1(jobRepository: JobRepository, taskExecutor: TaskExecutor): Step {
        return StepBuilder("Step1",jobRepository)
            .chunk<Data_Batch,Data_Batch_ETL>(3,transactionManager())
            .reader(itemReader())
            .processor(itemProcessor())
            .writer(itemWriter())
            .build()
    }

    @Bean
    fun Job1( jobRepository: JobRepository ,step1 : Step , listener : JobCompletionNotificationListener) : Job {
        return JobBuilder("Job1",jobRepository)
            .listener(listener)
            .start(step1)
            .build()
    }

    @Bean
    fun transactionManager() : PlatformTransactionManager{
        var transection = DataSourceTransactionManager(dataSource)
        return transection
    }
}

Data_Batch.kt #

@Entity(name = "Data_Batch")
class Data_Batch {
    @Id
    @GeneratedValue
    @Column(name = "batch_code", nullable = false)
    open var batch_code: Long? = null

    @Column(name = "batch_name", nullable = false)
    open var batch_name : String? = null
}

Data_Batch_ETL.kt #

@Entity(name = "Data_Batch_ETL")
class Data_Batch_ETL {
    @Id
    @GeneratedValue
    @Column(name = "batch_code", nullable = false)
    open var batch_code: Long? = null

    @Column(name = "batch_name", nullable = false)
    open var batch_name : String? = null

    @Column(name = "batch_source")
    open var batch_source : String? = null
}

Batch_Mapper.kt #

@Component
class Batch_Mapper : RowMapper<Data_Batch> {
    companion object{
        open val batch_code : String = "batch_code"
        open val batch_name : String = "batch_name"
    }

    override fun mapRow(rs: ResultSet, rowNum: Int): Data_Batch {
        var item = Data_Batch()
        item.batch_code = rs.getLong(batch_code)
        item.batch_name = rs.getString(batch_name)
        return item
    }
}

Batch_Procesor.kt #

@Component
class Batch_Procesor : ItemProcessor<Data_Batch,Data_Batch_ETL>{
    override fun process(item: Data_Batch): Data_Batch_ETL {
        val transform : Data_Batch_ETL = Data_Batch_ETL()
        transform.batch_code = item.batch_code!!
        transform.batch_name = item.batch_name!!.uppercase()
        transform.batch_source = "Data_Batch"
        return transform
    }
}

application.properties #

spring.application.name=spring_batch

spring.jpa.hibernate.ddl-auto=update
spring.jpa.generate-ddl=true
server.port=8094

spring.datasource.url=jdbc:postgresql://localhost:5432/backend2
spring.datasource.username=postgres
spring.datasource.password=password
spring.jpa.show-sql=true
logging.level.org.hibernate.engine.transaction.internal.TransactionImpl=DEBUG
spring.jpa.properties.hibernate.dialect=org.hibernate.dialect.PostgreSQLDialect
spring.datasource.driver-class-name=org.postgresql.Driver

server.shutdown=graceful
spring.lifecycle.timeout-per-shutdown-phase=60s

spring.batch.job.enabled: true
spring.batch.jdbc.initialize-schema=always

内容的提问来源于stack exchange，提问作者Rabbit DT