Make chunk tracker thread bound in ChunkOrientedStep

Before this commit, the chunk tracker was instance-based,
which resulted in progress state being shared and overridden
across threads of execution.

This commit changes the scope of the chunk tracker to be
thread local so that each thread has its own progress state.

Resolves #5099

Author: fmbenhassine

spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkOrientedStep.java

@@ -131,7 +131,7 @@ public class ChunkOrientedStep<I, O> extends AbstractStep {
 	 */
 	private final int chunkSize;
 
-	private final ChunkTracker chunkTracker = new ChunkTracker();
+	private final ThreadLocal<ChunkTracker> chunkTracker = ThreadLocal.withInitial(ChunkTracker::new);
 
 	private final CompositeChunkListener<I, O> compositeChunkListener = new CompositeChunkListener<>();
 
@@ -359,7 +359,7 @@ protected void close(ExecutionContext executionContext) throws Exception {
 	@Override
 	protected void doExecute(StepExecution stepExecution) throws Exception {
 		stepExecution.getExecutionContext().put(STEP_TYPE_KEY, this.getClass().getName());
-		while (this.chunkTracker.moreItems() && !interrupted(stepExecution)) {
+		while (this.chunkTracker.get().moreItems() && !interrupted(stepExecution)) {
 			// process next chunk in its own transaction
 			this.transactionTemplate.executeWithoutResult(transactionStatus -> {
 				ChunkTransactionEvent chunkTransactionEvent = new ChunkTransactionEvent(stepExecution.getStepName(),
@@ -389,7 +389,7 @@ private void processChunkConcurrently(TransactionStatus status, StepContribution
 		List<Future<O>> itemProcessingTasks = new LinkedList<>();
 		try {
 			// read items and submit concurrent item processing tasks
-			for (int i = 0; i < this.chunkSize && this.chunkTracker.moreItems(); i++) {
+			for (int i = 0; i < this.chunkSize && this.chunkTracker.get().moreItems(); i++) {
 				I item = readItem(contribution);
 				if (item != null) {
 					Future<O> itemProcessingFuture = this.taskExecutor.submit(() -> processItem(item, contribution));
@@ -480,7 +480,7 @@ private boolean interrupted(StepExecution stepExecution) {
 
 	private Chunk<I> readChunk(StepContribution contribution) throws Exception {
 		Chunk<I> chunk = new Chunk<>();
-		for (int i = 0; i < chunkSize && this.chunkTracker.moreItems(); i++) {
+		for (int i = 0; i < chunkSize && this.chunkTracker.get().moreItems(); i++) {
 			I item = readItem(contribution);
 			if (item != null) {
 				chunk.add(item);
@@ -505,7 +505,7 @@ private Chunk<I> readChunk(StepContribution contribution) throws Exception {
 			this.compositeItemReadListener.beforeRead();
 			item = doRead();
 			if (item == null) {
-				this.chunkTracker.noMoreItems();
+				this.chunkTracker.get().noMoreItems();
 			}
 			else {
 				contribution.incrementReadCount();

spring-batch-samples/src/main/java/org/springframework/batch/samples/partitioning/local/PartitionJdbcJobConfiguration.java

@@ -0,0 +1,86 @@
+/*
+ * Copyright 2025-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.samples.partitioning.local;
+
+import javax.sql.DataSource;
+
+import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
+import org.springframework.batch.core.configuration.annotation.EnableJdbcJobRepository;
+import org.springframework.batch.core.configuration.annotation.StepScope;
+import org.springframework.batch.core.job.Job;
+import org.springframework.batch.core.job.builder.JobBuilder;
+import org.springframework.batch.core.repository.JobRepository;
+import org.springframework.batch.core.step.Step;
+import org.springframework.batch.core.step.builder.StepBuilder;
+import org.springframework.batch.infrastructure.item.database.JdbcCursorItemReader;
+import org.springframework.batch.infrastructure.item.database.builder.JdbcCursorItemReaderBuilder;
+import org.springframework.batch.samples.common.ColumnRangePartitioner;
+import org.springframework.batch.samples.common.DataSourceConfiguration;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.core.task.SimpleAsyncTaskExecutor;
+import org.springframework.jdbc.core.DataClassRowMapper;
+
+@Configuration
+@EnableBatchProcessing
+@EnableJdbcJobRepository
+@Import(DataSourceConfiguration.class)
+public class PartitionJdbcJobConfiguration {
+
+	record Owner(int id, String firstName, String lastName) {
+	}
+
+	@Bean
+	@StepScope
+	public JdbcCursorItemReader<Owner> ownersReader(DataSource dataSource,
+			@Value("#{stepExecutionContext['minValue']}") int minValue,
+			@Value("#{stepExecutionContext['maxValue']}") int maxValue) {
+		String query = String.format("SELECT * FROM OWNERS WHERE ID BETWEEN %s AND %s", minValue, maxValue);
+		return new JdbcCursorItemReaderBuilder<Owner>().name("ownersReader")
+			.sql(query)
+			.dataSource(dataSource)
+			.rowMapper(new DataClassRowMapper<>(Owner.class))
+			.build();
+	}
+
+	@Bean
+	public Step partitionedStep(JobRepository jobRepository, DataSource dataSource,
+			JdbcCursorItemReader<Owner> ownersReader) {
+		Step workerStep = new StepBuilder("workerStep", jobRepository).<Owner, Owner>chunk(3)
+			.reader(ownersReader)
+			.writer(chunk -> System.out
+				.println(Thread.currentThread().getName() + " - writing chunk: " + chunk.getItems()))
+			.build();
+
+		ColumnRangePartitioner ownersPartitioner = new ColumnRangePartitioner();
+		ownersPartitioner.setColumn("ID");
+		ownersPartitioner.setDataSource(dataSource);
+		ownersPartitioner.setTable("OWNERS");
+		return new StepBuilder(jobRepository).partitioner("workerStep", ownersPartitioner)
+			.step(workerStep)
+			.gridSize(2)
+			.taskExecutor(new SimpleAsyncTaskExecutor())
+			.build();
+	}
+
+	@Bean
+	public Job job(JobRepository jobRepository, Step partitionedStep) {
+		return new JobBuilder(jobRepository).start(partitionedStep).build();
+	}
+
+}

spring-batch-samples/src/main/java/org/springframework/batch/samples/partitioning/local/README.md

@@ -4,19 +4,26 @@
 
 The purpose of this sample is to show multi-threaded step execution
 using the `PartitionHandler` SPI.  The example uses a
-`TaskExecutorPartitionHandler` to spread the work of reading
-some files across multiple threads, with one `Step` execution
-per thread.  The key components are the `PartitionStep` and the
-`MultiResourcePartitioner` which is responsible for dividing up
-the work.  Notice that the readers and writers in the `Step`
-that is being partitioned are step-scoped, so that their state does
-not get shared across threads of execution.
+`TaskExecutorPartitionHandler` to spread the work across multiple threads,
+with one `Step` execution per thread.
+
+For the XML sample, the configuration is in `/org/springframework/batch/samples/partition/jdbc/partitionJdbcJob.xml`.
+The key components are the `PartitionStep` and the `MultiResourcePartitioner` which is responsible for dividing up
+the work. 
+
+For the Java sample, the configuration is in `org.springframework.batch.samples.partitioning.local.PartitionJdbcJobConfiguration`.
+The key component is the `ColumnRangePartitioner` which is responsible for dividing up the work.
+
+Notice that the readers and writers in the `Step` that is being partitioned are step-scoped, so that their state does not get shared across threads of execution.
 
 ### Run the sample
 
 You can run the sample from the command line as following:
 
 ```
 $>cd spring-batch-samples
-$>../mvnw -Dtest=MailJobFunctionalTests#testLaunchJob test
+# Launch the sample using the XML configuration
+$>../mvnw -Dtest=PartitionJdbcJobFunctionalTests#testUpdateCredit test
+# Launch the sample using the Java configuration
+$>../mvnw -Dtest=PartitionJdbcJobFunctionalTests#testLaunchJobWithJavaConfiguration test
 ```

spring-batch-samples/src/test/java/org/springframework/batch/samples/partition/jdbc/PartitionJdbcJobFunctionalTests.java

@@ -16,29 +16,30 @@
 
 package org.springframework.batch.samples.partition.jdbc;
 
-import java.util.ArrayList;
-import java.util.LinkedHashSet;
-import java.util.List;
-import java.util.Set;
+import java.util.*;
 
+import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.batch.core.BatchStatus;
 import org.springframework.batch.core.job.Job;
 import org.springframework.batch.core.job.JobExecution;
 import org.springframework.batch.core.job.parameters.JobParameters;
 import org.springframework.batch.core.launch.JobOperator;
+import org.springframework.batch.core.step.StepExecution;
 import org.springframework.batch.infrastructure.item.ExecutionContext;
 import org.springframework.batch.infrastructure.item.ItemReader;
 import org.springframework.batch.infrastructure.item.ItemStream;
 import org.springframework.batch.samples.domain.trade.CustomerCredit;
 import org.springframework.batch.samples.domain.trade.internal.CustomerCreditIncreaseProcessor;
+import org.springframework.batch.samples.partitioning.local.PartitionJdbcJobConfiguration;
 import org.springframework.batch.test.JobOperatorTestUtils;
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
+import org.springframework.context.annotation.AnnotationConfigApplicationContext;
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -97,6 +98,26 @@ void testUpdateCredit() throws Exception {
 		}
 	}
 
+	@Test
+	void testLaunchJobWithJavaConfiguration() throws Exception {
+		// given
+		ApplicationContext context = new AnnotationConfigApplicationContext(PartitionJdbcJobConfiguration.class);
+		JobOperator jobOperator = context.getBean(JobOperator.class);
+		Job job = context.getBean(Job.class);
+
+		// when
+		JobExecution jobExecution = jobOperator.start(job, new JobParameters());
+
+		// then
+		assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus());
+		Collection<StepExecution> stepExecutions = jobExecution.getStepExecutions();
+		// one manager step + 2 worker steps
+		Assertions.assertEquals(3, stepExecutions.size());
+		StepExecution managerStepExecution = stepExecutions.iterator().next();
+		Assertions.assertEquals(10, managerStepExecution.getReadCount());
+		Assertions.assertEquals(10, managerStepExecution.getWriteCount());
+	}
+
 	/**
 	 * Read all credits using the provided reader.
 	 */