Skip to content

Commit 3d7086f

Browse files
kyungraekyungrae
committed
Stop a job execution by signalling and awaiting its running step(s)
JobOperator.stop() ran on the caller thread and itself persisted the running StepExecution (jobRepository.update). The thread executing the step persists the same BATCH_STEP_EXECUTION row, so the two raced on the optimistic lock: whichever lost either propagated an OptimisticLockingFailureException out of stop() or was driven into an UNKNOWN state (and could no longer be restarted). Make the thread executing the step the sole writer of its StepExecution: - stop() no longer persists step executions. It marks the job STOPPING in a short transaction, signals each running step, then waits - outside any transaction - for the step to terminate and persist its own stopped state, with a configurable timeout (JobExecutionStopException on expiry). This also gives graceful shutdown the durability it needs: the caller blocks until the stopped state is persisted. - StoppableStep gains subscribeToTermination(StepExecution); AbstractStep completes the returned future once the execution terminates. - StoppableStep.stop() default now only sets terminateOnly; the worker owns the STOPPED / exit status / end time transition. - Revert the stop-time StepExecution version re-sync in SimpleJobRepository.update, which only narrowed the race window. - stop() is no longer wrapped in the operator's transaction. Resolves #5308 Signed-off-by: Kyungrae Kim <rlarudfo93@gmail.com>
1 parent 91a81b6 commit 3d7086f

7 files changed

Lines changed: 204 additions & 54 deletions

File tree

‎spring-batch-core/src/main/java/org/springframework/batch/core/launch/JobExecutionStopException.java‎

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,36 @@
1+
/*
2+
* Copyright 2026-present the original author or authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package org.springframework.batch.core.launch;
17+
18+
/**
19+
* Exception thrown when a request to stop a job execution does not complete within the
20+
* configured timeout, i.e. its running step(s) did not reach a terminal state in time.
21+
*
22+
* @author Kyungrae Kim
23+
* @since 6.0
24+
*/
25+
public class JobExecutionStopException extends RuntimeException {
26+
27+
/**
28+
* Create a {@link JobExecutionStopException} with a message and a cause.
29+
* @param msg the message to signal the cause of failure
30+
* @param cause the underlying cause
31+
*/
32+
public JobExecutionStopException(String msg, Throwable cause) {
33+
super(msg, cause);
34+
}
35+
36+
}

‎spring-batch-core/src/main/java/org/springframework/batch/core/launch/JobOperator.java‎

Lines changed: 8 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -187,14 +187,17 @@ Long startNextInstance(String jobName) throws NoSuchJobException, JobParametersN
187187
boolean stop(long executionId) throws NoSuchJobExecutionException, JobExecutionNotRunningException;
188188

189189
/**
190-
* Send a stop signal to the supplied {@link JobExecution}. The signal is successfully
191-
* sent if this method returns true, but that doesn't mean that the job has stopped.
192-
* The only way to be sure of that is to poll the job execution status.
190+
* Stop the supplied {@link JobExecution} and wait for its running step(s) to stop.
191+
* The job execution is marked as
192+
* {@link org.springframework.batch.core.BatchStatus#STOPPING STOPPING}, the running
193+
* step(s) are signalled to stop, and this method blocks until they have terminated
194+
* and persisted their stopped state, up to a configurable timeout.
193195
* @param jobExecution the running {@link JobExecution}
194-
* @return true if the message was successfully sent (does not guarantee that the job
195-
* has stopped)
196+
* @return {@code true} once the job execution has stopped
196197
* @throws JobExecutionNotRunningException if the supplied {@link JobExecution} is not
197198
* running (so cannot be stopped)
199+
* @throws JobExecutionStopException if the running step(s) do not stop within the
200+
* configured timeout
198201
*/
199202
boolean stop(JobExecution jobExecution) throws JobExecutionNotRunningException;
200203

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

Lines changed: 17 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616
package org.springframework.batch.core.launch.support;
1717

1818
import java.lang.reflect.Method;
19+
import java.time.Duration;
1920

2021
import io.micrometer.observation.ObservationRegistry;
2122
import org.apache.commons.logging.Log;
@@ -77,6 +78,8 @@ public class JobOperatorFactoryBean implements FactoryBean<JobOperator>, Applica
7778

7879
private JobParametersConverter jobParametersConverter = new DefaultJobParametersConverter();
7980

81+
private Duration stopTimeout = Duration.ofSeconds(30);
82+
8083
@SuppressWarnings("NullAway.Init")
8184
private TaskExecutor taskExecutor;
8285

@@ -175,6 +178,18 @@ public void setTransactionManager(PlatformTransactionManager transactionManager)
175178
this.transactionManager = transactionManager;
176179
}
177180

181+
/**
182+
* Set how long {@code stop(JobExecution)} waits for the running step(s) to actually
183+
* stop before failing with a {@code JobExecutionStopException}. Defaults to 30
184+
* seconds.
185+
* @param stopTimeout the maximum time to wait for a job to stop
186+
* @since 6.0
187+
*/
188+
public void setStopTimeout(Duration stopTimeout) {
189+
Assert.notNull(stopTimeout, "stopTimeout must not be null");
190+
this.stopTimeout = stopTimeout;
191+
}
192+
178193
/**
179194
* Set the transaction attributes source to use in the created proxy.
180195
* @param transactionAttributeSource the transaction attributes source to use in the
@@ -217,6 +232,8 @@ private TaskExecutorJobOperator getTarget() throws Exception {
217232
taskExecutorJobOperator.setObservationRegistry(this.observationRegistry);
218233
}
219234
taskExecutorJobOperator.setJobParametersConverter(this.jobParametersConverter);
235+
taskExecutorJobOperator.setTransactionManager(this.transactionManager);
236+
taskExecutorJobOperator.setStopTimeout(this.stopTimeout);
220237
taskExecutorJobOperator.afterPropertiesSet();
221238
return taskExecutorJobOperator;
222239
}
@@ -226,10 +243,8 @@ private static class DefaultJobOperatorTransactionAttributeSource extends Method
226243
public DefaultJobOperatorTransactionAttributeSource() {
227244
DefaultTransactionAttribute transactionAttribute = new DefaultTransactionAttribute();
228245
try {
229-
Method stopMethod = TaskExecutorJobOperator.class.getMethod("stop", JobExecution.class);
230246
Method abandonMethod = TaskExecutorJobOperator.class.getMethod("abandon", JobExecution.class);
231247
Method recoverMethod = TaskExecutorJobOperator.class.getMethod("recover", JobExecution.class);
232-
addTransactionalMethod(stopMethod, transactionAttribute);
233248
addTransactionalMethod(abandonMethod, transactionAttribute);
234249
addTransactionalMethod(recoverMethod, transactionAttribute);
235250
}

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

Lines changed: 97 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515
*/
1616
package org.springframework.batch.core.launch.support;
1717

18+
import java.time.Duration;
1819
import java.time.LocalDateTime;
1920
import java.util.ArrayList;
2021
import java.util.LinkedHashMap;
@@ -24,13 +25,16 @@
2425
import java.util.Properties;
2526
import java.util.Set;
2627
import java.util.TreeSet;
28+
import java.util.concurrent.CompletableFuture;
29+
import java.util.concurrent.ExecutionException;
30+
import java.util.concurrent.TimeUnit;
31+
import java.util.concurrent.TimeoutException;
2732

2833
import org.apache.commons.logging.Log;
2934
import org.apache.commons.logging.LogFactory;
3035
import org.jspecify.annotations.NullUnmarked;
3136

3237
import org.springframework.batch.core.BatchStatus;
33-
import org.springframework.batch.core.ExitStatus;
3438
import org.springframework.batch.core.job.Job;
3539
import org.springframework.batch.core.job.JobExecution;
3640
import org.springframework.batch.core.job.JobInstance;
@@ -54,10 +58,13 @@
5458
import org.springframework.batch.core.launch.JobInstanceAlreadyCompleteException;
5559
import org.springframework.batch.core.repository.JobRepository;
5660
import org.springframework.batch.core.launch.JobRestartException;
61+
import org.springframework.batch.core.launch.JobExecutionStopException;
62+
import org.springframework.batch.infrastructure.support.transaction.ResourcelessTransactionManager;
63+
import org.springframework.transaction.PlatformTransactionManager;
64+
import org.springframework.transaction.support.TransactionTemplate;
5765
import org.springframework.batch.core.scope.context.StepSynchronizationManager;
5866
import org.springframework.batch.core.step.StepLocator;
5967
import org.springframework.batch.core.step.tasklet.StoppableTasklet;
60-
import org.springframework.batch.core.step.tasklet.Tasklet;
6168
import org.springframework.batch.core.step.tasklet.TaskletStep;
6269
import org.springframework.batch.infrastructure.support.PropertiesConverter;
6370
import org.springframework.beans.factory.InitializingBean;
@@ -102,6 +109,34 @@ public class SimpleJobOperator extends TaskExecutorJobLauncher implements JobOpe
102109

103110
private final Log logger = LogFactory.getLog(getClass());
104111

112+
private PlatformTransactionManager transactionManager = new ResourcelessTransactionManager();
113+
114+
private Duration stopTimeout = Duration.ofSeconds(30);
115+
116+
/**
117+
* Set the transaction manager used to persist the stopping state of a job execution
118+
* atomically before waiting for it to stop. Defaults to a
119+
* {@link ResourcelessTransactionManager}.
120+
* @param transactionManager the transaction manager to use
121+
* @since 6.0
122+
*/
123+
public void setTransactionManager(PlatformTransactionManager transactionManager) {
124+
Assert.notNull(transactionManager, "transactionManager must not be null");
125+
this.transactionManager = transactionManager;
126+
}
127+
128+
/**
129+
* Set how long {@link #stop(JobExecution)} waits for the running step(s) to actually
130+
* stop before giving up with a {@link JobExecutionStopException}. Defaults to 30
131+
* seconds (aligned with the Spring graceful shutdown convention).
132+
* @param stopTimeout the maximum time to wait for the job to stop
133+
* @since 6.0
134+
*/
135+
public void setStopTimeout(Duration stopTimeout) {
136+
Assert.notNull(stopTimeout, "stopTimeout must not be null");
137+
this.stopTimeout = stopTimeout;
138+
}
139+
105140
/**
106141
* Check mandatory properties.
107142
*
@@ -342,51 +377,78 @@ public boolean stop(JobExecution jobExecution) throws JobExecutionNotRunningExce
342377
if (logger.isInfoEnabled()) {
343378
logger.info("Stopping job execution: " + jobExecution);
344379
}
345-
jobExecution.setStatus(BatchStatus.STOPPING); // will be upgraded to STOPPED in
346-
// JobRepository.update
347-
jobExecution.setExitStatus(ExitStatus.STOPPED);
348-
jobExecution.setEndTime(LocalDateTime.now());
349-
jobRepository.update(jobExecution);
350-
jobRepository.updateExecutionContext(jobExecution);
351380

381+
List<CompletableFuture<StepExecution>> terminations = new ArrayList<>();
382+
List<Runnable> stopSignals = new ArrayList<>();
352383
Job job = jobRegistry.getJob(jobExecution.getJobInstance().getJobName());
353-
if (job != null) {
354-
if (job instanceof StepLocator stepLocator) {
355-
// can only process as StepLocator is the only way to get the step object
356-
// get the current stepExecution
357-
for (StepExecution stepExecution : jobExecution.getStepExecutions()) {
358-
if (stepExecution.getStatus().isRunning()) {
359-
// have the step execution that's running -> need to 'stop' it
360-
Step step = stepLocator.getStep(stepExecution.getStepName());
361-
if (step != null) {
362-
if (step instanceof TaskletStep taskletStep) {
363-
Tasklet tasklet = taskletStep.getTasklet();
364-
if (tasklet instanceof StoppableTasklet stoppableTasklet) {
365-
StepSynchronizationManager.register(stepExecution);
366-
stoppableTasklet.stop(stepExecution);
367-
jobRepository.update(stepExecution);
368-
jobRepository.updateExecutionContext(stepExecution);
369-
StepSynchronizationManager.release();
370-
}
371-
}
372-
if (step instanceof StoppableStep stoppableStep) {
373-
StepSynchronizationManager.register(stepExecution);
374-
stoppableStep.stop(stepExecution);
375-
jobRepository.update(stepExecution);
376-
jobRepository.updateExecutionContext(stepExecution);
377-
StepSynchronizationManager.release();
378-
}
384+
if (job instanceof StepLocator stepLocator) {
385+
for (StepExecution stepExecution : jobExecution.getStepExecutions()) {
386+
if (!stepExecution.getStatus().isRunning()) {
387+
continue;
388+
}
389+
Step step = stepLocator.getStep(stepExecution.getStepName());
390+
if (step instanceof StoppableStep stoppableStep) {
391+
terminations.add(stoppableStep.subscribeToTermination(stepExecution));
392+
stopSignals.add(() -> {
393+
StepSynchronizationManager.register(stepExecution);
394+
try {
395+
stoppableStep.stop(stepExecution);
379396
}
380-
}
397+
finally {
398+
StepSynchronizationManager.release();
399+
}
400+
});
401+
}
402+
if (step instanceof TaskletStep taskletStep
403+
&& taskletStep.getTasklet() instanceof StoppableTasklet stoppableTasklet) {
404+
stopSignals.add(() -> {
405+
StepSynchronizationManager.register(stepExecution);
406+
try {
407+
stoppableTasklet.stop(stepExecution);
408+
}
409+
finally {
410+
StepSynchronizationManager.release();
411+
}
412+
});
381413
}
382414
}
383415
// TODO what if the job is not a StepLocator? ie a job with no steps?
384416
// FIXME Job should provide a stop() method
385417

386418
}
419+
420+
// Persist STOPPING in its own short transaction that commits at once,
421+
// so it is durable and holds no lock during the wait.
422+
new TransactionTemplate(this.transactionManager).executeWithoutResult(transactionStatus -> {
423+
jobExecution.setStatus(BatchStatus.STOPPING);
424+
jobRepository.update(jobExecution);
425+
});
426+
427+
stopSignals.forEach(Runnable::run);
428+
awaitStop(jobExecution, terminations);
387429
return true;
388430
}
389431

432+
private void awaitStop(JobExecution jobExecution, List<CompletableFuture<StepExecution>> terminations) {
433+
try {
434+
CompletableFuture.allOf(terminations.toArray(new CompletableFuture[0]))
435+
.get(this.stopTimeout.toMillis(), TimeUnit.MILLISECONDS);
436+
}
437+
catch (TimeoutException e) {
438+
throw new JobExecutionStopException("Timed out after " + this.stopTimeout
439+
+ " while waiting for job execution " + jobExecution.getId() + " to stop", e);
440+
}
441+
catch (ExecutionException e) {
442+
throw new JobExecutionStopException(
443+
"Failure while waiting for job execution " + jobExecution.getId() + " to stop", e.getCause());
444+
}
445+
catch (InterruptedException e) {
446+
Thread.currentThread().interrupt();
447+
throw new JobExecutionStopException(
448+
"Interrupted while waiting for job execution " + jobExecution.getId() + " to stop", e);
449+
}
450+
}
451+
390452
@Override
391453
@Deprecated(since = "6.0", forRemoval = true)
392454
public JobExecution abandon(long jobExecutionId)

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

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -164,7 +164,6 @@ public void update(StepExecution stepExecution) {
164164
this.jobExecutionDao.synchronizeStatus(jobExecution);
165165

166166
if (jobExecution.isStopped() || jobExecution.isStopping()) {
167-
this.stepExecutionDao.synchronizeStatus(stepExecution);
168167
stepExecution.setTerminateOnly();
169168
}
170169

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

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,10 @@
1818
import java.time.Duration;
1919
import java.time.LocalDateTime;
2020
import java.util.List;
21+
import java.util.Map;
22+
import java.util.Set;
23+
import java.util.concurrent.CompletableFuture;
24+
import java.util.concurrent.ConcurrentHashMap;
2125
import java.util.stream.Collectors;
2226

2327
import io.micrometer.observation.Observation;
@@ -74,6 +78,10 @@ public abstract class AbstractStep implements StoppableStep, InitializingBean, B
7478

7579
private final CompositeStepExecutionListener stepExecutionListener = new CompositeStepExecutionListener();
7680

81+
private final Set<Long> executingStepExecutions = ConcurrentHashMap.newKeySet();
82+
83+
private final Map<Long, CompletableFuture<StepExecution>> terminationSignals = new ConcurrentHashMap<>();
84+
7785
private JobRepository jobRepository;
7886

7987
protected ObservationRegistry observationRegistry;
@@ -212,6 +220,7 @@ public final void execute(StepExecution stepExecution)
212220
throws JobInterruptedException, UnexpectedJobExecutionException {
213221

214222
Assert.notNull(stepExecution, "stepExecution must not be null");
223+
this.executingStepExecutions.add(stepExecution.getId());
215224
stepExecution.getExecutionContext().put(SpringBatchVersion.BATCH_VERSION_KEY, SpringBatchVersion.getVersion());
216225

217226
if (logger.isDebugEnabled()) {
@@ -355,7 +364,28 @@ public final void execute(StepExecution stepExecution)
355364
if (logger.isDebugEnabled()) {
356365
logger.debug("Step execution complete: " + stepExecution.getSummary());
357366
}
367+
368+
// Notify any caller of stop(StepExecution) that this execution has terminated
369+
// and its final metadata has been saved.
370+
this.executingStepExecutions.remove(stepExecution.getId());
371+
CompletableFuture<StepExecution> terminationSignal = this.terminationSignals.remove(stepExecution.getId());
372+
if (terminationSignal != null) {
373+
terminationSignal.complete(stepExecution);
374+
}
375+
}
376+
}
377+
378+
@Override
379+
public CompletableFuture<StepExecution> subscribeToTermination(StepExecution stepExecution) {
380+
Long stepExecutionId = stepExecution.getId();
381+
CompletableFuture<StepExecution> terminationSignal = this.terminationSignals
382+
.computeIfAbsent(stepExecutionId, key -> new CompletableFuture<>());
383+
// If the execution is not running in this JVM, it has already terminated.
384+
if (!this.executingStepExecutions.contains(stepExecution.getId())) {
385+
terminationSignal.complete(stepExecution);
386+
this.terminationSignals.remove(stepExecutionId, terminationSignal);
358387
}
388+
return terminationSignal;
359389
}
360390

361391
private void stopObservation(StepExecution stepExecution, Observation observation) {

0 commit comments

Comments
 (0)