Fix exhausted retry handling with non skippable exceptions

Before this commit, chunk scanning was unconditionally preformed
when the retry policy is exhausted and non skippable exceptions
are configured. This resulted in duplicate item writes which is not
intended.

This commit fixes that by checking the skip policy before scanning
the chunk on an exhausted retry policy.

Resolves #5091

Author: fmbenhassine

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

@@ -685,7 +685,8 @@ private void writeChunk(Chunk<O> chunk, StepContribution contribution) throws Ex
 			chunkWriteEvent.chunkWriteStatus = BatchMetrics.STATUS_FAILURE;
 			observation.lowCardinalityKeyValue(fullyQualifiedMetricName + ".status", BatchMetrics.STATUS_FAILURE);
 			observation.error(exception);
-			if (this.faultTolerant && exception instanceof RetryException retryException) {
+			if (this.faultTolerant && exception instanceof RetryException retryException
+					&& !this.skipPolicy.shouldSkip(retryException.getCause(), -1)) {
 				logger.info("Retry exhausted while attempting to write items, scanning the chunk", retryException);
 				ChunkScanEvent chunkScanEvent = new ChunkScanEvent(contribution.getStepExecution().getStepName(),
 						contribution.getStepExecution().getId());
@@ -696,6 +697,7 @@ private void writeChunk(Chunk<O> chunk, StepContribution contribution) throws Ex
 				logger.info("Chunk scan completed");
 			}
 			else {
+				logger.error("Retry exhausted after last attempt in recovery path, but exception is not skippable");
 				throw exception;
 			}
 		}

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

@@ -49,6 +49,9 @@ public LimitCheckingExceptionHierarchySkipPolicy(Set<Class<? extends Throwable>>
 
 	@Override
 	public boolean shouldSkip(Throwable t, long skipCount) throws SkipLimitExceededException {
+		if (skipCount < 0) {
+			return !isSkippable(t);
+		}
 		if (!isSkippable(t)) {
 			return false;
 		}

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

@@ -45,6 +45,7 @@
 import org.springframework.context.annotation.AnnotationConfigApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.core.retry.RetryException;
 import org.springframework.core.retry.RetryPolicy;
 import org.springframework.core.task.SimpleAsyncTaskExecutor;
 import org.springframework.dao.DataIntegrityViolationException;
@@ -240,6 +241,36 @@ void testSkipInReadInConcurrentMode() throws Exception {
 		Assertions.assertEquals(0, stepExecution.getWriteSkipCount());
 	}
 
+	// Issue https://github.com/spring-projects/spring-batch/issues/5091
+	@Test
+	void testExhaustedRetryWithNonSkippableException() throws Exception {
+		// given
+		ApplicationContext context = new AnnotationConfigApplicationContext(TestConfiguration.class,
+				ExhaustedRetryStepConfiguration.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.FAILED.getExitCode(), jobExecution.getExitStatus().getExitCode());
+		StepExecution stepExecution = jobExecution.getStepExecutions().iterator().next();
+		Throwable failureException = stepExecution.getFailureExceptions().iterator().next();
+		Assertions.assertInstanceOf(FatalStepExecutionException.class, failureException);
+		Assertions.assertInstanceOf(RetryException.class, failureException.getCause());
+		// the test has a single chunk (3 items with chunk size = 3) that is rolled back
+		// upon retry exhaustion
+		Assertions.assertEquals(3, stepExecution.getReadCount());
+		Assertions.assertEquals(0, stepExecution.getWriteCount());
+		Assertions.assertEquals(0, stepExecution.getCommitCount());
+		Assertions.assertEquals(1, stepExecution.getRollbackCount());
+		Assertions.assertEquals(0, stepExecution.getReadSkipCount());
+		// no scan is triggered as the exception is non skippable
+		Assertions.assertEquals(0, stepExecution.getWriteSkipCount());
+	}
+
 	@Configuration
 	static class StepConfiguration {
 
@@ -265,6 +296,27 @@ public Step step(JobRepository jobRepository, JdbcTransactionManager transaction
 
 	}
 
+	@Configuration
+	static class ExhaustedRetryStepConfiguration {
+
+		@Bean
+		public Step step(JobRepository jobRepository, JdbcTransactionManager transactionManager) {
+			List<String> items = List.of("one", "two", "three");
+			return new ChunkOrientedStepBuilder<String, String>(jobRepository, 3).reader(new ListItemReader<>(items))
+				.writer(chunk -> {
+					for (String item : chunk) {
+						if (item.equalsIgnoreCase("two")) {
+							throw new RuntimeException("Simulated write error on item two");
+						}
+					}
+				})
+				.transactionManager(transactionManager)
+				.faultTolerant()
+				.build();
+		}
+
+	}
+
 	@Configuration
 	static class ConcurrentStepConfiguration {