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

BATCH-1730: allow flow to be repeated in a PartitionStep

Browse Source
This commit is contained in:
Dave Syer
2011-05-05 08:58:48 +01:00
parent 3e95630fa6
commit 3f38056ff3
6 changed files with 269 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/SimpleStepExecutionSplitter.java
View File

@@ -247,11 +247,11 @@ public class SimpleStepExecutionSplitter implements StepExecutionSplitter, Initi
stepExecution.setExecutionContext(context);
}
return shouldStart(allowStartIfComplete, lastStepExecution) || isRestart;
return shouldStart(allowStartIfComplete, stepExecution, lastStepExecution) || isRestart;
}
private boolean shouldStart(boolean allowStartIfComplete, StepExecution lastStepExecution)
private boolean shouldStart(boolean allowStartIfComplete, StepExecution stepExecution, StepExecution lastStepExecution)
throws JobExecutionException {
if (lastStepExecution == null) {
@@ -268,6 +268,10 @@ public class SimpleStepExecutionSplitter implements StepExecutionSplitter, Initi
if (stepStatus == BatchStatus.COMPLETED) {
if (!allowStartIfComplete) {
if (isSameJobExecution(stepExecution, lastStepExecution)) {
// it's always OK to start again in the same JobExecution
return true;
}
// step is complete, false should be returned, indicating that
// the step should not be started
return false;
@@ -295,4 +299,11 @@ public class SimpleStepExecutionSplitter implements StepExecutionSplitter, Initi
}
private boolean isSameJobExecution(StepExecution stepExecution, StepExecution lastStepExecution) {
if (stepExecution.getJobExecutionId()==null) {
return lastStepExecution.getJobExecutionId()==null;
}
return stepExecution.getJobExecutionId().equals(lastStepExecution.getJobExecutionId());
}
}

59
spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/FlowStepParserTests.java
View File

@@ -29,6 +29,8 @@ import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.job.flow.FlowExecutionStatus;
import org.springframework.batch.core.job.flow.JobExecutionDecider;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean;
import org.springframework.beans.factory.annotation.Autowired;
@@ -36,15 +38,14 @@ import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Dave Syer
*
*
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
public class FlowStepParserTests {
@Autowired
@Qualifier("job1")
private Job job1;
@@ -53,12 +54,20 @@ public class FlowStepParserTests {
@Qualifier("job2")
private Job job2;
@Autowired
@Qualifier("job3")
private Job job3;
@Autowired
@Qualifier("job4")
private Job job4;
@Autowired
private JobRepository jobRepository;
@Autowired
private MapJobRepositoryFactoryBean mapJobRepositoryFactoryBean;
@Before
public void setUp() {
mapJobRepositoryFactoryBean.clear();
@@ -86,6 +95,36 @@ public class FlowStepParserTests {
assertEquals("[job2.s1, job2.flow, s2, s3, job2.s4]", stepNames.toString());
}
@Test
public void testRepeatedFlow() throws Exception {
assertNotNull(job3);
JobExecution jobExecution = jobRepository.createJobExecution(job3.getName(), new JobParameters());
job3.execute(jobExecution);
assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus());
List<String> stepNames = getStepNames(jobExecution);
assertEquals(6, stepNames.size());
assertEquals("[job3.flow, s2, s3, job3.flow, s2, s3]", stepNames.toString());
}
@Test
// TODO: BATCH-1745
public void testRestartedFlow() throws Exception {
assertNotNull(job4);
JobExecution jobExecution = jobRepository.createJobExecution(job4.getName(), new JobParameters());
job4.execute(jobExecution);
assertEquals(BatchStatus.FAILED, jobExecution.getStatus());
List<String> stepNames = getStepNames(jobExecution);
assertEquals(3, stepNames.size());
assertEquals("[job4.flow, s2, s3]", stepNames.toString());
jobExecution = jobRepository.createJobExecution(job4.getName(), new JobParameters());
job4.execute(jobExecution);
assertEquals(BatchStatus.FAILED, jobExecution.getStatus());
stepNames = getStepNames(jobExecution);
assertEquals(1, stepNames.size());
// The flow executes again, but all the steps were already complete
assertEquals("[job4.flow]", stepNames.toString());
}
private List<String> getStepNames(JobExecution jobExecution) {
List<String> list = new ArrayList<String>();
for (StepExecution stepExecution : jobExecution.getStepExecutions()) {
@@ -94,4 +133,16 @@ public class FlowStepParserTests {
return list;
}
public static class Decider implements JobExecutionDecider {
int count = 0;
public FlowExecutionStatus decide(JobExecution jobExecution, StepExecution stepExecution) {
if (count++ < 2) {
return new FlowExecutionStatus("OK");
}
return new FlowExecutionStatus("END");
}
}
}

107
spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/PartitionStepWithFlowParserTests.java Normal file
View File

@@ -0,0 +1,107 @@
/*
* Copyright 2006-2007 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.configuration.xml;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import org.junit.Before;
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.JobParametersBuilder;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.job.flow.FlowExecutionStatus;
import org.springframework.batch.core.job.flow.JobExecutionDecider;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Dave Syer
* @author Josh Long
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
public class PartitionStepWithFlowParserTests {
@Autowired
@Qualifier("job1")
private Job job1;
@Autowired
@Qualifier("nameStoringTasklet")
private NameStoringTasklet nameStoringTasklet;
@Autowired
private JobRepository jobRepository;
@Autowired
private MapJobRepositoryFactoryBean mapJobRepositoryFactoryBean;
private List<String> savedStepNames = new ArrayList<String>();
@Before
public void setUp() {
nameStoringTasklet.setStepNamesList(savedStepNames);
mapJobRepositoryFactoryBean.clear();
}
@Test
public void testRepeatedFlowStep() throws Exception {
assertNotNull(job1);
JobExecution jobExecution = jobRepository.createJobExecution(job1.getName(), new JobParametersBuilder()
.addLong("gridSize", 1L).toJobParameters());
job1.execute(jobExecution);
assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus());
Collections.sort(savedStepNames);
assertEquals("[s2, s2, s2, s2, s3, s3, s3, s3]", savedStepNames.toString());
List<String> stepNames = getStepNames(jobExecution);
assertEquals(14, stepNames.size());
assertEquals("[s1, s1, s1:partition0, s1:partition0, s1:partition1, s1:partition1, s2, s2, s2, s2, s3, s3, s3, s3]", stepNames.toString());
}
private List<String> getStepNames(JobExecution jobExecution) {
List<String> list = new ArrayList<String>();
for (StepExecution stepExecution : jobExecution.getStepExecutions()) {
list.add(stepExecution.getStepName());
}
Collections.sort(list);
return list;
}
public static class Decider implements JobExecutionDecider {
int count = 0;
public FlowExecutionStatus decide(JobExecution jobExecution, StepExecution stepExecution) {
if (count++<2) {
return new FlowExecutionStatus("OK");
}
return new FlowExecutionStatus("END");
}
}
}

46
spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/SimpleStepExecutionSplitterTests.java
View File

@@ -14,7 +14,9 @@ import java.util.Set;
import org.junit.Before;
import org.junit.Test;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobExecutionException;
import org.springframework.batch.core.JobInstance;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
@@ -55,11 +57,12 @@ public class SimpleStepExecutionSplitterTests {
@Test
public void testSimpleStepExecutionProviderJobRepositoryStepPartitioner() throws Exception {
final Map<String, ExecutionContext> map = Collections.singletonMap("foo", new ExecutionContext());
SimpleStepExecutionSplitter splitter = new SimpleStepExecutionSplitter(jobRepository, true, step.getName(), new Partitioner() {
public Map<String, ExecutionContext> partition(int gridSize) {
return map;
}
});
SimpleStepExecutionSplitter splitter = new SimpleStepExecutionSplitter(jobRepository, true, step.getName(),
new Partitioner() {
public Map<String, ExecutionContext> partition(int gridSize) {
return map;
}
});
assertEquals(1, splitter.split(stepExecution, 2).size());
}
@@ -118,14 +121,25 @@ public class SimpleStepExecutionSplitterTests {
}
@Test
public void testCompleteStatus() throws Exception {
public void testCompleteStatusAfterFailure() throws Exception {
SimpleStepExecutionSplitter provider = new SimpleStepExecutionSplitter(jobRepository, false, step.getName(),
new SimplePartitioner());
Set<StepExecution> split = provider.split(stepExecution, 2);
assertEquals(2, split.size());
StepExecution nextExecution = update(split, stepExecution, BatchStatus.COMPLETED, false);
// If already complete in another JobExecution we don't execute again
assertEquals(0, provider.split(nextExecution, 2).size());
}
@Test
public void testCompleteStatusSameJobExecution() throws Exception {
SimpleStepExecutionSplitter provider = new SimpleStepExecutionSplitter(jobRepository, false, step.getName(),
new SimplePartitioner());
Set<StepExecution> split = provider.split(stepExecution, 2);
assertEquals(2, split.size());
stepExecution = update(split, stepExecution, BatchStatus.COMPLETED);
// If already complete we don't execute again
assertEquals(0, provider.split(stepExecution, 2).size());
// If already complete in the same JobExecution we should execute again
assertEquals(2, provider.split(stepExecution, 2).size());
}
@Test
@@ -164,6 +178,11 @@ public class SimpleStepExecutionSplitterTests {
private StepExecution update(Set<StepExecution> split, StepExecution stepExecution, BatchStatus status)
throws Exception {
return update(split, stepExecution, status, true);
}
private StepExecution update(Set<StepExecution> split, StepExecution stepExecution, BatchStatus status,
boolean sameJobExecution) throws Exception {
ExecutionContext executionContext = stepExecution.getExecutionContext();
@@ -177,7 +196,16 @@ public class SimpleStepExecutionSplitterTests {
stepExecution.setStatus(status);
jobRepository.update(stepExecution);
stepExecution = stepExecution.getJobExecution().createStepExecution(stepExecution.getStepName());
JobExecution jobExecution = stepExecution.getJobExecution();
if (!sameJobExecution) {
jobExecution.setStatus(BatchStatus.FAILED);
jobExecution.setEndTime(new Date());
jobRepository.update(jobExecution);
JobInstance jobInstance = jobExecution.getJobInstance();
jobExecution = jobRepository.createJobExecution(jobInstance.getJobName(), jobInstance.getJobParameters());
}
stepExecution = jobExecution.createStepExecution(stepExecution.getStepName());
stepExecution.setExecutionContext(executionContext);
jobRepository.add(stepExecution);

26
spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/FlowStepParserTests-context.xml
View File

@@ -1,6 +1,6 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/batch"
xmlns:beans="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
<beans:beans xmlns="http://www.springframework.org/schema/batch" xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/batch http://www.springframework.org/schema/batch/spring-batch-2.1.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd">
@@ -20,6 +20,26 @@
<step id="job2.s4" parent="step4" />
</job>
<job id="job3">
<decision decider="decider" id="job3.d1">
<next on="OK" to="job3.flow" />
<end on="END" />
</decision>
<step id="job3.flow" parent="flow.step" next="job3.d1" />
</job>
<job id="job4">
<step id="job4.flow" parent="startable">
<flow parent="flow" />
<fail on="*" />
</step>
</job>
<!-- BATCH-1745: awkward parent bean for allow-start-if-complete -->
<beans:bean id="startable" abstract="true">
<beans:property name="allowStartIfComplete" value="true" />
</beans:bean>
<flow id="flow" abstract="true">
<step id="s2" parent="step2" next="s3" />
<step id="s3" parent="step3" />
@@ -29,4 +49,6 @@
<flow parent="flow" />
</step>
<beans:bean id="decider" class="org.springframework.batch.core.configuration.xml.FlowStepParserTests$Decider" />
</beans:beans>

33
spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/PartitionStepWithFlowParserTests-context.xml Normal file
View File

@@ -0,0 +1,33 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/batch" xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/batch http://www.springframework.org/schema/batch/spring-batch-2.1.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd">
<beans:import resource="common-context.xml" />
<job id="job1">
<decision decider="decider" id="d1">
<next on="OK" to="s1" />
<end on="END" />
</decision>
<step id="s1" next="d1">
<partition partitioner="partitioner">
<handler grid-size="2" />
<step>
<flow parent="flow" />
</step>
</partition>
</step>
</job>
<flow id="flow" abstract="true">
<step id="s2" parent="step2" next="s3" />
<step id="s3" parent="step3" />
</flow>
<beans:bean id="partitioner" class="org.springframework.batch.core.partition.support.SimplePartitioner" />
<beans:bean id="decider" class="org.springframework.batch.core.configuration.xml.FlowStepParserTests$Decider" />
</beans:beans>