Fix chunk processing on skip in read

Before this commit, when a skip occurs on read,
the step was prematurely stopping chunk reading
(ie the chunk size was not honored).

This commit fixes that by ensuring that the step
continues reading next items even when a skip occurs
during the read operation.

Resolves #5084

Author: fmbenhassine

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

@@ -386,7 +386,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; i++) {
+			for (int i = 0; i < this.chunkSize && this.chunkTracker.moreItems(); i++) {
 				I item = readItem(contribution);
 				if (item != null) {
 					Future<O> itemProcessingFuture = this.taskExecutor.submit(() -> processItem(item, contribution));
@@ -477,14 +477,11 @@ private boolean interrupted(StepExecution stepExecution) {
 
 	private Chunk<I> readChunk(StepContribution contribution) throws Exception {
 		Chunk<I> chunk = new Chunk<>();
-		for (int i = 0; i < chunkSize; i++) {
+		for (int i = 0; i < chunkSize && this.chunkTracker.moreItems(); i++) {
 			I item = readItem(contribution);
 			if (item != null) {
 				chunk.add(item);
 			}
-			else {
-				break;
-			}
 		}
 		return chunk;
 	}

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

@@ -15,8 +15,10 @@
  */
 package org.springframework.batch.core.step.item;
 
+import java.util.List;
 import java.util.Set;
 
+import org.jspecify.annotations.Nullable;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
@@ -31,12 +33,14 @@
 import org.springframework.batch.core.step.Step;
 import org.springframework.batch.core.step.StepExecution;
 import org.springframework.batch.core.step.builder.ChunkOrientedStepBuilder;
+import org.springframework.batch.core.step.skip.AlwaysSkipItemSkipPolicy;
 import org.springframework.batch.core.step.skip.LimitCheckingExceptionHierarchySkipPolicy;
 import org.springframework.batch.core.step.skip.SkipLimitExceededException;
 import org.springframework.batch.infrastructure.item.ItemProcessor;
 import org.springframework.batch.infrastructure.item.ItemReader;
 import org.springframework.batch.infrastructure.item.ItemWriter;
 import org.springframework.batch.infrastructure.item.file.FlatFileParseException;
+import org.springframework.batch.infrastructure.item.support.ListItemReader;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.annotation.AnnotationConfigApplicationContext;
 import org.springframework.context.annotation.Bean;
@@ -188,6 +192,105 @@ void testConcurrentFaultTolerantChunkOrientedStepFailure() throws Exception {
 		System.clearProperty("skipLimit");
 	}
 
+	// Issue https://github.com/spring-projects/spring-batch/issues/5084
+	@Test
+	void testSkipInReadInSequentialMode() throws Exception {
+		// given
+		ApplicationContext context = new AnnotationConfigApplicationContext(TestConfiguration.class,
+				StepConfiguration.class);
+		JobOperator jobOperator = context.getBean(JobOperator.class);
+		Job job = context.getBean(Job.class);
+
+		// when
+		JobParameters jobParameters = new JobParametersBuilder().toJobParameters();
+		JobExecution jobExecution = jobOperator.start(job, jobParameters);
+
+		// then
+		Assertions.assertEquals(ExitStatus.COMPLETED.getExitCode(), jobExecution.getExitStatus().getExitCode());
+		StepExecution stepExecution = jobExecution.getStepExecutions().iterator().next();
+		Assertions.assertEquals(8, stepExecution.getReadCount());
+		Assertions.assertEquals(8, stepExecution.getWriteCount());
+		Assertions.assertEquals(2, stepExecution.getCommitCount());
+		Assertions.assertEquals(0, stepExecution.getRollbackCount());
+		Assertions.assertEquals(2, stepExecution.getReadSkipCount());
+		Assertions.assertEquals(0, stepExecution.getWriteSkipCount());
+	}
+
+	// Issue https://github.com/spring-projects/spring-batch/issues/5084
+	@Test
+	void testSkipInReadInConcurrentMode() throws Exception {
+		// given
+		ApplicationContext context = new AnnotationConfigApplicationContext(TestConfiguration.class,
+				ConcurrentStepConfiguration.class);
+		JobOperator jobOperator = context.getBean(JobOperator.class);
+		Job job = context.getBean(Job.class);
+
+		// when
+		JobParameters jobParameters = new JobParametersBuilder().toJobParameters();
+		JobExecution jobExecution = jobOperator.start(job, jobParameters);
+
+		// then
+		Assertions.assertEquals(ExitStatus.COMPLETED.getExitCode(), jobExecution.getExitStatus().getExitCode());
+		StepExecution stepExecution = jobExecution.getStepExecutions().iterator().next();
+		Assertions.assertEquals(8, stepExecution.getReadCount());
+		Assertions.assertEquals(8, stepExecution.getWriteCount());
+		Assertions.assertEquals(2, stepExecution.getCommitCount());
+		Assertions.assertEquals(0, stepExecution.getRollbackCount());
+		Assertions.assertEquals(2, stepExecution.getReadSkipCount());
+		Assertions.assertEquals(0, stepExecution.getWriteSkipCount());
+	}
+
+	@Configuration
+	static class StepConfiguration {
+
+		@Bean
+		public Step step(JobRepository jobRepository, JdbcTransactionManager transactionManager) {
+			List<String> items = List.of("one", "two", "three", "four", "five", "six", "seven", "eight", "nine", "ten");
+			return new ChunkOrientedStepBuilder<String, String>(jobRepository, 5).reader(new ListItemReader<>(items) {
+				@Override
+				public @Nullable String read() {
+					String item = super.read();
+					if ("three".equals(item) || "seven".equals(item)) {
+						throw new RuntimeException("Simulated read error on item: " + item);
+					}
+					return item;
+				}
+			}).writer(chunk -> {
+			})
+				.transactionManager(transactionManager)
+				.faultTolerant()
+				.skipPolicy(new AlwaysSkipItemSkipPolicy())
+				.build();
+		}
+
+	}
+
+	@Configuration
+	static class ConcurrentStepConfiguration {
+
+		@Bean
+		public Step step(JobRepository jobRepository, JdbcTransactionManager transactionManager) {
+			List<String> items = List.of("one", "two", "three", "four", "five", "six", "seven", "eight", "nine", "ten");
+			return new ChunkOrientedStepBuilder<String, String>(jobRepository, 5).reader(new ListItemReader<>(items) {
+				@Override
+				public @Nullable String read() {
+					String item = super.read();
+					if ("three".equals(item) || "seven".equals(item)) {
+						throw new RuntimeException("Simulated read error on item: " + item);
+					}
+					return item;
+				}
+			}).writer(chunk -> {
+			})
+				.transactionManager(transactionManager)
+				.taskExecutor(new SimpleAsyncTaskExecutor())
+				.faultTolerant()
+				.skipPolicy(new AlwaysSkipItemSkipPolicy())
+				.build();
+		}
+
+	}
+
 	@Configuration
 	static class FaultTolerantChunkOrientedStepConfiguration {
 

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

@@ -30,6 +30,7 @@
 import org.springframework.batch.infrastructure.item.ItemReader;
 import org.springframework.batch.infrastructure.item.ItemWriter;
 import org.springframework.batch.infrastructure.item.support.ListItemReader;
+import org.springframework.core.task.SimpleAsyncTaskExecutor;
 
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
@@ -64,7 +65,7 @@ void testFaultTolerantChunkOrientedStepSetupWithDefaultRetryLimit() {
 	}
 
 	@Test
-	void testReadNoMoreThanAvailableItems() throws Exception {
+	void testReadNoMoreThanAvailableItemsInSequentialMode() throws Exception {
 		// given
 		ItemReader<String> reader = mock();
 		ItemWriter<String> writer = chunk -> {
@@ -84,4 +85,26 @@ void testReadNoMoreThanAvailableItems() throws Exception {
 		verify(reader, times(6)).read();
 	}
 
+	@Test
+	void testReadNoMoreThanAvailableItemsInConcurrentMode() throws Exception {
+		// given
+		ItemReader<String> reader = mock();
+		ItemWriter<String> writer = chunk -> {
+		};
+		JobRepository jobRepository = new ResourcelessJobRepository();
+		when(reader.read()).thenReturn("1", "2", "3", "4", "5", null);
+		ChunkOrientedStep<String, String> step = new ChunkOrientedStep<>("step", 10, reader, writer, jobRepository);
+		step.setTaskExecutor(new SimpleAsyncTaskExecutor());
+		step.afterPropertiesSet();
+		JobInstance jobInstance = new JobInstance(1L, "job");
+		JobExecution jobExecution = new JobExecution(1L, jobInstance, new JobParameters());
+		StepExecution stepExecution = new StepExecution(1L, "step", jobExecution);
+
+		// when
+		step.execute(stepExecution);
+
+		// then
+		verify(reader, times(6)).read();
+	}
+
 }