Introduce ChunkTaskExecutorItemWriter to support local chunking

Resolves #5021

Author: fmbenhassine

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

@@ -16,23 +16,31 @@
 
 package org.springframework.batch.core.step.item;
 
-import org.jspecify.annotations.NullUnmarked;
-
 import org.springframework.batch.core.step.StepContribution;
 import org.springframework.batch.infrastructure.item.Chunk;
 
 /**
  * Interface defined for processing {@link Chunk}s.
  *
+ * @author Mahmoud Ben Hassine
  * @author Kyeonghoon Lee (Add FunctionalInterface annotation)
  * @since 2.0
- * @deprecated Since 6.0 with no replacement. Scheduled for removal in 7.0.
  */
-@NullUnmarked
-@Deprecated(since = "6.0", forRemoval = true)
 @FunctionalInterface
 public interface ChunkProcessor<I> {
 
-	void process(StepContribution contribution, Chunk<I> chunk) throws Exception;
+	@Deprecated(since = "6.0", forRemoval = true)
+	default void process(StepContribution contribution, Chunk<I> chunk) throws Exception {
+		process(chunk, contribution);
+	}
+
+	/**
+	 * Process the given chunk and update the contribution.
+	 * @param chunk the chunk to process
+	 * @param contribution the current step contribution
+	 * @throws Exception if there is any error during processing
+	 * @since 6.0
+	 */
+	void process(Chunk<I> chunk, StepContribution contribution) throws Exception;
 
 }

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

@@ -212,7 +212,7 @@ protected void writeItems(Chunk<O> items) throws Exception {
 	}
 
 	@Override
-	public final void process(StepContribution contribution, Chunk<I> inputs) throws Exception {
+	public final void process(Chunk<I> inputs, StepContribution contribution) throws Exception {
 
 		// Allow temporary state to be stored in the user data field
 		initializeUserData(inputs);

spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ChunkOrientedTaskletTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2006-2023 the original author or authors.
+ * Copyright 2006-2025 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.
@@ -52,7 +52,7 @@ public Chunk<String> provide(StepContribution contribution) throws Exception {
 			@Override
 			public void postProcess(StepContribution contribution, Chunk<String> chunk) {
 			}
-		}, (contribution, chunk) -> contribution.incrementWriteCount(1));
+		}, (chunk, contribution) -> contribution.incrementWriteCount(1));
 		StepContribution contribution = new StepContribution(
 				new StepExecution("foo", new JobExecution(1L, new JobInstance(123L, "job"), new JobParameters())));
 		handler.execute(contribution, context);
@@ -95,7 +95,7 @@ public Chunk<String> provide(StepContribution contribution) throws Exception {
 			@Override
 			public void postProcess(StepContribution contribution, Chunk<String> chunk) {
 			}
-		}, (contribution, chunk) -> contribution.incrementWriteCount(1));
+		}, (chunk, contribution) -> contribution.incrementWriteCount(1));
 		StepContribution contribution = new StepContribution(
 				new StepExecution("foo", new JobExecution(1L, new JobInstance(123L, "job"), new JobParameters())));
 		ExitStatus expected = contribution.getExitStatus();

spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkTaskExecutorItemWriter.java

@@ -0,0 +1,105 @@
+/*
+ * 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.integration.chunk;
+
+import org.springframework.batch.core.ExitStatus;
+import org.springframework.batch.core.step.StepContribution;
+import org.springframework.batch.core.step.StepExecution;
+import org.springframework.batch.core.listener.StepExecutionListener;
+import org.springframework.batch.core.step.item.ChunkProcessor;
+import org.springframework.batch.infrastructure.item.Chunk;
+import org.springframework.batch.infrastructure.item.ItemWriter;
+import org.springframework.core.task.TaskExecutor;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.FutureTask;
+
+/**
+ * Similar to {@code ChunkMessageChannelItemWriter}, this item writer submits chunk
+ * requests to local workers from a {@link TaskExecutor} instead of sending them over a
+ * message channel to remote workers.
+ *
+ * @param <T> type of items
+ * @see ChunkMessageChannelItemWriter
+ * @author Mahmoud Ben Hassine
+ * @since 6.0
+ */
+public class ChunkTaskExecutorItemWriter<T> implements ItemWriter<T>, StepExecutionListener {
+
+	@SuppressWarnings("NullAway.Init")
+	private StepExecution stepExecution;
+
+	private int sequence;
+
+	private final TaskExecutor taskExecutor;
+
+	private final ChunkProcessorChunkRequestHandler<T> chunkProcessorChunkHandler = new ChunkProcessorChunkRequestHandler<>();
+
+	private final Set<Future<ChunkResponse>> responses = new HashSet<>();
+
+	/**
+	 * Create a new {@link ChunkTaskExecutorItemWriter}.
+	 * @param chunkRequestProcessor the chunk processor to process chunks
+	 * @param taskExecutor the task executor to submit chunk processing tasks to
+	 */
+	public ChunkTaskExecutorItemWriter(ChunkProcessor<T> chunkRequestProcessor, TaskExecutor taskExecutor) {
+		this.taskExecutor = taskExecutor;
+		this.chunkProcessorChunkHandler.setChunkProcessor(chunkRequestProcessor);
+	}
+
+	@Override
+	public void write(Chunk<? extends T> chunk) {
+		ChunkRequest<T> request = new ChunkRequest<>(++sequence, chunk, this.stepExecution.getJobExecutionId(),
+				this.stepExecution.createStepContribution());
+		FutureTask<ChunkResponse> chunkResponseFutureTask = new FutureTask<>(
+				() -> this.chunkProcessorChunkHandler.handle(request));
+		this.responses.add(chunkResponseFutureTask);
+		this.taskExecutor.execute(chunkResponseFutureTask);
+	}
+
+	@Override
+	public void beforeStep(StepExecution stepExecution) {
+		this.stepExecution = stepExecution;
+	}
+
+	@Override
+	public ExitStatus afterStep(StepExecution stepExecution) {
+		try {
+			for (StepContribution contribution : getStepContributions()) {
+				stepExecution.apply(contribution);
+			}
+		}
+		catch (ExecutionException | InterruptedException e) {
+			return ExitStatus.FAILED.addExitDescription(e);
+		}
+		return ExitStatus.COMPLETED.addExitDescription("Waited for " + this.responses.size() + " results.");
+	}
+
+	private Collection<StepContribution> getStepContributions() throws ExecutionException, InterruptedException {
+		List<StepContribution> contributions = new ArrayList<>();
+		for (Future<ChunkResponse> task : this.responses) {
+			contributions.add(task.get().getStepContribution());
+		}
+		return contributions;
+	}
+
+}

spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkRequestHandlerTests.java

@@ -33,7 +33,7 @@ class ChunkProcessorChunkRequestHandlerTests {
 	@Test
 	void testVanillaHandleChunk() throws Exception {
 		// given
-		handler.setChunkProcessor((contribution, chunk) -> count += chunk.size());
+		handler.setChunkProcessor((chunk, contribution) -> count += chunk.size());
 		StepContribution stepContribution = MetaDataInstanceFactory.createStepExecution().createStepContribution();
 		Chunk items = Chunk.of("foo", "bar");
 		ChunkRequest chunkRequest = new ChunkRequest<>(0, items, 12L, stepContribution);

spring-batch-samples/src/main/java/org/springframework/batch/samples/chunking/local/LocalChunkingJobConfiguration.java

@@ -0,0 +1,112 @@
+/*
+ * 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.chunking.local;
+
+import javax.sql.DataSource;
+
+import org.springframework.batch.core.ExitStatus;
+import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
+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.core.step.item.ChunkProcessor;
+import org.springframework.batch.infrastructure.item.database.JdbcBatchItemWriter;
+import org.springframework.batch.infrastructure.item.database.builder.JdbcBatchItemWriterBuilder;
+import org.springframework.batch.infrastructure.item.file.FlatFileItemReader;
+import org.springframework.batch.infrastructure.item.file.builder.FlatFileItemReaderBuilder;
+import org.springframework.batch.integration.chunk.ChunkTaskExecutorItemWriter;
+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.io.Resource;
+import org.springframework.jdbc.support.JdbcTransactionManager;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import org.springframework.transaction.support.TransactionTemplate;
+
+@Configuration
+@EnableBatchProcessing
+@Import(DataSourceConfiguration.class)
+public class LocalChunkingJobConfiguration {
+
+	public record Vet(String firstname, String lastname) {
+	}
+
+	@Bean
+	public Job job(JobRepository jobRepository, Step chunkingStep) {
+		return new JobBuilder("job", jobRepository).start(chunkingStep).build();
+	}
+
+	@Bean
+	public Step chunkingStep(JobRepository jobRepository, JdbcTransactionManager transactionManager,
+			FlatFileItemReader<Vet> itemReader, ChunkTaskExecutorItemWriter<Vet> itemWriter) {
+		return new StepBuilder("chunkingStep", jobRepository).<Vet, Vet>chunk(2)
+			.transactionManager(transactionManager)
+			.reader(itemReader)
+			.writer(itemWriter)
+			.build();
+	}
+
+	@Bean
+	@StepScope
+	public FlatFileItemReader<Vet> itemReader(@Value("#{jobParameters['inputFile']}") Resource resource) {
+		return new FlatFileItemReaderBuilder<Vet>().name("vetItemReader")
+			.resource(resource)
+			.delimited()
+			.names("firstname", "lastname")
+			.targetType(Vet.class)
+			.build();
+	}
+
+	@Bean
+	public ChunkTaskExecutorItemWriter<Vet> itemWriter(ChunkProcessor<Vet> chunkProcessor) {
+		ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
+		taskExecutor.setCorePoolSize(4);
+		taskExecutor.setThreadNamePrefix("worker-thread-");
+		taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
+		taskExecutor.afterPropertiesSet();
+		return new ChunkTaskExecutorItemWriter<>(chunkProcessor, taskExecutor);
+	}
+
+	@Bean
+	public ChunkProcessor<Vet> chunkProcessor(DataSource dataSource, TransactionTemplate transactionTemplate) {
+		String sql = "insert into vets (firstname, lastname) values (?, ?)";
+		JdbcBatchItemWriter<Vet> itemWriter = new JdbcBatchItemWriterBuilder<Vet>().dataSource(dataSource)
+			.sql(sql)
+			.itemPreparedStatementSetter((item, ps) -> {
+				ps.setString(1, item.firstname());
+				ps.setString(2, item.lastname());
+			})
+			.build();
+
+		return (chunk, contribution) -> transactionTemplate.executeWithoutResult(transactionStatus -> {
+			try {
+				itemWriter.write(chunk);
+				contribution.incrementWriteCount(chunk.size());
+				contribution.setExitStatus(ExitStatus.COMPLETED);
+			}
+			catch (Exception e) {
+				transactionStatus.setRollbackOnly();
+				contribution.setExitStatus(ExitStatus.FAILED.addExitDescription(e));
+			}
+		});
+	}
+
+}

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

@@ -0,0 +1,23 @@
+## Local Chunking Sample
+
+### About
+
+This sample shows how to use the `ChunkTaskExecutorItemWriter` to write items in a
+multi-threaded way.  The example uses a `ThreadPoolTaskExecutor` to write items
+in parallel with multiple threads, each one writing a chunk of items in its own transaction.
+
+In this sample, data items of type `Vet` are read from a flat file and written to
+a relational database table. The `Vet` class is a simple domain object representing a
+veterinarian, with properties `firstName`, and `lastName`. The flat file contains records
+of veterinarians, and each record is mapped to a `Vet` object using a `FlatFileItemReader`.
+The `JdbcBatchItemWriter` is then used to persist these `Vet` objects into a relational database
+table named `vets`. The table has columns corresponding to the properties of the `Vet` class.
+
+### Run the sample
+
+You can run the sample from the command line as following:
+
+```
+$>cd spring-batch-samples
+$>../mvnw -Dtest=LocalChunkingJobFunctionalTests#testLaunchJobWithJavaConfig test
+```

spring-batch-samples/src/main/java/org/springframework/batch/samples/common/DataSourceConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021-2023 the original author or authors.
+ * Copyright 2021-2025 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.
@@ -20,8 +20,10 @@
 
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.jdbc.support.JdbcTransactionManager;
 import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
+import org.springframework.transaction.support.TransactionTemplate;
 
 /**
  * @author Mahmoud Ben Hassine
@@ -38,9 +40,19 @@ public DataSource dataSource() {
 			.build();
 	}
 
+	@Bean
+	public JdbcTemplate jdbcTemplate(DataSource dataSource) {
+		return new JdbcTemplate(dataSource);
+	}
+
 	@Bean
 	public JdbcTransactionManager transactionManager(DataSource dataSource) {
 		return new JdbcTransactionManager(dataSource);
 	}
 
+	@Bean
+	public TransactionTemplate transactionTemplate(JdbcTransactionManager transactionManager) {
+		return new TransactionTemplate(transactionManager);
+	}
+
 }

spring-batch-samples/src/main/resources/org/springframework/batch/samples/chunking/local/data/vets.csv

@@ -0,0 +1,6 @@
+foo1,bar1
+foo2,bar2
+foo3,bar3
+foo4,bar4
+foo5,bar5
+foo6,bar6

spring-batch-samples/src/main/resources/org/springframework/batch/samples/common/business-schema-hsqldb.sql

@@ -10,6 +10,7 @@ DROP TABLE  GAMES IF EXISTS;
 DROP TABLE  PLAYER_SUMMARY IF EXISTS;
 DROP TABLE  ERROR_LOG IF EXISTS;
 DROP TABLE  OWNERS IF EXISTS;
+DROP TABLE VETS IF EXISTS;
 
 -- Autogenerated: do not edit this file
 
@@ -104,6 +105,12 @@ CREATE TABLE ERROR_LOG (
 
 -- PetClinic sample tables
 
+CREATE TABLE VETS (
+  ID        INTEGER IDENTITY PRIMARY KEY,
+  FIRSTNAME VARCHAR(30),
+  LASTNAME  VARCHAR(30)
+);
+
 CREATE TABLE OWNERS (
     ID         INTEGER IDENTITY PRIMARY KEY,
     FIRSTNAME  VARCHAR(30),