Skip to content

Prevent step execution leakage between flow job executions #5528

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
jjj-n wants to merge 2 commits into from
Closed

Prevent step execution leakage between flow job executions #5528

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2b0d0cd
Prevent step execution leakage between flow job executions
jjj-n Sep 8, 2026
f24bdc6
Merge upstream main to resolve PR #5528 conflicts
jjj-n Sep 18, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/JobFlowExecutor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@
*/
public class JobFlowExecutor implements FlowExecutor {

private static final ThreadLocal<StepExecution> stepExecutionHolder = new ThreadLocal<>();
private final ThreadLocal<StepExecution> stepExecutionHolder = new ThreadLocal<>();

private final JobExecution execution;

Expand Down
12 changes: 6 additions & 6 deletions ...-batch-core/src/main/java/org/springframework/batch/core/job/flow/support/SimpleFlow.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -166,6 +166,12 @@ public FlowExecution resume(String stateName, FlowExecutor executor) throws Flow
}
status = state.handle(executor);
stepExecution = executor.getStepExecution();

if (logger.isDebugEnabled()) {
logger.debug("Completed state=" + stateName + " with status=" + status);
}

state = nextState(stateName, status, stepExecution);
}
catch (FlowExecutionException e) {
executor.close(new FlowExecution(stateName, status));
Expand All @@ -176,12 +182,6 @@ public FlowExecution resume(String stateName, FlowExecutor executor) throws Flow
throw new FlowExecutionException(
String.format("Ended flow=%s at state=%s with exception", name, stateName), e);
}

if (logger.isDebugEnabled()) {
logger.debug("Completed state=" + stateName + " with status=" + status);
}

state = nextState(stateName, status, stepExecution);
}

FlowExecution result = new FlowExecution(stateName, status);
Expand Down
115 changes: 113 additions & 2 deletions ...batch-core/src/test/java/org/springframework/batch/core/job/flow/FlowJobFailureTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,8 +15,11 @@
*/
package org.springframework.batch.core.job.flow;

import org.springframework.batch.infrastructure.support.DatabaseType;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;

import java.util.ArrayList;
import java.util.List;
Expand All @@ -28,8 +31,10 @@
import org.springframework.batch.core.job.JobExecution;
import org.springframework.batch.core.job.JobInstance;
import org.springframework.batch.core.job.JobInterruptedException;
import org.springframework.batch.core.job.SimpleStepHandler;
import org.springframework.batch.core.job.parameters.JobParameters;
import org.springframework.batch.core.step.StepExecution;
import org.springframework.batch.core.step.Step;
import org.springframework.batch.core.job.UnexpectedJobExecutionException;
import org.springframework.batch.core.job.flow.support.SimpleFlow;
import org.springframework.batch.core.job.flow.support.StateTransition;
Expand All @@ -39,6 +44,7 @@
import org.springframework.batch.core.repository.support.JdbcJobRepositoryFactoryBean;
import org.springframework.batch.core.step.StepSupport;
import org.springframework.batch.infrastructure.item.ExecutionContext;
import org.springframework.batch.infrastructure.support.DatabaseType;
import org.springframework.jdbc.support.JdbcTransactionManager;
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase;
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
Expand All @@ -57,6 +63,8 @@ class FlowJobFailureTests {

private JobExecution execution;

private JobRepository jobRepository;

@BeforeEach
void init() throws Exception {
EmbeddedDatabase embeddedDatabase = new EmbeddedDatabaseBuilder()
Expand All @@ -68,7 +76,7 @@ void init() throws Exception {
factory.setDataSource(embeddedDatabase);
factory.setTransactionManager(new JdbcTransactionManager(embeddedDatabase));
factory.afterPropertiesSet();
JobRepository jobRepository = factory.getObject();
jobRepository = factory.getObject();
job.setJobRepository(jobRepository);
JobParameters jobParameters = new JobParameters();
JobInstance jobInstance = jobRepository.createJobInstance("job", jobParameters);
Expand All @@ -92,6 +100,109 @@ void testStepFailure() throws Exception {
assertEquals(BatchStatus.FAILED, execution.getStatus());
}

@Test
void testUnmatchedTransitionDoesNotAffectOtherJob() throws Exception {
job.setFlow(failingFlow());
job.afterPropertiesSet();
job.execute(execution);
assertEquals(BatchStatus.FAILED, execution.getStatus());
StepExecution failedStep = jobRepository.getLastStepExecution(execution.getJobInstance(), "step");
assertEquals(BatchStatus.FAILED, failedStep.getStatus());

FlowJob otherJob = new FlowJob("otherJob");
otherJob.setJobRepository(jobRepository);
SimpleFlow otherFlow = new SimpleFlow("otherFlow");
StepState otherStep = new StepState(new StepSupport("otherStep") {
@Override
public void execute(StepExecution stepExecution) {
stepExecution.setStatus(BatchStatus.COMPLETED);
stepExecution.setExitStatus(ExitStatus.COMPLETED);
jobRepository.update(stepExecution);
}
});
otherFlow.setStateTransitions(List.of(StateTransition.createEndStateTransition(otherStep)));
otherJob.setFlow(otherFlow);
otherJob.afterPropertiesSet();
JobParameters parameters = new JobParameters();
JobInstance instance = jobRepository.createJobInstance("otherJob", parameters);
JobExecution otherExecution = jobRepository.createJobExecution(instance, parameters, new ExecutionContext());
otherJob.execute(otherExecution);
assertEquals(BatchStatus.COMPLETED, otherExecution.getStatus());

StepExecution reloaded = jobRepository.getLastStepExecution(execution.getJobInstance(), "step");
assertEquals(BatchStatus.FAILED, reloaded.getStatus());
assertEquals(failedStep.getVersion(), reloaded.getVersion());
}

@Test
void testUnmatchedTransitionDoesNotPreventRestart() throws Exception {
job.setFlow(failingFlow());
job.afterPropertiesSet();
job.execute(execution);
assertEquals(BatchStatus.FAILED, execution.getStatus());

JobExecution restart = jobRepository.createJobExecution(execution.getJobInstance(), new JobParameters(),
new ExecutionContext());
job.execute(restart);

assertEquals(BatchStatus.FAILED, restart.getStatus());
assertEquals(1, restart.getStepExecutions().size());
assertEquals(2, jobRepository.getStepExecutionCount(execution.getJobInstance(), "step"));
assertEquals(BatchStatus.FAILED,
jobRepository.getLastStepExecution(execution.getJobInstance(), "step").getStatus());
}

@Test
void testUnmatchedTransitionClosesExecutor() throws Exception {
JobFlowExecutor executor = new JobFlowExecutor(jobRepository, new SimpleStepHandler(jobRepository), execution);
SimpleFlow flow = failingFlow();
try {
FlowExecutionException exception = assertThrows(FlowExecutionException.class, () -> flow.start(executor));
assertTrue(exception.getMessage().contains("Next state not found"));
assertNull(executor.getStepExecution());
}
finally {
executor.close(new FlowExecution("step", FlowExecutionStatus.FAILED));
}
}

@Test
void testExecutorsKeepSeparateStepExecutions() throws Exception {
JobExecution otherExecution = jobRepository.createJobExecution(execution.getJobInstance(), new JobParameters(),
new ExecutionContext());
JobFlowExecutor executor = new JobFlowExecutor(jobRepository, new SimpleStepHandler(jobRepository), execution);
JobFlowExecutor otherExecutor = new JobFlowExecutor(jobRepository, new SimpleStepHandler(jobRepository),
otherExecution);
try {
executor.executeStep(failingStep());
StepExecution stepExecution = executor.getStepExecution();
assertNull(otherExecutor.getStepExecution());
otherExecutor.close(new FlowExecution("other", FlowExecutionStatus.COMPLETED));
assertSame(stepExecution, executor.getStepExecution());
}
finally {
executor.close(new FlowExecution("step", FlowExecutionStatus.FAILED));
}
}

private SimpleFlow failingFlow() {
SimpleFlow flow = new SimpleFlow("job");
flow.setStateTransitions(
List.of(StateTransition.createEndStateTransition(new StepState(failingStep()), "COMPLETED")));
return flow;
}

private Step failingStep() {
return new StepSupport("step") {
@Override
public void execute(StepExecution stepExecution) {
stepExecution.setStatus(BatchStatus.FAILED);
stepExecution.setExitStatus(ExitStatus.FAILED);
jobRepository.update(stepExecution);
}
};
}

@Test
void testStepStatusUnknown() throws Exception {
SimpleFlow flow = new SimpleFlow("job");
Expand Down