GH-5308: Serialize step execution updates with concurrent stop requests

[kyungrae]

Problem

When JobOperator.stop(jobExecution) is called while a step is running, the stopping thread persists the step execution's stopped state (jobRepository.update(stepExecution)) on its own thread — concurrently with the worker thread still committing chunks for the same BATCH_STEP_EXECUTION row. Both issue optimistic-locking updates:

UPDATE BATCH_STEP_EXECUTION SET ..., VERSION = ? WHERE STEP_EXECUTION_ID = ? AND VERSION = ?

so one matches 0 rows and fails with OptimisticLockingFailureException. It is timing- and vendor-sensitive: frequent on MySQL (REPEATABLE READ), occasional on PostgreSQL (READ COMMITTED), almost never on in-memory HSQLDB — which is why CI rarely catches it and GracefulShutdownFunctionalTests was @Disabled.

A contributing factor: SimpleJobRepository.update(StepExecution) re-read the row version via stepExecutionDao.synchronizeStatus(stepExecution) while the job was stopping. Under MySQL REPEATABLE READ that read returns the stale snapshot version, overwriting the correct in-memory version and guaranteeing the stopping thread's update loses.

Solution

Guard every update to a step execution's metadata with a per-execution lock, held across the surrounding transaction's commit, shared between the worker and the stopping thread:

• AbstractStep keeps one Semaphore per running step execution and exposes StoppableStep.callUnderLock(StepExecution, Runnable). The worker's start/chunk/final updates and the operator's stop update all run under it, so they serialize and neither observes a stale version.
• TaskletStep and ChunkOrientedStep take this shared lock around their chunk transactions.
• SimpleJobRepository.update(StepExecution) no longer re-reads the version while stopping — under the lock the shared in-memory execution already holds the current version (the stale re-read was the root failure on MySQL).
• The operator sets the job execution status to STOPPED directly instead of STOPPING (which update(JobExecution) upgraded to STOPPED anyway).
• Re-enables GracefulShutdownFunctionalTests (disabled under GracefulShutdownFunctionalTests.testStopJob fails intermittently due to a race condition #5308).

Validation

JobOperatorFunctionalTests and GracefulShutdownFunctionalTests, 100 runs per vendor (MySQL 8 / PostgreSQL 16 in Docker, HSQLDB in-memory), after the fix:

Test	Vendor	Baseline (no fix)	After this PR
GracefulShutdownFunctionalTests	HSQLDB	3/100	0/100
GracefulShutdownFunctionalTests	MySQL	27/100	0/100
GracefulShutdownFunctionalTests	PostgreSQL	3/100	0/100
JobOperatorFunctionalTests	HSQLDB	1/100	0/100
JobOperatorFunctionalTests	MySQL	22/100	0/100
JobOperatorFunctionalTests	PostgreSQL	8/100	0/100

(0 failures / 600 runs. Before-fix numbers: see #5442.) Plus a new AbstractStepTests unit test asserting t updates to one step execution; full spring-batch-core suite passes.

Resolves #5308

[nikhiln64]

The per execution semaphore moved onto AbstractStep and shared with the stopping thread is a clean way to keep a chunk commit from racing the stop update inside one JVM, and the new AbstractStepTests case shows callUnderLock serialising concurrent callers on the same execution. My concern is the change that rides along with it in SimpleJobRepository.update, where stepExecutionDao.synchronizeStatus(stepExecution) is removed from the isStopped or isStopping branch.

That call was doing more than status reconciliation. It reloads VERSION from the row and, when the in memory version is stale, resets stepExecution version to the database value, which is exactly what keeps the next optimistic locking update from throwing OptimisticLockingFailureException when two holders of the same StepExecution both write. The new lock replaces that protection only for writers in the same JVM, since callUnderLock runs the action unlocked when getStepExecutionLock returns null and the comment there names the case, not executing in this JVM. So for a remote partitioned or remote chunking step, the worker updates the row in its own JVM while the operator on the manager stops the same execution through the null lock path with synchronizeStatus now gone, and the reconciliation that used to absorb that race is no longer there.

The added test exercises callUnderLock within a single execute() on one JVM, so it stays green even if the remote path regresses. Could you add a case that stops a step whose execution is not registered in the stopping thread's map, standing in for the remote worker, and asserts the stop update still succeeds against a concurrently bumped version. If that path does still need synchronizeStatus, keeping it only for the null lock branch would restore the old behaviour without giving up the new in JVM lock.

Two smaller things. The semaphore is put into the map at the top of execute() and removed in its finally, so a stop arriving before execute() populates the map, or after it clears, takes the unlocked branch, and for a stop that lands in that window against a still running worker the serialisation is silently skipped. And setting the execution straight to STOPPED instead of STOPPING drops the STOPPING observation entirely, so anything polling for the intermediate state, or restart logic that distinguishes the two, now only ever sees STOPPED. Both may be intended but are worth a line in the description.

[kyungrae]

Thank you for the detailed review — it pushed me to study how PartitionStep and RemoteStep actually work, and I would like to share what I found before changing the PR.

1. On restoring stepExecutionDao.synchronizeStatus(stepExecution)

I don't think restoring it is the right fix, even when scoped to the unlocked branch.

The VERSION column exists to detect that another party has written the row, so that an update built on an outdated snapshot is rejected rather than applied. OptimisticLockingFailureException is that detection doing its job — it is the signal, not the defect.

synchronizeStatus re-reads the version and adopts it, so the update that follows is accepted even though the in-memory StepExecution was taken before the external write. The row is then written from that outdated snapshot. This does not resolve the concurrent modification — it only removes our ability to notice it.

If another party really has updated the row, what we should do depends on the state the step execution is in and on who performed that write, and I don't believe there is a generally safe rule for that. Defining such a rule is the genuinely hard part of this problem. Re-synchronizing the version silently replaces that decision with "write anyway", which I consider more dangerous than the exception it avoids.

So my preference is to eliminate the concurrent write rather than suppress its detection. (Section 2 below covers why I believe the remote step implementations do not have a concurrent writer on those rows in the first place.)

Some history on where that call comes from:

• e5fbc2a (Spring batch terminate in started status after sigterm #4023) made stop() mark the job and step executions STOPPED with an end time before updating the repository, which introduced a second writer on the step execution row.
• 644d7e6 (StepExecution Update in SimpleJobOperator.stop() Causes JobExecution.BatchStatus.UNKNOWN after graceful stop #5120, stop() does not prevent upcoming steps to be executed anymore #5114) then added version synchronization in SimpleJobRepository.update(StepExecution) to suppress the resulting OptimisticLockingFailureException.
• 692acba (Performance in SimpleJobRepository.update(StepExecution) since Spring Batch 6.x #5360) later replaced that full re-read with the cheaper stepExecutionDao.synchronizeStatus(...).

So the line this PR removes was introduced as a mitigation for the lock contention Once the contention itself is removed, removing it should not change behaviour.

2. On remote partitioning and remote chunking

You are right that a Semaphore cannot cross JVM boundaries, and this PR does not address cross-JVM concurrency. Having studied both remote step implementations, however, I do not believe the removal regresses them.

The worker's StepExecution row has a single writer after dispatch. For both PartitionStep and RemoteStep, that row is written only by the JVM that runs AbstractStep.execute() for it. The manager writes it only before dispatch (RemoteStep creates and updates it at lines 96-101; SimpleStepExecutionSplitter creates it for partitions) and afterwards only reads it (RemoteStep.pollRemoteStep, MessageChannelPartitionHandler.pollReplies).

SimpleJobOperator.stop() never reaches those rows. The loop at line 355 does iterate them — the splitter adds them to the same JobExecution — and they even pass the isRunning() check, because the manager's in-memory copies stay at their initial STARTING status. But stepLocator.getStep(stepExecution.getStepName()) at line 358 returns null: the execution name carries the partition suffix (workerStep:partition0, from SimpleStepExecutionSplitter), the manager's job only knows managerStep, and PartitionStep is not a StepLocator, so no delegation occurs. I verified this by replaying that lookup against a partitioned job:

stepExecution name=managerStep            getStep(name)=PartitionStep  -> stopped
stepExecution name=workerStep:partition2  getStep(name)=NULL  -> skipped
stepExecution name=workerStep:partition1  getStep(name)=NULL  -> skipped
stepExecution name=workerStep:partition0  getStep(name)=NULL  -> skipped

Stop reaches workers only through the shared database. stop() writes BATCH_JOB_EXECUTION.STATUS, and each worker observes it inside its own SimpleJobRepository.update(StepExecution), where jobExecutionDao.synchronizeStatus(jobExecution) refreshes the parent status and setTerminateOnly() is then applied. Note that this happens on the worker thread, not on the stopping thread.

Even if the step name did resolve, StoppableStep.stop() only flips an in-memory flag on the manager's own copy of the object, so it could not interrupt a worker running in another JVM in any case. Cross-JVM interruption is by design a database-driven, chunk-boundary mechanism.

3. On the incomplete critical section

You are right that the critical section in this PR is not complete: it does not cover the window before stepExecutionLocks.put(...) and after remove(...), and it cannot cover writers in other JVMs.

Rather than patching those gaps here, I think the underlying question is which approach to #5308 to take, and I would like the maintainer's guidance before investing further.

@fmbenhassine — there are two alternative directions on the table for #5308:

• This PR (GH-5308: Serialize step execution updates with concurrent stop requests #5448) keeps stop() writing the step execution and guards the two writers with a per-execution semaphore. As noted above, that boundary is inherently partial.
• GH-5308: Stop a job execution by signalling and awaiting its running step(s) #5442 removes the second writer instead. stop() no longer persists step executions: it marks the job STOPPING in a short transaction, signals each running step, and then waits — outside any transaction — for the executing thread to terminate and persist its own stopped state, with a configurable timeout. StoppableStep.stop() then only sets terminateOnly, and the thread executing the step owns the STOPPED / exit status / end time transition, making it the sole writer of that row.

I currently favour the second direction, and it also covers the concern behind #4023 better than a purely non-blocking stop does. The dangling STARTED status there comes from the process being terminated without a shutdown hook — for example a pod being killed in Kubernetes — before the stopped state was ever persisted. Because stop() in #5442 blocks until the executing thread has persisted the terminal state, the caller has a durable point to synchronize graceful shutdown against, instead of returning while the status may still be in flight.

Could you indicate which direction you would prefer? I am happy to complete either one.