Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

BATCH-2016: Added new flag to indicate if the step is being executed on

Browse Source 
a restart.  If it is, the transition will be followed.  If not, we'll
look to see if we should be looking for somewhere else to restart.
This commit is contained in:
Michael Minella
2014-03-10 10:47:56 -05:00
parent 2c62f20bbd
commit 4e5b4c12ba
6 changed files with 158 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
spring-batch-core/src/main/java/org/springframework/batch/core/job/SimpleStepHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2006-2013 the original author or authors.
* Copyright 2006-2014 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -54,15 +54,15 @@ public class SimpleStepHandler implements StepHandler, InitializingBean {
}
/**
* @param jobRepository
* @param jobRepository a {@link org.springframework.batch.core.repository.JobRepository}
*/
public SimpleStepHandler(JobRepository jobRepository) {
this(jobRepository, new ExecutionContext());
}
/**
* @param jobRepository
* @param executionContext
* @param jobRepository a {@link org.springframework.batch.core.repository.JobRepository}
* @param executionContext the {@link org.springframework.batch.item.ExecutionContext} for the current Step
*/
public SimpleStepHandler(JobRepository jobRepository, ExecutionContext executionContext) {
this.jobRepository = jobRepository;
@@ -125,6 +125,10 @@ public class SimpleStepHandler implements StepHandler, InitializingBean {
if (isRestart) {
currentStepExecution.setExecutionContext(lastStepExecution.getExecutionContext());
if(lastStepExecution.getExecutionContext().containsKey("batch.executed")) {
currentStepExecution.getExecutionContext().remove("batch.executed");
}
}
else {
currentStepExecution.setExecutionContext(new ExecutionContext(executionContext));
@@ -135,6 +139,7 @@ public class SimpleStepHandler implements StepHandler, InitializingBean {
logger.info("Executing step: [" + step.getName() + "]");
try {
step.execute(currentStepExecution);
currentStepExecution.getExecutionContext().put("batch.executed", true);
}
catch (JobInterruptedException e) {
// Ensure that the job gets the message that it is stopping
@@ -154,9 +159,6 @@ public class SimpleStepHandler implements StepHandler, InitializingBean {
}
}
else {
// currentStepExecution.setExitStatus(ExitStatus.NOOP);
}
return currentStepExecution;
}
@@ -165,7 +167,7 @@ public class SimpleStepHandler implements StepHandler, InitializingBean {
* Detect whether a step execution belongs to this job execution.
* @param jobExecution the current job execution
* @param stepExecution an existing step execution
* @return
* @return true if the {@link org.springframework.batch.core.StepExecution} is part of the {@link org.springframework.batch.core.JobExecution}
*/
private boolean stepExecutionPartOfExistingJobExecution(JobExecution jobExecution, StepExecution stepExecution) {
return stepExecution != null && stepExecution.getJobExecutionId() != null
@@ -176,8 +178,8 @@ public class SimpleStepHandler implements StepHandler, InitializingBean {
* Given a step and configuration, return true if the step should start,
* false if it should not, and throw an exception if the job should finish.
* @param lastStepExecution the last step execution
* @param jobInstance
* @param step
* @param jobInstance the current job instance
* @param step the step to execute
*
* @throws StartLimitExceededException if the start limit has been exceeded
* for this step
@@ -201,7 +203,7 @@ public class SimpleStepHandler implements StepHandler, InitializingBean {
+ "so it may be dangerous to proceed. Manual intervention is probably necessary.");
}
if ((stepStatus == BatchStatus.COMPLETED && step.isAllowStartIfComplete() == false)
if ((stepStatus == BatchStatus.COMPLETED && !step.isAllowStartIfComplete())
|| stepStatus == BatchStatus.ABANDONED) {
// step is complete, false should be returned, indicating that the
// step should not be started

45
spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/support/SimpleFlow.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2006-2013 the original author or authors.
* Copyright 2006-2014 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -15,19 +15,8 @@
*/
package org.springframework.batch.core.job.flow.support;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.core.JobExecutionException;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.job.flow.Flow;
@@ -38,6 +27,16 @@ import org.springframework.batch.core.job.flow.FlowExecutor;
import org.springframework.batch.core.job.flow.State;
import org.springframework.beans.factory.InitializingBean;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;
/**
* A {@link Flow} that branches conditionally depending on the exit status of
* the last {@link State}. The input parameters are the state transitions (in no
@@ -159,7 +158,7 @@ public class SimpleFlow implements Flow, InitializingBean {
catch (Exception e) {
executor.close(new FlowExecution(stateName, status));
throw new FlowExecutionException(String.format("Ended flow=%s at state=%s with exception", name,
stateName), e);
stateName), e);
}
logger.debug("Completed state="+stateName+" with status="+status);
@@ -180,8 +179,9 @@ public class SimpleFlow implements Flow, InitializingBean {
if(stepExecution != null) {
Boolean reRun = (Boolean) stepExecution.getExecutionContext().get("batch.restart");
Boolean executed = (Boolean) stepExecution.getExecutionContext().get("batch.executed");
if(reRun != null && reRun && status == FlowExecutionStatus.STOPPED && !state.getName().endsWith(stepExecution.getStepName())) {
if((executed == null || !executed) && reRun != null && reRun && status == FlowExecutionStatus.STOPPED && !state.getName().endsWith(stepExecution.getStepName())) {
continued = true;
}
}
@@ -191,7 +191,7 @@ public class SimpleFlow implements Flow, InitializingBean {
/**
* @return the next {@link Step} (or null if this is the end)
* @throws JobExecutionException
* @throws org.springframework.batch.core.job.flow.FlowExecutionException
*/
private State nextState(String stateName, FlowExecutionStatus status) throws FlowExecutionException {
@@ -199,7 +199,7 @@ public class SimpleFlow implements Flow, InitializingBean {
if (set == null) {
throw new FlowExecutionException(String.format("No transitions found in flow=%s for state=%s", getName(),
stateName));
stateName));
}
String next = null;
@@ -216,18 +216,15 @@ public class SimpleFlow implements Flow, InitializingBean {
}
if (next == null) {
throw new FlowExecutionException(String.format(
"Next state not found in flow=%s for state=%s with exit status=%s", getName(), stateName, status.getName()));
throw new FlowExecutionException(String.format("Next state not found in flow=%s for state=%s with exit status=%s", getName(), stateName, status.getName()));
}
if (!stateMap.containsKey(next)) {
throw new FlowExecutionException(String.format("Next state not specified in flow=%s for next=%s",
getName(), next));
getName(), next));
}
State state = stateMap.get(next);
return state;
return stateMap.get(next);
}
@@ -281,10 +278,10 @@ public class SimpleFlow implements Flow, InitializingBean {
if (!hasEndStep) {
throw new IllegalArgumentException(
"No end state was found. You must specify at least one transition with no next state.");
"No end state was found. You must specify at least one transition with no next state.");
}
startState = stateTransitions.get(0).getState();
}
}
}

37
spring-batch-core/src/test/java/org/springframework/batch/core/step/RestartInPriorStepTests.java
View File

@@ -1,10 +1,20 @@
/*
* Copyright 2014 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.core.step;
import static org.junit.Assert.assertEquals;
import java.util.Map;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.batch.core.BatchStatus;
@@ -26,6 +36,10 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import java.util.Map;
import static org.junit.Assert.assertEquals;
/**
* @author Michael Minella
*
@@ -46,26 +60,13 @@ public class RestartInPriorStepTests {
@Autowired
private Job job;
/**
* @throws java.lang.Exception
*/
@Before
public void setUp() throws Exception {
}
@Test
public void test() throws Exception {
//
// Run 1
//
JobExecution run1 = jobLauncher.run(job, new JobParameters());
assertEquals(BatchStatus.STOPPED, run1.getStatus());
assertEquals(2, run1.getStepExecutions().size());
//
// Run 2
//
JobExecution run2 = jobLauncher.run(job, new JobParameters());
assertEquals(BatchStatus.COMPLETED, run2.getStatus());

67
spring-batch-core/src/test/java/org/springframework/batch/core/step/RestartLoopTests.java Normal file
View File

@@ -0,0 +1,67 @@
/*
* Copyright 2014 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.core.step;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import static org.junit.Assert.assertEquals;
/**
* @author Michael Minella
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
public class RestartLoopTests {
@Autowired
private Job job;
@Autowired
private JobLauncher jobLauncher;
@Test
public void test() throws Exception {
// Run 1
JobExecution jobExecution1 = jobLauncher.run(job, new JobParameters());
assertEquals(BatchStatus.STOPPED, jobExecution1.getStatus());
// Run 2
JobExecution jobExecution2 = jobLauncher.run(job, new JobParameters());
assertEquals(BatchStatus.STOPPED, jobExecution2.getStatus());
}
public static class DefaultTasklet implements Tasklet {
@Override
public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
return RepeatStatus.FINISHED;
}
}
}

8
spring-batch-core/src/test/resources/org/springframework/batch/core/step/RestartInPriorStepTests-context.xml
View File

@@ -4,8 +4,6 @@
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/util
http://www.springframework.org/schema/util/spring-util.xsd
http://www.springframework.org/schema/batch
http://www.springframework.org/schema/batch/spring-batch-2.2.xsd">
@@ -54,7 +52,9 @@
</step>
<step id="step2">
<tasklet ref="decidingTasklet"/>
<!-- Second Run -->
<next on="ES3" to="step3"/>
<!-- First Run -->
<stop on="ES4" restart="step4"/>
</step>
<step id="step3" next="step4">
@@ -62,12 +62,12 @@
<chunk reader="step2ItemReader" writer="step2ItemWriter" commit-interval="10" />
</tasklet>
</step>
<step id="step4" next="completionDecider">
<step id="step4" next="completionDeciderStep">
<tasklet>
<chunk reader="step3ItemReader" writer="step3ItemWriter" commit-interval="10" />
</tasklet>
</step>
<decision decider="completionDecider" id="completionDecider">
<decision decider="completionDecider" id="completionDeciderStep">
<next on="CONTINUE" to="step3"/>
<end on="END"/>
</decision>

34
spring-batch-core/src/test/resources/org/springframework/batch/core/step/RestartLoopTests-context.xml Normal file
View File

@@ -0,0 +1,34 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/batch http://www.springframework.org/schema/batch/spring-batch-2.2.xsd">
<job id="pattern.replay" xmlns="http://www.springframework.org/schema/batch">
<step id="pattern.replay.1" next="pattern.replay.2">
<tasklet ref="tasklet1" allow-start-if-complete="true"/>
</step>
<step id="pattern.replay.2" next="pattern.replay.3">
<tasklet ref="tasklet2" allow-start-if-complete="true"/>
</step>
<step id="pattern.replay.3">
<tasklet ref="tasklet3" allow-start-if-complete="true"/>
<stop on="*" restart="pattern.replay.1" />
<fail on="FAILED" />
</step>
</job>
<bean id="tasklet1" class="org.springframework.batch.core.step.RestartLoopTests$DefaultTasklet"/>
<bean id="tasklet2" class="org.springframework.batch.core.step.RestartLoopTests$DefaultTasklet"/>
<bean id="tasklet3" class="org.springframework.batch.core.step.RestartLoopTests$DefaultTasklet"/>
<bean id="transactionManager" class="org.springframework.batch.support.transaction.ResourcelessTransactionManager"/>
<bean id="jobRepository" class="org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean">
<property name="transactionManager" ref="transactionManager"/>
</bean>
<bean id="jobLauncher" class="org.springframework.batch.core.launch.support.SimpleJobLauncher">
<property name="jobRepository" ref="jobRepository"/>
</bean>
</beans>