Fix race condition in StepContribution.filterCount

Change filterCount field from long to AtomicLong to ensure
thread-safety when incrementing filter count in concurrent chunk
processing.

Resolves #5188

Signed-off-by: mugeon <pos04167@kakao.com>

Author: KMGeon

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

@@ -17,6 +17,7 @@
 
 import java.io.Serializable;
 import java.util.Objects;
+import java.util.concurrent.atomic.AtomicLong;
 
 import org.springframework.batch.core.ExitStatus;
 
@@ -36,7 +37,7 @@ public class StepContribution implements Serializable {
 
 	private long writeCount = 0;
 
-	private long filterCount = 0;
+	private final AtomicLong filterCount = new AtomicLong(0);
 
 	private final long parentSkipCount;
 
@@ -86,7 +87,7 @@ public void incrementFilterCount() {
 	 * @param count The {@code long} amount to increment by.
 	 */
 	public void incrementFilterCount(long count) {
-		filterCount += count;
+		filterCount.addAndGet(count);
 	}
 
 	/**
@@ -125,7 +126,7 @@ public long getWriteCount() {
 	 * @return the filter counter.
 	 */
 	public long getFilterCount() {
-		return filterCount;
+		return filterCount.get();
 	}
 
 	/**
@@ -220,7 +221,7 @@ public StepExecution getStepExecution() {
 
 	@Override
 	public String toString() {
-		return "[StepContribution: read=" + readCount + ", written=" + writeCount + ", filtered=" + filterCount
+		return "[StepContribution: read=" + readCount + ", written=" + writeCount + ", filtered=" + filterCount.get()
 				+ ", readSkips=" + readSkipCount + ", writeSkips=" + writeSkipCount + ", processSkips="
 				+ processSkipCount + ", exitStatus=" + exitStatus.getExitCode() + "]";
 	}
@@ -229,15 +230,16 @@ public String toString() {
 	public boolean equals(Object o) {
 		if (!(o instanceof StepContribution that))
 			return false;
-		return readCount == that.readCount && writeCount == that.writeCount && filterCount == that.filterCount
-				&& parentSkipCount == that.parentSkipCount && readSkipCount == that.readSkipCount
-				&& writeSkipCount == that.writeSkipCount && processSkipCount == that.processSkipCount
-				&& Objects.equals(stepExecution, that.stepExecution) && Objects.equals(exitStatus, that.exitStatus);
+		return readCount == that.readCount && writeCount == that.writeCount
+				&& filterCount.get() == that.filterCount.get() && parentSkipCount == that.parentSkipCount
+				&& readSkipCount == that.readSkipCount && writeSkipCount == that.writeSkipCount
+				&& processSkipCount == that.processSkipCount && Objects.equals(stepExecution, that.stepExecution)
+				&& Objects.equals(exitStatus, that.exitStatus);
 	}
 
 	@Override
 	public int hashCode() {
-		return Objects.hash(stepExecution, readCount, writeCount, filterCount, parentSkipCount, readSkipCount,
+		return Objects.hash(stepExecution, readCount, writeCount, filterCount.get(), parentSkipCount, readSkipCount,
 				writeSkipCount, processSkipCount, exitStatus);
 	}
 

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

@@ -19,6 +19,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.RepeatedTest;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.batch.core.ExitStatus;
@@ -58,14 +59,14 @@ public class ChunkOrientedStepTests {
 	@Test
 	void testInheritedPropertiesOnBuild() {
 		ChunkOrientedStep<String, String> step = new StepBuilder("step", new ResourcelessJobRepository())
-			.<String, String>chunk(5)
-			.reader(new ListItemReader<>(List.of("foo", "bar")))
-			.writer(items -> {
-			})
-			// inherited properties from StepBuilderHelper
-			.allowStartIfComplete(true)
-			.startLimit(5)
-			.build();
+				.<String, String>chunk(5)
+				.reader(new ListItemReader<>(List.of("foo", "bar")))
+				.writer(items -> {
+				})
+				// inherited properties from StepBuilderHelper
+				.allowStartIfComplete(true)
+				.startLimit(5)
+				.build();
 
 		Assertions.assertTrue(step.isAllowStartIfComplete());
 		Assertions.assertEquals(5, step.getStartLimit());
@@ -74,23 +75,23 @@ void testInheritedPropertiesOnBuild() {
 	@Test
 	void testFaultTolerantChunkOrientedStepSetupWithDefaultSkipLimit() {
 		Assertions.assertDoesNotThrow(() -> new StepBuilder(mock()).chunk(5)
-			.reader(new ListItemReader<>(List.of("item1", "item2")))
-			.writer(items -> {
-			})
-			.faultTolerant()
-			.skip(Exception.class)
-			.build());
+				.reader(new ListItemReader<>(List.of("item1", "item2")))
+				.writer(items -> {
+				})
+				.faultTolerant()
+				.skip(Exception.class)
+				.build());
 	}
 
 	@Test
 	void testFaultTolerantChunkOrientedStepSetupWithDefaultRetryLimit() {
 		Assertions.assertDoesNotThrow(() -> new StepBuilder(mock()).chunk(5)
-			.reader(new ListItemReader<>(List.of("item1", "item2")))
-			.writer(items -> {
-			})
-			.faultTolerant()
-			.retry(Exception.class)
-			.build());
+				.reader(new ListItemReader<>(List.of("item1", "item2")))
+				.writer(items -> {
+				})
+				.faultTolerant()
+				.retry(Exception.class)
+				.build());
 	}
 
 	@Test
@@ -147,14 +148,14 @@ void testRetryLimitWithoutRetryDoesNotRetryErrors() throws Exception {
 
 		ChunkOrientedStep<String, String> step = new ChunkOrientedStepBuilder<String, String>(
 				new ResourcelessJobRepository(), 2)
-			.reader(new ListItemReader<>(List.of("item1")))
-			.processor(processor)
-			.writer(items -> {
-			})
-			.transactionManager(new ResourcelessTransactionManager())
-			.faultTolerant()
-			.retryLimit(3)
-			.build();
+				.reader(new ListItemReader<>(List.of("item1")))
+				.processor(processor)
+				.writer(items -> {
+				})
+				.transactionManager(new ResourcelessTransactionManager())
+				.faultTolerant()
+				.retryLimit(3)
+				.build();
 
 		JobInstance jobInstance = new JobInstance(1L, "job");
 		JobExecution jobExecution = new JobExecution(1L, jobInstance, new JobParameters());
@@ -182,13 +183,13 @@ void testRetryLimitWithoutRetryRetriesExceptions() throws Exception {
 		ListItemWriter<String> listItemWriter = new ListItemWriter<>();
 		ChunkOrientedStep<String, String> step = new ChunkOrientedStepBuilder<String, String>(
 				new ResourcelessJobRepository(), 2)
-			.reader(listItemReader)
-			.processor(processor)
-			.writer(listItemWriter)
-			.transactionManager(new ResourcelessTransactionManager())
-			.faultTolerant()
-			.retryLimit(3)
-			.build();
+				.reader(listItemReader)
+				.processor(processor)
+				.writer(listItemWriter)
+				.transactionManager(new ResourcelessTransactionManager())
+				.faultTolerant()
+				.retryLimit(3)
+				.build();
 
 		JobInstance jobInstance = new JobInstance(1L, "job");
 		JobExecution jobExecution = new JobExecution(1L, jobInstance, new JobParameters());
@@ -216,14 +217,14 @@ void testExplicitRetryConfigurationTakesPrecedence() throws Exception {
 
 		ChunkOrientedStep<String, String> step = new ChunkOrientedStepBuilder<String, String>(
 				new ResourcelessJobRepository(), 2)
-			.reader(listItemReader)
-			.processor(processor)
-			.writer(listItemWriter)
-			.transactionManager(new ResourcelessTransactionManager())
-			.faultTolerant()
-			.retry(IllegalStateException.class)
-			.retryLimit(3)
-			.build();
+				.reader(listItemReader)
+				.processor(processor)
+				.writer(listItemWriter)
+				.transactionManager(new ResourcelessTransactionManager())
+				.faultTolerant()
+				.retry(IllegalStateException.class)
+				.retryLimit(3)
+				.build();
 
 		JobInstance jobInstance = new JobInstance(1L, "job");
 		JobExecution jobExecution = new JobExecution(1L, jobInstance, new JobParameters());
@@ -258,7 +259,7 @@ void testDoSkipInProcessShouldThrowNonSkippableProcessExceptionWhenSkipPolicyRet
 		step.setItemProcessor(processor);
 		step.setFaultTolerant(true);
 		step.setRetryPolicy(RetryPolicy.withMaxRetries(1)); // retry once (initial + 1
-															// retry)
+		// retry)
 		step.setSkipPolicy(new NeverSkipItemSkipPolicy()); // never skip
 		step.afterPropertiesSet();
 
@@ -297,14 +298,14 @@ class SkippableException extends RuntimeException {
 
 		JobRepository jobRepository = new ResourcelessJobRepository();
 		ChunkOrientedStep<String, String> step = new StepBuilder("step", jobRepository).<String, String>chunk(1)
-			.reader(reader)
-			.writer(writer)
-			.faultTolerant()
-			.retry(SkippableException.class)
-			.retryLimit(1)
-			.skip(SkippableException.class)
-			.skipLimit(1)
-			.build();
+				.reader(reader)
+				.writer(writer)
+				.faultTolerant()
+				.retry(SkippableException.class)
+				.retryLimit(1)
+				.skip(SkippableException.class)
+				.skipLimit(1)
+				.build();
 
 		JobInstance jobInstance = new JobInstance(1L, "job");
 		JobExecution jobExecution = new JobExecution(1L, jobInstance, new JobParameters());
@@ -319,4 +320,76 @@ class SkippableException extends RuntimeException {
 		assertEquals(1, stepExecution.getSkipCount());
 	}
 
+	/**
+	 * Test to verify that filterCount is accurate in concurrent mode. This test detects
+	 * race condition in StepContribution.incrementFilterCount().
+	 */
+	@RepeatedTest(5)
+	void testFilterCountAccuracyInConcurrentMode() throws Exception {
+		// given
+		int itemCount = 10000;
+		AtomicInteger readCounter = new AtomicInteger(0);
+
+		// Thread-safe reader
+		ItemReader<Integer> reader = () -> {
+			int current = readCounter.incrementAndGet();
+			return current <= itemCount ? current : null;
+		};
+
+		ItemProcessor<Integer, Integer> filteringProcessor = item -> null;
+
+		ItemWriter<Integer> writer = chunk -> {
+		};
+
+		JobRepository jobRepository = new ResourcelessJobRepository();
+		ChunkOrientedStep<Integer, Integer> step = new ChunkOrientedStep<>("step", 100, reader, writer, jobRepository);
+		step.setItemProcessor(filteringProcessor);
+		step.setTaskExecutor(new SimpleAsyncTaskExecutor()); // Enable concurrent mode
+		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
+		assertEquals(itemCount, stepExecution.getFilterCount(), "Race condition detected! Expected " + itemCount
+				+ " filtered items, but got " + stepExecution.getFilterCount());
+	}
+
+	@Test
+	void testFilterCountAccuracyInSequentialMode() throws Exception {
+		// given
+		int itemCount = 10000;
+		AtomicInteger readCounter = new AtomicInteger(0);
+
+		ItemReader<Integer> reader = () -> {
+			int current = readCounter.incrementAndGet();
+			return current <= itemCount ? current : null;
+		};
+
+		ItemProcessor<Integer, Integer> filteringProcessor = item -> null;
+		ItemWriter<Integer> writer = chunk -> {
+		};
+
+		JobRepository jobRepository = new ResourcelessJobRepository();
+		ChunkOrientedStep<Integer, Integer> step = new ChunkOrientedStep<>("step", 100, reader, writer, jobRepository);
+		step.setItemProcessor(filteringProcessor);
+		// No TaskExecutor - sequential mode
+		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
+		assertEquals(itemCount, stepExecution.getFilterCount(), "Sequential mode should have accurate filter count");
+	}
+
 }
+