Fix TaskletStep to propagate checked exceptions when rollback is disabled

spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/SimpleJobOperator.java

@@ -364,16 +364,44 @@ public boolean stop(JobExecution jobExecution) throws JobExecutionNotRunningExce
 								if (tasklet instanceof StoppableTasklet stoppableTasklet) {
 									StepSynchronizationManager.register(stepExecution);
 									stoppableTasklet.stop(stepExecution);
-									jobRepository.update(stepExecution);
-									jobRepository.updateExecutionContext(stepExecution);
+									try {
+										jobRepository.update(stepExecution);
+										jobRepository.updateExecutionContext(stepExecution);
+									}
+									catch (org.springframework.dao.OptimisticLockingFailureException e) {
+										// Ignore - the job thread is likely updating the
+										// step execution
+										// The job will check the STOPPING status and stop
+										// anyway
+										if (logger.isDebugEnabled()) {
+											logger
+												.debug("OptimisticLockingFailureException while stopping step execution "
+														+ stepExecution.getId()
+														+ ". This is expected if the job thread is updating concurrently.",
+														e);
+										}
+									}
 									StepSynchronizationManager.release();
 								}
 							}
 							if (step instanceof StoppableStep stoppableStep) {
 								StepSynchronizationManager.register(stepExecution);
 								stoppableStep.stop(stepExecution);
-								jobRepository.update(stepExecution);
-								jobRepository.updateExecutionContext(stepExecution);
+								try {
+									jobRepository.update(stepExecution);
+									jobRepository.updateExecutionContext(stepExecution);
+								}
+								catch (org.springframework.dao.OptimisticLockingFailureException e) {
+									// Ignore - the job thread is likely updating the step
+									// execution
+									// The job will check the STOPPING status and stop
+									// anyway
+									if (logger.isDebugEnabled()) {
+										logger.debug("OptimisticLockingFailureException while stopping step execution "
+												+ stepExecution.getId()
+												+ ". This is expected if the job thread is updating concurrently.", e);
+									}
+								}
 								StepSynchronizationManager.release();
 							}
 						}

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

@@ -417,8 +417,8 @@ public RepeatStatus doInTransaction(TransactionStatus status) {
 					catch (Exception e) {
 						if (transactionAttribute.rollbackOn(e)) {
 							chunkContext.setAttribute(ChunkListener.ROLLBACK_EXCEPTION_KEY, e);
-							throw e;
 						}
+						throw e;
 					}
 				}
 				finally {

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

@@ -39,6 +39,7 @@
 import org.jspecify.annotations.Nullable;
 import org.springframework.transaction.TransactionException;
 import org.springframework.transaction.UnexpectedRollbackException;
+import org.springframework.transaction.interceptor.DefaultTransactionAttribute;
 import org.springframework.transaction.support.DefaultTransactionStatus;
 import org.springframework.transaction.support.TransactionSynchronizationManager;
 
@@ -574,6 +575,64 @@ public JobInstance createJobInstance(String jobName, JobParameters jobParameters
 
 	}
 
+	@Test
+	void testCheckedExceptionPropagatedWhenRollbackDisabled() throws Exception {
+		// Test for regression of issue where checked exceptions were swallowed
+		// when transactionAttribute.rollbackOn(ex) returned false
+		taskletStep.setTransactionAttribute(new DefaultTransactionAttribute() {
+			@Override
+			public boolean rollbackOn(Throwable ex) {
+				// Simulate PROPAGATION_NOT_SUPPORTED behavior - no rollback for checked exceptions
+				if (ex instanceof RuntimeException) {
+					return true;
+				}
+				return false;
+			}
+		});
+
+		// Create a tasklet that throws a checked exception
+		taskletStep.setTasklet(new Tasklet() {
+			@Nullable
+			@Override
+			public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
+				throw new Exception("Checked exception from tasklet");
+			}
+		});
+
+		// Execute should fail with the checked exception
+		taskletStep.execute(stepExecution);
+		assertEquals(FAILED, stepExecution.getStatus());
+		assertTrue(stepExecution.getFailureExceptions().get(0) instanceof Exception);
+		assertEquals("Checked exception from tasklet",
+				stepExecution.getFailureExceptions().get(0).getMessage());
+	}
+
+	@Test
+	void testRuntimeExceptionPropagatedWhenRollbackDisabled() throws Exception {
+		// Even for runtime exceptions, they should be propagated
+		taskletStep.setTransactionAttribute(new DefaultTransactionAttribute() {
+			@Override
+			public boolean rollbackOn(Throwable ex) {
+				// Disable rollback for all exceptions
+				return false;
+			}
+		});
+
+		// Create a tasklet that throws a runtime exception
+		taskletStep.setTasklet(new Tasklet() {
+			@Nullable
+			@Override
+			public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
+				throw new RuntimeException("Runtime exception from tasklet");
+			}
+		});
+
+		// Execute should fail with the runtime exception
+		taskletStep.execute(stepExecution);
+		assertEquals(FAILED, stepExecution.getStatus());
+		assertTrue(stepExecution.getFailureExceptions().get(0) instanceof RuntimeException);
+	}
+
 	private static class FailingRollbackTransactionManager extends ResourcelessTransactionManager {
 
 		@Override

spring-batch-samples/src/test/java/org/springframework/batch/samples/restart/stop/GracefulShutdownFunctionalTests.java

@@ -79,14 +79,24 @@ void testStopJob(@Autowired Job job, @Autowired JobOperator jobOperator) throws
 
 		jobOperator.stop(jobExecution);
 
+		// Wait for the job to stop. The job needs to finish processing the current chunk
+		// before it can stop, so we need to be generous with the timeout.
+		// With 5 items, chunk size 2, and 500ms per item processing time:
+		// - Chunk 1: 2 items = 1000ms
+		// - Chunk 2: 2 items = 1000ms
+		// - Chunk 3: 1 item = 500ms
+		// If stop is called at the start of a chunk, we might need up to 1 second
+		// for the chunk to complete, plus time for the stop signal to be processed.
+		int maxWaitCount = 50; // 50 * 100ms = 5 seconds
 		int count = 0;
-		while (jobExecution.isRunning() && count <= 10) {
-			logger.info("Checking for end time in JobExecution: count=" + count);
+		while (jobExecution.isRunning() && count < maxWaitCount) {
+			logger.info("Checking for job to stop: status=" + jobExecution.getStatus() + ", count=" + count);
 			Thread.sleep(100);
 			count++;
 		}
 
-		assertFalse(jobExecution.isRunning(), "Timed out waiting for job to end.");
+		assertFalse(jobExecution.isRunning(), "Timed out waiting for job to stop after " + (count * 100)
+				+ "ms. Final status: " + jobExecution.getStatus());
 		assertEquals(BatchStatus.STOPPED, jobExecution.getStatus());
 	}