From 1ed2e13b3b48b7e467fc89db38584dc4eff17a4a Mon Sep 17 00:00:00 2001 From: dsyer Date: Tue, 8 Dec 2009 17:37:56 +0000 Subject: [PATCH] RESOLVED - issue BATCH-1458: Add FlowStep: a Step implementation that executes a flow --- .../configuration/xml/AbstractStepParser.java | 50 +++-- .../configuration/xml/FlowElementParser.java | 14 +- .../xml/SimpleFlowFactoryBean.java | 4 +- .../core/configuration/xml/SplitParser.java | 6 +- .../xml/StepParserStepFactoryBean.java | 50 +++++ .../configuration/xml/TopLevelFlowParser.java | 7 + .../batch/core/job/AbstractJob.java | 137 ++---------- .../batch/core/job/SimpleStepHandler.java | 201 ++++++++++++++++++ .../batch/core/job/StepHandler.java | 33 +++ .../batch/core/job/flow/FlowJob.java | 95 +-------- .../batch/core/job/flow/FlowStep.java | 48 +++++ .../batch/core/job/flow/JobFlowExecutor.java | 117 ++++++++++ .../AbstractJobRepositoryFactoryBean.java | 10 + .../configuration/xml/spring-batch-2.1.xsd | 110 +++++++--- .../xml/FlowStepParserTests.java | 97 +++++++++ .../core/job/SimpleStepHandlerTests.java | 106 +++++++++ .../batch/core/job/flow/FlowStepTests.java | 135 ++++++++++++ .../xml/FlowJobParserTests-context.xml | 10 +- .../xml/FlowStepParserTests-context.xml | 32 +++ ...StepParserParentAttributeTests-context.xml | 4 +- .../main/resources/jobs/partitionFileJob.xml | 4 +- 21 files changed, 995 insertions(+), 275 deletions(-) create mode 100644 spring-batch-core/src/main/java/org/springframework/batch/core/job/SimpleStepHandler.java create mode 100644 spring-batch-core/src/main/java/org/springframework/batch/core/job/StepHandler.java create mode 100644 spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowStep.java create mode 100644 spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/JobFlowExecutor.java create mode 100644 spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/FlowStepParserTests.java create mode 100644 spring-batch-core/src/test/java/org/springframework/batch/core/job/SimpleStepHandlerTests.java create mode 100644 spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/FlowStepTests.java create mode 100644 spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/FlowStepParserTests-context.xml diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/AbstractStepParser.java b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/AbstractStepParser.java index 3e21095c1..90776b247 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/AbstractStepParser.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/AbstractStepParser.java @@ -23,6 +23,7 @@ import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.config.RuntimeBeanReference; import org.springframework.beans.factory.parsing.CompositeComponentDefinition; import org.springframework.beans.factory.support.AbstractBeanDefinition; +import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.support.GenericBeanDefinition; import org.springframework.beans.factory.support.ManagedList; import org.springframework.beans.factory.xml.ParserContext; @@ -57,6 +58,8 @@ public abstract class AbstractStepParser { private static final String TASKLET_ELE = "tasklet"; + private static final String FLOW_ELE = "flow"; + private static final String CHUNK_ELE = "chunk"; private static final String LISTENERS_ELE = "listeners"; @@ -79,18 +82,19 @@ public abstract class AbstractStepParser { */ protected AbstractBeanDefinition parseStep(Element stepElement, ParserContext parserContext, String jobFactoryRef) { - AbstractBeanDefinition bd = new GenericBeanDefinition(); + BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(); + AbstractBeanDefinition bd = builder.getRawBeanDefinition(); - @SuppressWarnings("unchecked") - List taskletElements = DomUtils.getChildElementsByTagName(stepElement, TASKLET_ELE); - if (taskletElements.size() == 1) { + Element taskletElement = DomUtils.getChildElementByTagName(stepElement, TASKLET_ELE); + if (taskletElement!=null) { boolean stepUnderspecified = CoreNamespaceUtils.isUnderspecified(stepElement); - parseTasklet(stepElement, taskletElements.get(0), bd, parserContext, stepUnderspecified); + parseTasklet(stepElement, taskletElement, bd, parserContext, stepUnderspecified); } - else if (taskletElements.size() > 1) { - parserContext.getReaderContext().error( - "The '<" + TASKLET_ELE + "/>' element may not appear more than once in a single <" - + stepElement.getNodeName() + "/>.", stepElement); + + Element flowElement = DomUtils.getChildElementByTagName(stepElement, FLOW_ELE); + if (flowElement!=null) { + boolean stepUnderspecified = CoreNamespaceUtils.isUnderspecified(stepElement); + parseFlow(stepElement, flowElement, bd, parserContext, stepUnderspecified); } String parentRef = stepElement.getAttribute(PARENT_ATTR); @@ -103,6 +107,11 @@ public abstract class AbstractStepParser { bd.setAbstract(Boolean.valueOf(isAbstract)); } + String jobRepositoryRef = stepElement.getAttribute(JOB_REPO_ATTR); + if (StringUtils.hasText(jobRepositoryRef)) { + builder.addPropertyReference("jobRepository", jobRepositoryRef); + } + if (StringUtils.hasText(jobFactoryRef)) { bd.setAttribute("jobParserJobFactoryBeanRef", jobFactoryRef); } @@ -157,6 +166,25 @@ public abstract class AbstractStepParser { handleTaskletElement(taskletElement, bd, parserContext); } + private void parseFlow(Element stepElement, Element flowElement, AbstractBeanDefinition bd, + ParserContext parserContext, boolean stepUnderspecified) { + + bd.setBeanClass(StepParserStepFactoryBean.class); + bd.setAttribute("isNamespaceStep", true); + String flowRef = flowElement.getAttribute(PARENT_ATTR); + String idAttribute = stepElement.getAttribute(ID_ATTR); + + BeanDefinition flowDefinition = new GenericBeanDefinition(); + flowDefinition.setParentName(flowRef); + MutablePropertyValues propertyValues = flowDefinition.getPropertyValues(); + if (StringUtils.hasText(idAttribute)) { + propertyValues.addPropertyValue("name", idAttribute); + } + + bd.getPropertyValues().addPropertyValue("flow", flowDefinition); + + } + private void validateTaskletAttributesAndSubelements(Element taskletElement, ParserContext parserContext, boolean stepUnderspecified, String taskletRef, List chunkElements, List beanElements, List refElements) { @@ -288,10 +316,6 @@ public abstract class AbstractStepParser { } private void handleTaskletAttributes(Element taskletElement, MutablePropertyValues propertyValues) { - String jobRepositoryRef = taskletElement.getAttribute(JOB_REPO_ATTR); - if (StringUtils.hasText(jobRepositoryRef)) { - propertyValues.addPropertyValue("jobRepository", new RuntimeBeanReference(jobRepositoryRef)); - } String transactionManagerRef = taskletElement.getAttribute("transaction-manager"); if (StringUtils.hasText(transactionManagerRef)) { propertyValues.addPropertyValue("transactionManager", new RuntimeBeanReference(transactionManagerRef)); diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/FlowElementParser.java b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/FlowElementParser.java index 8a23b641f..c13e75d2a 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/FlowElementParser.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/FlowElementParser.java @@ -35,22 +35,26 @@ import org.w3c.dom.Element; */ public class FlowElementParser { + private static final String ID_ATTR = "id"; + + private static final String REF_ATTR = "parent"; + /** * Parse the flow and turn it into a list of transitions. * * @param element the <flow/gt; element to parse * @param parserContext the parser context for the bean factory - * @return a collection of bean definitions for + * @return a collection of bean definitions for * {@link org.springframework.batch.core.job.flow.support.StateTransition} * instances objects */ public Collection parse(Element element, ParserContext parserContext) { - String refAttribute = element.getAttribute("ref"); - String idAttribute = element.getAttribute("id"); + String refAttribute = element.getAttribute(REF_ATTR); + String idAttribute = element.getAttribute(ID_ATTR); - BeanDefinitionBuilder stateBuilder = - BeanDefinitionBuilder.genericBeanDefinition("org.springframework.batch.core.job.flow.support.state.FlowState"); + BeanDefinitionBuilder stateBuilder = BeanDefinitionBuilder + .genericBeanDefinition("org.springframework.batch.core.job.flow.support.state.FlowState"); AbstractBeanDefinition flowDefinition = new GenericBeanDefinition(); flowDefinition.setParentName(refAttribute); diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/SimpleFlowFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/SimpleFlowFactoryBean.java index 822ea4890..515226ff8 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/SimpleFlowFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/SimpleFlowFactoryBean.java @@ -103,7 +103,7 @@ public class SimpleFlowFactoryBean implements FactoryBean, InitializingBean { String stateName = prefix + oldName; if (state instanceof StepState) { Step step = ((StepState) state).getStep(); - return new StepState(stateName, new DelegateStep(step, stateName)); + return new StepState(stateName, new DelegateStep(stateName, step)); } return new DelegateState(stateName, state); } @@ -154,7 +154,7 @@ public class SimpleFlowFactoryBean implements FactoryBean, InitializingBean { private final String name; - private DelegateStep(Step step, String name) { + private DelegateStep(String name, Step step) { this.step = step; this.name = name; } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/SplitParser.java b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/SplitParser.java index 1064e01f5..0a106c949 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/SplitParser.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/SplitParser.java @@ -44,6 +44,10 @@ import org.w3c.dom.Element; */ public class SplitParser { + /** + * + */ + private static final String PARENT_ATTR = "parent"; private final String jobFactoryRef; /** @@ -91,7 +95,7 @@ public class SplitParser { int i = 0; String prefix = idAttribute.startsWith(jobFactoryRef) ? idAttribute : jobFactoryRef+"."+idAttribute; for (Element nextElement : flowElements) { - String ref = nextElement.getAttribute("ref"); + String ref = nextElement.getAttribute(PARENT_ATTR); if (StringUtils.hasText(ref)) { if (nextElement.getElementsByTagName("*").getLength() > 0) { parserContext.getReaderContext().error("A in a must have ref= or nested , but not both.", nextElement); diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBean.java index 35b32ebda..84e5d7dd2 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBean.java @@ -24,6 +24,8 @@ import org.springframework.batch.classify.BinaryExceptionClassifier; import org.springframework.batch.core.Step; import org.springframework.batch.core.StepExecutionListener; import org.springframework.batch.core.StepListener; +import org.springframework.batch.core.job.flow.Flow; +import org.springframework.batch.core.job.flow.FlowStep; import org.springframework.batch.core.repository.JobRepository; import org.springframework.batch.core.step.item.FaultTolerantStepFactoryBean; import org.springframework.batch.core.step.item.SimpleStepFactoryBean; @@ -79,6 +81,11 @@ class StepParserStepFactoryBean implements FactoryBean, BeanNameAware { private Tasklet tasklet; private PlatformTransactionManager transactionManager; + + // + // Floe Elements + // + private Flow flow; // // Tasklet Elements @@ -164,6 +171,11 @@ class StepParserStepFactoryBean implements FactoryBean, BeanNameAware { configureTaskletStep(ts); return ts; } + else if (flow != null) { + FlowStep ts = new FlowStep(); + configureFlowStep(ts); + return ts; + } else { throw new IllegalStateException("Step [" + name + "] has neither a element nor a 'ref' attribute referencing a Tasklet."); @@ -309,6 +321,33 @@ class StepParserStepFactoryBean implements FactoryBean, BeanNameAware { } } + @SuppressWarnings("serial") + private void configureFlowStep(FlowStep ts) { + if (name != null) { + ts.setName(name); + } + if (allowStartIfComplete != null) { + ts.setAllowStartIfComplete(allowStartIfComplete); + } + if (jobRepository != null) { + ts.setJobRepository(jobRepository); + } + if (startLimit != null) { + ts.setStartLimit(startLimit); + } + if (flow != null) { + ts.setFlow(flow); + } + if (listeners != null) { + int i = 0; + StepExecutionListener[] newListeners = new StepExecutionListener[listeners.length]; + for (StepListener listener : listeners) { + newListeners[i++] = (StepExecutionListener) listener; + } + ts.setStepExecutionListeners(newListeners); + } + } + private void validateFaultTolerantSettings() { validateDependency("skippable-exception-classes", skippableExceptionClasses, "skip-limit", skipLimit, true); validateDependency("retryable-exception-classes", retryableExceptionClasses, "retry-limit", retryLimit, true); @@ -388,6 +427,17 @@ class StepParserStepFactoryBean implements FactoryBean, BeanNameAware { this.name = name; } } + + // ========================================================= + // Flow Attributes + // ========================================================= + + /** + * @param flow the flow to set + */ + public void setFlow(Flow flow) { + this.flow = flow; + } // ========================================================= // Tasklet Attributes diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/TopLevelFlowParser.java b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/TopLevelFlowParser.java index c77a13508..17d304ae8 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/TopLevelFlowParser.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/TopLevelFlowParser.java @@ -17,6 +17,7 @@ package org.springframework.batch.core.configuration.xml; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.util.StringUtils; import org.w3c.dom.Element; /** @@ -27,6 +28,8 @@ public class TopLevelFlowParser extends AbstractFlowParser { private static final String ID_ATTR = "id"; + private static final String ABSTRACT_ATTR = "abstract"; + /** * @param element the top level element containing a flow definition * @param parserContext the {@link ParserContext} @@ -36,6 +39,10 @@ public class TopLevelFlowParser extends AbstractFlowParser { String flowName = element.getAttribute(ID_ATTR); builder.getRawBeanDefinition().setAttribute("flowName", flowName); builder.addPropertyValue("name", flowName); + String abstractAttr = element.getAttribute(ABSTRACT_ATTR); + if (StringUtils.hasText(abstractAttr)) { + builder.setAbstract(abstractAttr.equals("true")); + } super.doParse(element, parserContext, builder); } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/AbstractJob.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/AbstractJob.java index 0f981902b..2551f683e 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/job/AbstractJob.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/AbstractJob.java @@ -27,7 +27,6 @@ import org.springframework.batch.core.Job; import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.JobExecutionException; import org.springframework.batch.core.JobExecutionListener; -import org.springframework.batch.core.JobInstance; import org.springframework.batch.core.JobInterruptedException; import org.springframework.batch.core.JobParametersIncrementer; import org.springframework.batch.core.StartLimitExceededException; @@ -39,7 +38,6 @@ import org.springframework.batch.core.listener.CompositeJobExecutionListener; import org.springframework.batch.core.repository.JobRepository; import org.springframework.batch.core.repository.JobRestartException; import org.springframework.batch.core.step.StepLocator; -import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.repeat.RepeatException; import org.springframework.beans.factory.BeanNameAware; import org.springframework.beans.factory.InitializingBean; @@ -71,6 +69,8 @@ public abstract class AbstractJob implements Job, StepLocator, BeanNameAware, In private JobParametersValidator jobParametersValidator = new DefaultJobParametersValidator(); + private StepHandler stepHandler; + /** * Default constructor. */ @@ -224,6 +224,16 @@ public abstract class AbstractJob implements Job, StepLocator, BeanNameAware, In */ public void setJobRepository(JobRepository jobRepository) { this.jobRepository = jobRepository; + stepHandler = new SimpleStepHandler(jobRepository); + } + + /** + * Convenience method for subclasses to access the job repository. + * + * @return the jobRepository + */ + protected JobRepository getJobRepository() { + return jobRepository; } /** @@ -250,7 +260,7 @@ public abstract class AbstractJob implements Job, StepLocator, BeanNameAware, In public final void execute(JobExecution execution) { logger.debug("Job execution starting: " + execution); - + try { jobParametersValidator.validate(execution.getJobInstance().getJobParameters()); @@ -338,77 +348,8 @@ public abstract class AbstractJob implements Job, StepLocator, BeanNameAware, In */ protected final StepExecution handleStep(Step step, JobExecution execution) throws JobInterruptedException, JobRestartException, StartLimitExceededException { - if (execution.isStopping()) { - throw new JobInterruptedException("JobExecution interrupted."); - } + return stepHandler.handleStep(step, execution); - JobInstance jobInstance = execution.getJobInstance(); - - StepExecution lastStepExecution = jobRepository.getLastStepExecution(jobInstance, step.getName()); - if (stepExecutionPartOfExistingJobExecution(execution, lastStepExecution)) { - // If the last execution of this step was in the same job, it's probably - // intentional so we want to run it again... - logger.info(String.format("Duplicate step [%s] detected in execution of job=[%s]. " + - "If either step fails, both will be executed again on restart.", step.getName(), name)); - lastStepExecution = null; - } - StepExecution currentStepExecution = lastStepExecution; - - if (shouldStart(lastStepExecution, jobInstance, step)) { - - currentStepExecution = execution.createStepExecution(step.getName()); - - boolean isRestart = (lastStepExecution != null && !lastStepExecution.getStatus().equals( - BatchStatus.COMPLETED)); - - if (isRestart) { - currentStepExecution.setExecutionContext(lastStepExecution.getExecutionContext()); - } - else { - currentStepExecution.setExecutionContext(new ExecutionContext()); - } - - jobRepository.add(currentStepExecution); - - logger.info("Executing step: [" + step + "]"); - try { - step.execute(currentStepExecution); - } - catch (JobInterruptedException e) { - // Ensure that the job gets the message that it is stopping - // and can pass it on to other steps that are executing - // concurrently. - execution.setStatus(BatchStatus.STOPPING); - throw e; - } - - jobRepository.updateExecutionContext(execution); - - if (currentStepExecution.getStatus() == BatchStatus.STOPPING - || currentStepExecution.getStatus() == BatchStatus.STOPPED) { - // Ensure that the job gets the message that it is stopping - execution.setStatus(BatchStatus.STOPPING); - throw new JobInterruptedException("Job interrupted by step execution"); - } - - } - else { - // currentStepExecution.setExitStatus(ExitStatus.NOOP); - } - - return currentStepExecution; - - } - - /** - * Detect whether a step execution belongs to this job execution. - * @param jobExecution the current job execution - * @param stepExecution an existing step execution - * @return - */ - private boolean stepExecutionPartOfExistingJobExecution(JobExecution jobExecution, StepExecution stepExecution) { - return stepExecution != null && stepExecution.getJobExecutionId() != null - && stepExecution.getJobExecutionId().equals(jobExecution.getId()); } /** @@ -419,55 +360,7 @@ public abstract class AbstractJob implements Job, StepLocator, BeanNameAware, In * @param stepExecution */ protected void updateStepExecution(StepExecution stepExecution) { - jobRepository.update(stepExecution); - } - - /** - * 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 - * - * @throws StartLimitExceededException if the start limit has been exceeded - * for this step - * @throws JobRestartException if the job is in an inconsistent state from - * an earlier failure - */ - private boolean shouldStart(StepExecution lastStepExecution, JobInstance jobInstance, Step step) - throws JobRestartException, StartLimitExceededException { - - BatchStatus stepStatus; - if (lastStepExecution == null) { - stepStatus = BatchStatus.STARTING; - } - else { - stepStatus = lastStepExecution.getStatus(); - } - - if (stepStatus == BatchStatus.UNKNOWN) { - throw new JobRestartException("Cannot restart step from UNKNOWN status. " - + "The last execution ended with a failure that could not be rolled back, " - + "so it may be dangerous to proceed. Manual intervention is probably necessary."); - } - - if ((stepStatus == BatchStatus.COMPLETED && step.isAllowStartIfComplete() == false) - || stepStatus == BatchStatus.ABANDONED) { - // step is complete, false should be returned, indicating that the - // step should not be started - logger.info("Step already complete or not restartable, so no action to execute: " + lastStepExecution); - return false; - } - - if (jobRepository.getStepExecutionCount(jobInstance, step.getName()) < step.getStartLimit()) { - // step start count is less than start max, return true - return true; - } - else { - // start max has been exceeded, throw an exception. - throw new StartLimitExceededException("Maximum start limit exceeded for step: " + step.getName() - + "StartMax: " + step.getStartLimit()); - } + stepHandler.updateStepExecution(stepExecution); } /** diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/SimpleStepHandler.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/SimpleStepHandler.java new file mode 100644 index 000000000..34c7a039b --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/SimpleStepHandler.java @@ -0,0 +1,201 @@ +/* + * Copyright 2006-2009 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.job; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.batch.core.BatchStatus; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobInstance; +import org.springframework.batch.core.JobInterruptedException; +import org.springframework.batch.core.StartLimitExceededException; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.repository.JobRepository; +import org.springframework.batch.core.repository.JobRestartException; +import org.springframework.batch.item.ExecutionContext; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.util.Assert; + +/** + * @author Dave Syer + * + */ +public class SimpleStepHandler implements StepHandler, InitializingBean { + + private static final Log logger = LogFactory.getLog(SimpleStepHandler.class); + + private JobRepository jobRepository; + + /** + * Convenient default constructor for configuration usage. + */ + public SimpleStepHandler() { + this(null); + } + + /** + * @param jobRepository + */ + public SimpleStepHandler(JobRepository jobRepository) { + super(); + this.jobRepository = jobRepository; + } + + /** + * Check mandatory properties (jobRepository). + * + * @see InitializingBean#afterPropertiesSet() + */ + public void afterPropertiesSet() throws Exception { + Assert.state(jobRepository!=null, "A JobRepository must be provided"); + } + + /** + * @param jobRepository the jobRepository to set + */ + public void setJobRepository(JobRepository jobRepository) { + this.jobRepository = jobRepository; + } + + public StepExecution handleStep(Step step, JobExecution execution) throws JobInterruptedException, + JobRestartException, StartLimitExceededException { + if (execution.isStopping()) { + throw new JobInterruptedException("JobExecution interrupted."); + } + + JobInstance jobInstance = execution.getJobInstance(); + + StepExecution lastStepExecution = jobRepository.getLastStepExecution(jobInstance, step.getName()); + if (stepExecutionPartOfExistingJobExecution(execution, lastStepExecution)) { + // If the last execution of this step was in the same job, it's probably + // intentional so we want to run it again... + logger.info(String.format("Duplicate step [%s] detected in execution of job=[%s]. " + + "If either step fails, both will be executed again on restart.", step.getName(), jobInstance.getJobName())); + lastStepExecution = null; + } + StepExecution currentStepExecution = lastStepExecution; + + if (shouldStart(lastStepExecution, jobInstance, step)) { + + currentStepExecution = execution.createStepExecution(step.getName()); + + boolean isRestart = (lastStepExecution != null && !lastStepExecution.getStatus().equals( + BatchStatus.COMPLETED)); + + if (isRestart) { + currentStepExecution.setExecutionContext(lastStepExecution.getExecutionContext()); + } + else { + currentStepExecution.setExecutionContext(new ExecutionContext()); + } + + jobRepository.add(currentStepExecution); + + logger.info("Executing step: [" + step + "]"); + try { + step.execute(currentStepExecution); + } + catch (JobInterruptedException e) { + // Ensure that the job gets the message that it is stopping + // and can pass it on to other steps that are executing + // concurrently. + execution.setStatus(BatchStatus.STOPPING); + throw e; + } + + jobRepository.updateExecutionContext(execution); + + if (currentStepExecution.getStatus() == BatchStatus.STOPPING + || currentStepExecution.getStatus() == BatchStatus.STOPPED) { + // Ensure that the job gets the message that it is stopping + execution.setStatus(BatchStatus.STOPPING); + throw new JobInterruptedException("Job interrupted by step execution"); + } + + } + else { + // currentStepExecution.setExitStatus(ExitStatus.NOOP); + } + + return currentStepExecution; + } + + public void updateStepExecution(StepExecution stepExecution) { + jobRepository.update(stepExecution); + } + + /** + * Detect whether a step execution belongs to this job execution. + * @param jobExecution the current job execution + * @param stepExecution an existing step execution + * @return + */ + private boolean stepExecutionPartOfExistingJobExecution(JobExecution jobExecution, StepExecution stepExecution) { + return stepExecution != null && stepExecution.getJobExecutionId() != null + && stepExecution.getJobExecutionId().equals(jobExecution.getId()); + } + + /** + * 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 + * + * @throws StartLimitExceededException if the start limit has been exceeded + * for this step + * @throws JobRestartException if the job is in an inconsistent state from + * an earlier failure + */ + private boolean shouldStart(StepExecution lastStepExecution, JobInstance jobInstance, Step step) + throws JobRestartException, StartLimitExceededException { + + BatchStatus stepStatus; + if (lastStepExecution == null) { + stepStatus = BatchStatus.STARTING; + } + else { + stepStatus = lastStepExecution.getStatus(); + } + + if (stepStatus == BatchStatus.UNKNOWN) { + throw new JobRestartException("Cannot restart step from UNKNOWN status. " + + "The last execution ended with a failure that could not be rolled back, " + + "so it may be dangerous to proceed. Manual intervention is probably necessary."); + } + + if ((stepStatus == BatchStatus.COMPLETED && step.isAllowStartIfComplete() == false) + || stepStatus == BatchStatus.ABANDONED) { + // step is complete, false should be returned, indicating that the + // step should not be started + logger.info("Step already complete or not restartable, so no action to execute: " + lastStepExecution); + return false; + } + + if (jobRepository.getStepExecutionCount(jobInstance, step.getName()) < step.getStartLimit()) { + // step start count is less than start max, return true + return true; + } + else { + // start max has been exceeded, throw an exception. + throw new StartLimitExceededException("Maximum start limit exceeded for step: " + step.getName() + + "StartMax: " + step.getStartLimit()); + } + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/StepHandler.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/StepHandler.java new file mode 100644 index 000000000..c52c3199f --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/StepHandler.java @@ -0,0 +1,33 @@ +/* + * Copyright 2006-2009 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.job; + +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobInterruptedException; +import org.springframework.batch.core.StartLimitExceededException; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.repository.JobRestartException; + +public interface StepHandler { + + StepExecution handleStep(Step step, JobExecution execution) throws JobInterruptedException, JobRestartException, + StartLimitExceededException; + + void updateStepExecution(StepExecution stepExecution); + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowJob.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowJob.java index a8d942fce..9ffa83702 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowJob.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowJob.java @@ -18,17 +18,12 @@ package org.springframework.batch.core.job.flow; import java.util.Collection; import java.util.HashSet; -import org.springframework.batch.core.BatchStatus; -import org.springframework.batch.core.ExitStatus; import org.springframework.batch.core.Job; import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.JobExecutionException; -import org.springframework.batch.core.JobInterruptedException; -import org.springframework.batch.core.StartLimitExceededException; import org.springframework.batch.core.Step; -import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.job.AbstractJob; -import org.springframework.batch.core.repository.JobRestartException; +import org.springframework.batch.core.job.SimpleStepHandler; import org.springframework.batch.core.step.StepHolder; /** @@ -99,7 +94,8 @@ public class FlowJob extends AbstractJob { @Override protected void doExecute(final JobExecution execution) throws JobExecutionException { try { - JobFlowExecutor executor = new JobFlowExecutor(execution); + JobFlowExecutor executor = new JobFlowExecutor(new SimpleStepHandler(getJobRepository()), + execution); executor.updateJobExecutionStatus(flow.start(executor).getStatus()); } catch (FlowExecutionException e) { @@ -110,89 +106,4 @@ public class FlowJob extends AbstractJob { } } - /** - * @author Dave Syer - * - */ - private class JobFlowExecutor implements FlowExecutor { - - private final ThreadLocal stepExecutionHolder = new ThreadLocal(); - - private final JobExecution execution; - - private ExitStatus exitStatus = ExitStatus.EXECUTING; - - /** - * @param execution - */ - private JobFlowExecutor(JobExecution execution) { - this.execution = execution; - stepExecutionHolder.set(null); - } - - public String executeStep(Step step) throws JobInterruptedException, JobRestartException, - StartLimitExceededException { - StepExecution stepExecution = handleStep(step, execution); - stepExecutionHolder.set(stepExecution); - return stepExecution == null ? ExitStatus.COMPLETED.getExitCode() : stepExecution.getExitStatus() - .getExitCode(); - } - - public void abandonStepExecution() { - StepExecution lastStepExecution = stepExecutionHolder.get(); - if (lastStepExecution != null && lastStepExecution.getStatus().isGreaterThan(BatchStatus.STOPPING)) { - lastStepExecution.upgradeStatus(BatchStatus.ABANDONED); - updateStepExecution(lastStepExecution); - } - } - - public void updateJobExecutionStatus(FlowExecutionStatus status) { - execution.setStatus(findBatchStatus(status)); - exitStatus = exitStatus.and(new ExitStatus(status.getName())); - execution.setExitStatus(exitStatus); - } - - public JobExecution getJobExecution() { - return execution; - } - - public StepExecution getStepExecution() { - return stepExecutionHolder.get(); - } - - public void close(FlowExecution result) { - stepExecutionHolder.set(null); - } - - public boolean isRestart() { - if (getStepExecution() != null && getStepExecution().getStatus() == BatchStatus.ABANDONED) { - /* - * This is assumed to be the last step execution and it was - * marked abandoned, so we are in a restart of a stopped step. - * TODO: mark the step execution in some more definitive way? - */ - return true; - } - return execution.getStepExecutions().isEmpty(); - } - - public void addExitStatus(String code) { - exitStatus = exitStatus.and(new ExitStatus(code)); - } - - /** - * @param status - * @return - */ - private BatchStatus findBatchStatus(FlowExecutionStatus status) { - for (BatchStatus batchStatus : BatchStatus.values()) { - if (status.getName().startsWith(batchStatus.toString())) { - return batchStatus; - } - } - return BatchStatus.UNKNOWN; - } - - } - } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowStep.java new file mode 100644 index 000000000..873b74253 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowStep.java @@ -0,0 +1,48 @@ +package org.springframework.batch.core.job.flow; + +import org.springframework.batch.core.JobExecutionException; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.job.SimpleStepHandler; +import org.springframework.batch.core.job.StepHandler; +import org.springframework.batch.core.step.AbstractStep; +import org.springframework.util.Assert; + +public class FlowStep extends AbstractStep { + + private Flow flow; + + /** + * Public setter for the flow. + * + * @param flow the flow to set + */ + public void setFlow(Flow flow) { + this.flow = flow; + } + + /** + * Ensure that the flow is set. + * @see AbstractStep#afterPropertiesSet() + */ + @Override + public void afterPropertiesSet() throws Exception { + super.afterPropertiesSet(); + Assert.state(flow!=null, "A Flow must be provided"); + } + + @Override + protected void doExecute(StepExecution stepExecution) throws Exception { + try { + StepHandler stepHandler = new SimpleStepHandler(getJobRepository()); + FlowExecutor executor = new JobFlowExecutor(stepHandler, stepExecution.getJobExecution()); + executor.updateJobExecutionStatus(flow.start(executor).getStatus()); + } + catch (FlowExecutionException e) { + if (e.getCause() instanceof JobExecutionException) { + throw (JobExecutionException) e.getCause(); + } + throw new JobExecutionException("Flow execution ended unexpectedly", e); + } + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/JobFlowExecutor.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/JobFlowExecutor.java new file mode 100644 index 000000000..6dfe94605 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/JobFlowExecutor.java @@ -0,0 +1,117 @@ +/* + * Copyright 2006-2009 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.job.flow; + +import org.springframework.batch.core.BatchStatus; +import org.springframework.batch.core.ExitStatus; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobInterruptedException; +import org.springframework.batch.core.StartLimitExceededException; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.job.StepHandler; +import org.springframework.batch.core.repository.JobRestartException; + +/** + * Implementation of {@link FlowExecutor} for use in components that need to + * execute a flow related to a {@link JobExecution}. + * + * @author Dave Syer + * + */ +public class JobFlowExecutor implements FlowExecutor { + + private final ThreadLocal stepExecutionHolder = new ThreadLocal(); + + private final JobExecution execution; + + private ExitStatus exitStatus = ExitStatus.EXECUTING; + + private final StepHandler stepHandler; + + /** + * @param execution + */ + public JobFlowExecutor(StepHandler stepHandler, JobExecution execution) { + this.stepHandler = stepHandler; + this.execution = execution; + stepExecutionHolder.set(null); + } + + public String executeStep(Step step) throws JobInterruptedException, JobRestartException, + StartLimitExceededException { + StepExecution stepExecution = stepHandler.handleStep(step, execution); + stepExecutionHolder.set(stepExecution); + return stepExecution == null ? ExitStatus.COMPLETED.getExitCode() : stepExecution.getExitStatus().getExitCode(); + } + + public void abandonStepExecution() { + StepExecution lastStepExecution = stepExecutionHolder.get(); + if (lastStepExecution != null && lastStepExecution.getStatus().isGreaterThan(BatchStatus.STOPPING)) { + lastStepExecution.upgradeStatus(BatchStatus.ABANDONED); + stepHandler.updateStepExecution(lastStepExecution); + } + } + + public void updateJobExecutionStatus(FlowExecutionStatus status) { + execution.setStatus(findBatchStatus(status)); + exitStatus = exitStatus.and(new ExitStatus(status.getName())); + execution.setExitStatus(exitStatus); + } + + public JobExecution getJobExecution() { + return execution; + } + + public StepExecution getStepExecution() { + return stepExecutionHolder.get(); + } + + public void close(FlowExecution result) { + stepExecutionHolder.set(null); + } + + public boolean isRestart() { + if (getStepExecution() != null && getStepExecution().getStatus() == BatchStatus.ABANDONED) { + /* + * This is assumed to be the last step execution and it was marked + * abandoned, so we are in a restart of a stopped step. TODO: mark + * the step execution in some more definitive way? + */ + return true; + } + return execution.getStepExecutions().isEmpty(); + } + + public void addExitStatus(String code) { + exitStatus = exitStatus.and(new ExitStatus(code)); + } + + /** + * @param status + * @return + */ + private BatchStatus findBatchStatus(FlowExecutionStatus status) { + for (BatchStatus batchStatus : BatchStatus.values()) { + if (status.getName().startsWith(batchStatus.toString())) { + return batchStatus; + } + } + return BatchStatus.UNKNOWN; + } + +} \ No newline at end of file diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/AbstractJobRepositoryFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/AbstractJobRepositoryFactoryBean.java index 9ac6e611e..57ca1f396 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/AbstractJobRepositoryFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/AbstractJobRepositoryFactoryBean.java @@ -123,6 +123,16 @@ public abstract class AbstractJobRepositoryFactoryBean implements FactoryBean, I return transactionManager; } + /** + * Convenience method for clients to grab the {@link JobRepository} without + * a cast. + * @return the {@link JobRepository} from {@link #getObject()} + * @throws Exception if the repository could not be created + */ + public JobRepository getJobRepository() throws Exception { + return (JobRepository) getObject(); + } + private void initializeProxy() throws Exception { if (proxyFactory == null) { proxyFactory = new ProxyFactory(); diff --git a/spring-batch-core/src/main/resources/org/springframework/batch/core/configuration/xml/spring-batch-2.1.xsd b/spring-batch-core/src/main/resources/org/springframework/batch/core/configuration/xml/spring-batch-2.1.xsd index b4d607cda..0fe5cfa03 100644 --- a/spring-batch-core/src/main/resources/org/springframework/batch/core/configuration/xml/spring-batch-2.1.xsd +++ b/spring-batch-core/src/main/resources/org/springframework/batch/core/configuration/xml/spring-batch-2.1.xsd @@ -110,13 +110,23 @@ - + + - - - - - + + + + + + + + + + @@ -124,6 +134,7 @@ + @@ -136,12 +147,13 @@ - + + @@ -285,10 +297,28 @@ - + - + + + + + + + + + + + + + + + @@ -313,14 +343,16 @@ - + - - + @@ -351,20 +383,24 @@ - Declares job should include an externalized flow here. + Declares job should include an externalized flow + here. - + - - + @@ -426,13 +462,14 @@ - - + + - The tasklet is a reference to another bean definition that implements + The tasklet is a reference to another bean + definition that implements the Tasklet interface. @@ -944,15 +981,18 @@ A pattern to match against the exit status code. Use * and ? as wildcard characters. - When a step finishes the most specific match will be chosen to - select the next step. + When a step finishes + the most specific match will be chosen to + select the next step. + The name of the step to start on when the stopped job is restarted. - Must resolve to one of the other steps in this job. + Must resolve to one of the other steps + in this job. @@ -963,7 +1003,8 @@ Declares job should end at this point, without the possibility of restart. - BatchStatus will be COMPLETED. ExitStatus is configurable. + BatchStatus will be COMPLETED. + ExitStatus is configurable. @@ -971,8 +1012,10 @@ A pattern to match against the exit status code. Use * and ? as wildcard characters. - When a step finishes the most specific match will be chosen to - select the next step. + When a step finishes + the most specific match will be chosen to + select the next step. + A pattern to match against the exit status code. Use * and ? as wildcard characters. - When a step finishes the most specific match will be chosen to - select the next step. + When a step finishes + the most specific match will be chosen to + select the next step. + @@ -1071,7 +1118,8 @@ Should this list be merged with the corresponding list provided - by the parent? If not, it will overwrite the parent list. + by the parent? If not, it will overwrite the parent + list. 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 new file mode 100644 index 000000000..cfb44aa7a --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/FlowStepParserTests.java @@ -0,0 +1,97 @@ +/* + * 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.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.JobParameters; +import org.springframework.batch.core.StepExecution; +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 + * + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class FlowStepParserTests { + + @Autowired + @Qualifier("job1") + private Job job1; + + @Autowired + @Qualifier("job2") + private Job job2; + + @Autowired + private JobRepository jobRepository; + + @Autowired + private MapJobRepositoryFactoryBean mapJobRepositoryFactoryBean; + + @Before + public void setUp() { + mapJobRepositoryFactoryBean.clear(); + } + + @Test + public void testFlowStep() throws Exception { + assertNotNull(job1); + JobExecution jobExecution = jobRepository.createJobExecution(job1.getName(), new JobParameters()); + job1.execute(jobExecution); + assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus()); + List stepNames = getStepNames(jobExecution); + assertEquals(5, stepNames.size()); + assertEquals("[job1.s1, job1.flow, job1.flow.s2, job1.flow.s3, job1.s4]", stepNames.toString()); + } + + @Test + public void testFlowExternalStep() throws Exception { + assertNotNull(job2); + JobExecution jobExecution = jobRepository.createJobExecution(job1.getName(), new JobParameters()); + job2.execute(jobExecution); + assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus()); + List stepNames = getStepNames(jobExecution); + assertEquals(5, stepNames.size()); + assertEquals("[job2.s1, job2.flow, flow.step.s2, flow.step.s3, job2.s4]", stepNames.toString()); + } + + private List getStepNames(JobExecution jobExecution) { + List list = new ArrayList(); + for (StepExecution stepExecution : jobExecution.getStepExecutions()) { + list.add(stepExecution.getStepName()); + } + return list; + } + +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/job/SimpleStepHandlerTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/job/SimpleStepHandlerTests.java new file mode 100644 index 000000000..dbbfde9f7 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/job/SimpleStepHandlerTests.java @@ -0,0 +1,106 @@ +/* + * Copyright 2006-2009 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.job; + +import static org.junit.Assert.*; + +import org.junit.Before; +import org.junit.Test; +import org.springframework.batch.core.BatchStatus; +import org.springframework.batch.core.ExitStatus; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobInterruptedException; +import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.repository.JobRepository; +import org.springframework.batch.core.repository.dao.StepExecutionDao; +import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean; +import org.springframework.batch.core.step.StepSupport; + +/** + * @author Dave Syer + * + */ +public class SimpleStepHandlerTests { + + private JobRepository jobRepository; + + private JobExecution jobExecution; + + private SimpleStepHandler stepHandler; + + private StepExecutionDao stepExecutionDao; + + @Before + public void setUp() throws Exception { + MapJobRepositoryFactoryBean jobRepositoryFactoryBean = new MapJobRepositoryFactoryBean(); + jobRepository = jobRepositoryFactoryBean.getJobRepository(); + stepExecutionDao = jobRepositoryFactoryBean.getStepExecutionDao(); + jobExecution = jobRepository.createJobExecution("job", new JobParameters()); + stepHandler = new SimpleStepHandler(jobRepository); + stepHandler.afterPropertiesSet(); + } + + /** + * Test method for {@link SimpleStepHandler#afterPropertiesSet()}. + */ + @Test(expected = IllegalStateException.class) + public void testAfterPropertiesSet() throws Exception { + SimpleStepHandler stepHandler = new SimpleStepHandler(); + stepHandler.afterPropertiesSet(); + } + + /** + * Test method for + * {@link SimpleStepHandler#handleStep(org.springframework.batch.core.Step, org.springframework.batch.core.JobExecution)} + * . + */ + @Test + public void testHandleStep() throws Exception { + StepExecution stepExecution = stepHandler.handleStep(new StubStep("step"), jobExecution); + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + } + + /** + * Test method for + * {@link SimpleStepHandler#updateStepExecution(org.springframework.batch.core.StepExecution)} + * + */ + @Test + public void testUpdateStepExecution() { + StepExecution stepExecution = jobExecution.createStepExecution("step"); + jobRepository.add(stepExecution); + stepExecution.setStatus(BatchStatus.FAILED); + stepHandler.updateStepExecution(stepExecution); + assertEquals(stepExecution, stepExecutionDao.getStepExecution(jobExecution, stepExecution.getId())); + } + + private class StubStep extends StepSupport { + + private StubStep(String name) { + super(name); + } + + public void execute(StepExecution stepExecution) throws JobInterruptedException { + stepExecution.setStatus(BatchStatus.COMPLETED); + stepExecution.setExitStatus(ExitStatus.COMPLETED); + jobRepository.update(stepExecution); + } + + } + +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/FlowStepTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/FlowStepTests.java new file mode 100644 index 000000000..917d8b095 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/FlowStepTests.java @@ -0,0 +1,135 @@ +/* + * Copyright 2006-2009 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.job.flow; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.fail; + +import java.util.ArrayList; +import java.util.List; + +import org.junit.Before; +import org.junit.Test; +import org.springframework.batch.core.BatchStatus; +import org.springframework.batch.core.ExitStatus; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobInterruptedException; +import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.job.flow.support.SimpleFlow; +import org.springframework.batch.core.job.flow.support.StateTransition; +import org.springframework.batch.core.job.flow.support.state.EndState; +import org.springframework.batch.core.job.flow.support.state.StepState; +import org.springframework.batch.core.repository.JobRepository; +import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean; +import org.springframework.batch.core.step.StepSupport; + +/** + * @author Dave Syer + * + */ +public class FlowStepTests { + + private JobRepository jobRepository; + private JobExecution jobExecution; + + // TODO: add XML support + + @Before + public void setUp() throws Exception { + jobRepository = new MapJobRepositoryFactoryBean().getJobRepository(); + jobExecution = jobRepository.createJobExecution("job", new JobParameters()); + } + + /** + * Test method for {@link org.springframework.batch.core.job.flow.FlowStep#afterPropertiesSet()}. + */ + @Test(expected=IllegalStateException.class) + public void testAfterPropertiesSet() throws Exception{ + FlowStep step = new FlowStep(); + step.setJobRepository(jobRepository); + step.afterPropertiesSet(); + } + + /** + * Test method for {@link org.springframework.batch.core.job.flow.FlowStep#doExecute(org.springframework.batch.core.StepExecution)}. + */ + @Test + public void testDoExecute() throws Exception { + + FlowStep step = new FlowStep(); + step.setJobRepository(jobRepository); + + SimpleFlow flow = new SimpleFlow("job"); + List transitions = new ArrayList(); + transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("step1")), "step2")); + StepState step2 = new StepState(new StubStep("step2")); + transitions.add(StateTransition.createStateTransition(step2, ExitStatus.FAILED.getExitCode(), "end0")); + transitions.add(StateTransition.createStateTransition(step2, ExitStatus.COMPLETED.getExitCode(), "end1")); + transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.FAILED, "end0"))); + transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.COMPLETED, "end1"))); + flow.setStateTransitions(transitions); + + step.setFlow(flow); + step.afterPropertiesSet(); + + StepExecution stepExecution = jobExecution.createStepExecution("step"); + jobRepository.add(stepExecution); + step.execute(stepExecution); + + stepExecution = getStepExecution(jobExecution, "step"); + assertEquals(ExitStatus.COMPLETED, stepExecution.getExitStatus()); + stepExecution = getStepExecution(jobExecution, "step2"); + assertEquals(ExitStatus.COMPLETED, stepExecution.getExitStatus()); + assertEquals(3, jobExecution.getStepExecutions().size()); + + } + + /** + * @author Dave Syer + * + */ + private class StubStep extends StepSupport { + + private StubStep(String name) { + super(name); + } + + public void execute(StepExecution stepExecution) throws JobInterruptedException { + stepExecution.setStatus(BatchStatus.COMPLETED); + stepExecution.setExitStatus(ExitStatus.COMPLETED); + jobRepository.update(stepExecution); + } + + } + + /** + * @param jobExecution + * @param stepName + * @return the StepExecution corresponding to the specified step + */ + private StepExecution getStepExecution(JobExecution jobExecution, String stepName) { + for (StepExecution stepExecution : jobExecution.getStepExecutions()) { + if (stepExecution.getStepName().equals(stepName)) { + return stepExecution; + } + } + fail("No stepExecution found with name: [" + stepName + "]"); + return null; + } + +} diff --git a/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/FlowJobParserTests-context.xml b/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/FlowJobParserTests-context.xml index dfd781e84..ec02533b3 100644 --- a/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/FlowJobParserTests-context.xml +++ b/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/FlowJobParserTests-context.xml @@ -8,12 +8,12 @@ - + - + @@ -22,13 +22,13 @@ - + - - + + 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 new file mode 100644 index 000000000..977a72055 --- /dev/null +++ b/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/FlowStepParserTests-context.xml @@ -0,0 +1,32 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/StepParserParentAttributeTests-context.xml b/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/StepParserParentAttributeTests-context.xml index 31e9f3734..1489c9133 100644 --- a/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/StepParserParentAttributeTests-context.xml +++ b/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/StepParserParentAttributeTests-context.xml @@ -127,8 +127,8 @@ - - + + diff --git a/spring-batch-samples/src/main/resources/jobs/partitionFileJob.xml b/spring-batch-samples/src/main/resources/jobs/partitionFileJob.xml index d32b4afa2..c33af1a27 100644 --- a/spring-batch-samples/src/main/resources/jobs/partitionFileJob.xml +++ b/spring-batch-samples/src/main/resources/jobs/partitionFileJob.xml @@ -34,8 +34,8 @@ - - + +