From 3f38056ff3381b404a93fe540844b3919d1a00d3 Mon Sep 17 00:00:00 2001 From: Dave Syer Date: Thu, 5 May 2011 08:58:48 +0100 Subject: [PATCH] BATCH-1730: allow flow to be repeated in a PartitionStep --- .../support/SimpleStepExecutionSplitter.java | 15 ++- .../xml/FlowStepParserTests.java | 59 +++++++++- .../xml/PartitionStepWithFlowParserTests.java | 107 ++++++++++++++++++ .../SimpleStepExecutionSplitterTests.java | 46 ++++++-- .../xml/FlowStepParserTests-context.xml | 26 ++++- ...rtitionStepWithFlowParserTests-context.xml | 33 ++++++ 6 files changed, 269 insertions(+), 17 deletions(-) create mode 100644 spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/PartitionStepWithFlowParserTests.java create mode 100644 spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/PartitionStepWithFlowParserTests-context.xml diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/SimpleStepExecutionSplitter.java b/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/SimpleStepExecutionSplitter.java index 2cc8e302b..e6848a28d 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/SimpleStepExecutionSplitter.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/SimpleStepExecutionSplitter.java @@ -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()); + } + } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/FlowStepParserTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/FlowStepParserTests.java index 5e3324fc5..74998f1a8 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/FlowStepParserTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/FlowStepParserTests.java @@ -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 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 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 getStepNames(JobExecution jobExecution) { List list = new ArrayList(); 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"); + } + + } } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/PartitionStepWithFlowParserTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/PartitionStepWithFlowParserTests.java new file mode 100644 index 000000000..9a078ecfd --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/PartitionStepWithFlowParserTests.java @@ -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 savedStepNames = new ArrayList(); + + @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 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 getStepNames(JobExecution jobExecution) { + List list = new ArrayList(); + 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"); + } + + } + +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/SimpleStepExecutionSplitterTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/SimpleStepExecutionSplitterTests.java index 9d56b8c92..c306e3a56 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/SimpleStepExecutionSplitterTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/SimpleStepExecutionSplitterTests.java @@ -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 map = Collections.singletonMap("foo", new ExecutionContext()); - SimpleStepExecutionSplitter splitter = new SimpleStepExecutionSplitter(jobRepository, true, step.getName(), new Partitioner() { - public Map partition(int gridSize) { - return map; - } - }); + SimpleStepExecutionSplitter splitter = new SimpleStepExecutionSplitter(jobRepository, true, step.getName(), + new Partitioner() { + public Map 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 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 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 split, StepExecution stepExecution, BatchStatus status) throws Exception { + return update(split, stepExecution, status, true); + } + + private StepExecution update(Set 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); diff --git a/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/FlowStepParserTests-context.xml b/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/FlowStepParserTests-context.xml index da2e84e85..38b64438a 100644 --- a/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/FlowStepParserTests-context.xml +++ b/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/FlowStepParserTests-context.xml @@ -1,6 +1,6 @@ - @@ -20,6 +20,26 @@ + + + + + + + + + + + + + + + + + + + + @@ -29,4 +49,6 @@ + + \ No newline at end of file diff --git a/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/PartitionStepWithFlowParserTests-context.xml b/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/PartitionStepWithFlowParserTests-context.xml new file mode 100644 index 000000000..623acdbef --- /dev/null +++ b/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/PartitionStepWithFlowParserTests-context.xml @@ -0,0 +1,33 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file