Fix optimistic locking and add interruption check in ChunkOrientedStep

When JobOperator.stop() is called, the StepExecution version could become
out of sync between the stopping thread and the executing step thread,
causing OptimisticLockingFailureException.

This commit synchronizes the StepExecution version by fetching the latest
state from the batch metadata  before update. If the recent StepExecution is in
STOPPED status, it upgrades the current StepExecution's status and version
to match the database state, preventing version conflicts.

Additionally, added JobRepository.update() call in ChunkOrientedStep after
each chunk to match TaskletStep behavior. This ensures both step
implementations properly check for job interruption signals by synchronizing
with the latest StepExecution status from the database.

Move stepExecution.update() outside chunk transaction to prevent
version mismatch when transaction rolls back.

Problem:
- Chunk transaction rollback reverts DB version but leaves
  in-memory version incremented
- Next update() causes OptimisticLockingFailureException

Solution:
- Call getJobRepository().update(stepExecution) after transaction
  completes, outside transactionTemplate.executeWithoutResult()
- On success: update() is called with synchronized versions
- On failure: update() is skipped, preventing version conflict

Fixes #5120
Fixes #5114

Signed-off-by: kill9-no-mercy <kill9.no.mercy@gmail.com>

Author: KILL9-NO-MERCY

spring-batch-core/src/main/java/org/springframework/batch/core/job/JobExecution.java

@@ -213,6 +213,14 @@ public boolean isStopping() {
 		return status == BatchStatus.STOPPING;
 	}
 
+	/**
+	 * Test if this {@link StepExecution} indicates that it has been stopped.
+	 * @return {@code true} if the status is {@link BatchStatus#STOPPED}.
+	 */
+	public boolean isStopped() {
+		return status == BatchStatus.STOPPED;
+	}
+
 	/**
 	 * Sets the {@link ExecutionContext} for this execution.
 	 * @param executionContext The context.

spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/ResourcelessJobRepository.java

@@ -328,7 +328,7 @@ public long getStepExecutionCount(JobInstance jobInstance, String stepName) {
 	@Override
 	public void update(StepExecution stepExecution) {
 		stepExecution.setLastUpdated(LocalDateTime.now());
-		if (this.jobExecution != null && this.jobExecution.isStopping()) {
+		if (this.jobExecution != null && this.jobExecution.isStopped()) {
 			stepExecution.setTerminateOnly();
 		}
 	}

spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/SimpleJobRepository.java

@@ -158,8 +158,20 @@ public void update(StepExecution stepExecution) {
 		Assert.notNull(stepExecution.getId(), "StepExecution must already be saved (have an id assigned)");
 
 		stepExecution.setLastUpdated(LocalDateTime.now());
+
+		StepExecution latestStepExecution = getStepExecution(stepExecution.getId());
+		Assert.state(latestStepExecution != null,
+				"StepExecution with id " + stepExecution.getId() + "not found. Batch metadata state may be corrupted.");
+
+		if (latestStepExecution.getJobExecution().isStopped()) {
+			Integer version = latestStepExecution.getVersion();
+			if (version != null) {
+				stepExecution.setVersion(version);
+			}
+			stepExecution.setTerminateOnly();
+		}
+
 		stepExecutionDao.updateStepExecution(stepExecution);
-		checkForInterruption(stepExecution);
 	}
 
 	private void validateStepExecution(StepExecution stepExecution) {
@@ -180,22 +192,6 @@ public void updateExecutionContext(JobExecution jobExecution) {
 		ecDao.updateExecutionContext(jobExecution);
 	}
 
-	/**
-	 * Check to determine whether or not the JobExecution that is the parent of the
-	 * provided StepExecution has been interrupted. If, after synchronizing the status
-	 * with the database, the status has been updated to STOPPING, then the job has been
-	 * interrupted.
-	 * @param stepExecution the step execution
-	 */
-	private void checkForInterruption(StepExecution stepExecution) {
-		JobExecution jobExecution = stepExecution.getJobExecution();
-		jobExecutionDao.synchronizeStatus(jobExecution);
-		if (jobExecution.isStopping()) {
-			logger.info("Parent JobExecution is stopped, so passing message on to StepExecution");
-			stepExecution.setTerminateOnly();
-		}
-	}
-
 	@Override
 	public void deleteStepExecution(StepExecution stepExecution) {
 		this.ecDao.deleteExecutionContext(stepExecution);

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

@@ -373,6 +373,8 @@ protected void doExecute(StepExecution stepExecution) throws Exception {
 						? BatchMetrics.STATUS_ROLLED_BACK : BatchMetrics.STATUS_COMMITTED;
 				chunkTransactionEvent.commit();
 			});
+
+			getJobRepository().update(stepExecution);
 		}
 	}