Spring Batch分区作业报错：Reader/Processor/Writer抛出ScopeNotActiveException

问题背景 #

需实现从数据库A读取数据、转换后写入数据库B的Spring Batch分区作业，已完成非分区版本，但分区版作业中Reader、Processor、Writer均抛出ScopeNotActiveException，组件未正常调用。

异常信息 #

• Reader：Method threw 'org.springframework.beans.factory.support.ScopeNotActiveException' exception. Cannot evaluate org.springframework.batch.item.data.RepositoryItemReader$$SpringCGLIB$$0.toString()
• Processor：Method threw 'org.springframework.beans.factory.support.ScopeNotActiveException' exception. Cannot evaluate jdk.proxy2.$Proxy108.toString()
• Writer：Method threw 'org.springframework.beans.factory.support.ScopeNotActiveException' exception. Cannot evaluate jdk.proxy2.$Proxy109.toString()

配置代码 #

@Configuration 
@EnableBatchProcessing 
public class NewPartitionedBatchConfig {
    
    private final CustomerRepository customerRepository;
    private final SourceMigrationDataChanger sourceMigrationDataChanger;
    
    public NewPartitionedBatchConfig(
        CustomerRepository customerRepository,
        SourceMigrationDataChanger sourceMigrationDataChanger
    ) {
        this.customerRepository = customerRepository;
        this.sourceMigrationDataChanger = sourceMigrationDataChanger;
    }
    
    @Bean
    public PlatformTransactionManager transactionManager(DataSource dataSource) {
        return new DataSourceTransactionManager(dataSource);
    }
    
    @Bean
    public Job sampleJob(
        JobRepository jobRepository,
        Step partitionedStep
    ) {
        return new JobBuilder("sampleJob", jobRepository)
            .start(partitionedStep)
            .build();
    }
    
    // i've tried to declare it this way
    //    @Bean
    //    public Partitioner partitioner() {
        //        CustomPartitioner partitioner = new CustomPartitioner();
        //        partitioner.partition(GRID_SIZE);
        //        return partitioner;
    //    }
    
    @Bean
    public PartitionHandler partitionHandler(
        JobExecutorConfig jobExecutorConfig,
        Step workerStep
    ) {
        TaskExecutorPartitionHandler handler = new TaskExecutorPartitionHandler();
        //        handler.setTaskExecutor(jobExecutorConfig.threadPoolTaskExecutor()); //i've made a ThreadPoolTaskExecutor too
        handler.setTaskExecutor(jobExecutorConfig.simpleAsyncTaskExecutor());
        handler.setStep(workerStep);
        handler.setGridSize(GRID_SIZE);
        return handler;
    }
    
    @Bean
    public Step partitionedStep(
        JobRepository jobRepository,
        CustomPartitioner customPartitioner,
        //            BasicPartitioner basicPartitioner, //i've tried Basic Partitioner too
        PartitionHandler partitionHandler
    ) {
        return new StepBuilder("partitionedStep", jobRepository)
            .partitioner("workerStep", customPartitioner)
            //                .step(workerStep()) //i've tried to put a step directly without partitionedHandler already
            .partitionHandler(partitionHandler)
            .build();
    }
    
    @Bean
    public Step workerStep(
        PlatformTransactionManager transactionManager,
        JobRepository jobRepository,
        RepositoryItemReader<Customer> itemReader,
        ItemProcessor<Object, Map<String, Object>> itemProcessor,
        ItemWriter<Map<String, Object>> itemWriter
    ) {
        return new StepBuilder("workerStep", jobRepository)
            .<Customer, Map<String, Object>>chunk(BATCH_SIZE, transactionManager)
            .reader(itemReader)
            .processor(itemProcessor)
            .writer(itemWriter)
            .build();
    }
    
    @Bean
    @StepScope
    public RepositoryItemReader<Customer> itemReader(
        @Value("#{stepExecutionContext['partitionIndex'] ?: 0}") Integer partitionIndex,
        @Value("#{stepExecutionContext['gridSize']}") Integer gridSize
    ) {
        if (partitionIndex == null) {
            throw new IllegalStateException("partitionIndex is null. Ensure that the step execution context is properly set.");
        }
        
        RepositoryItemReader<Customer> reader = new RepositoryItemReader<>();
        reader.setRepository(customerRepository);
        //        reader.setMethodName("findCustomersByPage"); using JPA Pageable
        reader.setMethodName("findCustomersByPartition");
        reader.setArguments(Arrays.asList(partitionIndex, gridSize));
        reader.setPageSize(BATCH_SIZE);
        reader.setSort(Collections.singletonMap("id", Sort.Direction.ASC));
        return reader;
    }
    
    @Bean
    @StepScope
    public ItemProcessor<Object, Map<String, Object>> itemProcessor() {
        return customer -> {
            sourceMigrationDataChanger.updateCustomerMigrated((Customer) customer);
            
            Map<String, Object> defaultKafkaHash = new HashMap<>();
            defaultKafkaHash.put(KEY_DATA, customer);
            return defaultKafkaHash;
        };
    }
    
    @Bean
    @StepScope
    public ItemWriter<Map<String, Object>> itemWriter() {
        return items -> {
            Map<String, Object> defaultKafkaHash = new HashMap<>();
            defaultKafkaHash.put(KEY_DATA, items);
            //            it used to send some data through kafka, but omit this for now
        };
    }
}

自定义Partitioner代码 #

@Getter
@Component
public class CustomPartitioner implements Partitioner {
    
//    public Map<String, ExecutionContext> partitions = new HashMap<>();
    
    @Override
    public Map<String, ExecutionContext> partition(int gridSize) {
        System.out.println("Partitioning with gridSize: " + gridSize);
        
        gridSize=2;
        
        Map<String, ExecutionContext> partitions = new HashMap<>();
        for (int i = 0; i < gridSize; i++) {
            ExecutionContext context = new ExecutionContext();
            context.putInt("partitionIndex", i);
            context.putInt("gridSize", gridSize);
            context.putInt("startIndex", i * BATCH_SIZE);
            context.putInt("pageSize", BATCH_SIZE);
            
            partitions.put(PARTITION_KEY + i, context);
        }
        return partitions;
    }
}

排查思路 #

1. StepScope组件过早初始化 #

workerStep是单例Bean，直接注入了@StepScope标记的reader/processor/writer，容器启动时Spring会尝试初始化这些StepScope组件，但此时StepExecutionContext尚未创建，触发ScopeNotActiveException。

• 解决：给workerStep添加@StepScope注解，或者使用Provider<RepositoryItemReader<Customer>>这种延迟加载方式注入组件，避免提前初始化。

2. Partitioner与PartitionHandler的GridSize不统一 #

CustomPartitioner硬编码了gridSize=2，但PartitionHandler使用的是GRID_SIZE常量，两者不一致会导致分区上下文传递异常。

• 解决：移除Partitioner中的硬编码逻辑，直接使用方法传入的gridSize参数。

3. Partitioner的@Component注解冲突 #

CustomPartitioner被标记为@Component，会被Spring作为单例管理，分区步骤中需要动态生成分区上下文，单例可能导致上下文复用或传递异常。

• 解决：在NewPartitionedBatchConfig中用@Bean声明CustomPartitioner，替代@Component自动注册，确保实例按需创建。

4. TaskExecutor线程上下文传递缺失 #

使用SimpleAsyncTaskExecutor时，子线程无法继承父线程的StepExecutionContext，导致StepScope组件无法获取上下文属性。

• 解决：改用ThreadPoolTaskExecutor，并配置TaskDecorator传递上下文；或者使用Spring Batch提供的AsyncTaskExecutorWrapper来处理上下文传递。

5. StepExecutionContext属性默认值风险 #

Reader中@Value("#{stepExecutionContext['partitionIndex'] ?: 0}")设置了默认值0，若分区上下文未正确传递，会使用默认值但gridSize可能为null，引发后续异常。

• 解决：移除默认值，确保partitionIndex和gridSize从分区上下文必传；或者添加gridSize的非空校验，避免NPE。

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