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 9fdd20fb8..38cb5b1f0 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 @@ -55,10 +55,23 @@ public class SimpleFlowFactoryBean implements FactoryBean, InitializingBean { private Comparator stateTransitionComparator; + private Class flowType; + + /** + * @param stateTransitionComparator {@link Comparator} implementation that addresses + * the ordering of state evaluation + */ public void setStateTransitionComparator(Comparator stateTransitionComparator) { this.stateTransitionComparator = stateTransitionComparator; } + /** + * @param flowType Used to inject the type of flow (regular Spring Batch or JSR-352) + */ + public void setFlowType(Class flowType) { + this.flowType = flowType; + } + /** * The name of the flow that is created by this factory. * @@ -88,12 +101,18 @@ public class SimpleFlowFactoryBean implements FactoryBean, InitializingBean { @Override public void afterPropertiesSet() throws Exception { Assert.hasText(name, "The flow must have a name"); + + if(flowType == null) { + flowType = SimpleFlow.class; + } } + /* (non-Javadoc) + * @see org.springframework.beans.factory.FactoryBean#getObject() + */ @Override public Object getObject() throws Exception { - - SimpleFlow flow = new SimpleFlow(name); + SimpleFlow flow = flowType.getConstructor(String.class).newInstance(name); flow.setStateTransitionComparator(stateTransitionComparator); diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/support/SimpleFlow.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/support/SimpleFlow.java index 82f76b3f0..7b3feed10 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/support/SimpleFlow.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/support/SimpleFlow.java @@ -171,7 +171,7 @@ public class SimpleFlow implements Flow, InitializingBean { logger.debug("Completed state="+stateName+" with status="+status); - state = nextState(stateName, status); + state = nextState(stateName, status, stepExecution); } FlowExecution result = new FlowExecution(stateName, status); @@ -180,28 +180,19 @@ public class SimpleFlow implements Flow, InitializingBean { } - private boolean isFlowContinued(State state, FlowExecutionStatus status, StepExecution stepExecution) { - boolean continued = true; + protected Map> getTransitionMap() { + return transitionMap; + } - continued = state != null && status!=FlowExecutionStatus.STOPPED; - - if(stepExecution != null) { - Boolean reRun = (Boolean) stepExecution.getExecutionContext().get("batch.restart"); - - if(reRun != null && reRun && status == FlowExecutionStatus.STOPPED && !state.getName().endsWith(stepExecution.getStepName())) { - continued = true; - } - } - - return continued; + protected Map getStateMap() { + return stateMap; } /** * @return the next {@link Step} (or null if this is the end) * @throws JobExecutionException */ - private State nextState(String stateName, FlowExecutionStatus status) throws FlowExecutionException { - + protected State nextState(String stateName, FlowExecutionStatus status, StepExecution stepExecution) throws FlowExecutionException { Set set = transitionMap.get(stateName); if (set == null) { @@ -238,6 +229,22 @@ public class SimpleFlow implements Flow, InitializingBean { } + private boolean isFlowContinued(State state, FlowExecutionStatus status, StepExecution stepExecution) { + boolean continued = true; + + continued = state != null && status!=FlowExecutionStatus.STOPPED; + + if(stepExecution != null) { + Boolean reRun = (Boolean) stepExecution.getExecutionContext().get("batch.restart"); + + if(reRun != null && reRun && status == FlowExecutionStatus.STOPPED && !state.getName().endsWith(stepExecution.getStepName())) { + continued = true; + } + } + + return continued; + } + /** * Analyse the transitions provided and generate all the information needed * to execute the flow. diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/StepContext.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/StepContext.java index f4f262d68..d3d915d87 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/StepContext.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/StepContext.java @@ -18,13 +18,16 @@ package org.springframework.batch.core.jsr; import java.io.Serializable; import java.util.List; import java.util.Properties; +import java.util.concurrent.atomic.AtomicBoolean; import javax.batch.runtime.BatchStatus; import javax.batch.runtime.Metric; import org.springframework.batch.core.ExitStatus; import org.springframework.batch.core.StepExecution; +import org.springframework.batch.item.util.ExecutionContextUserSupport; import org.springframework.util.Assert; +import org.springframework.util.ClassUtils; /** * Wrapper class to provide the {@link javax.batch.runtime.context.StepContext} functionality @@ -36,10 +39,12 @@ import org.springframework.util.Assert; * @since 3.0 */ public class StepContext implements javax.batch.runtime.context.StepContext { - + private final static String PERSISTENT_USER_DATA_KEY = "batch_jsr_persistentUserData"; private StepExecution stepExecution; private Object transientUserData; private Properties properties = new Properties(); + private AtomicBoolean exitStatusSet = new AtomicBoolean(); + private final ExecutionContextUserSupport executionContextUserSupport = new ExecutionContextUserSupport(ClassUtils.getShortName(StepContext.class)); public StepContext(StepExecution stepExecution, Properties properties) { Assert.notNull(stepExecution, "A StepExecution is required"); @@ -93,7 +98,7 @@ public class StepContext implements javax.batch.runtime.context.StepContext { */ @Override public Serializable getPersistentUserData() { - return (Serializable) stepExecution.getExecutionContext().get("batch_jsr_persistentUserData"); + return (Serializable) stepExecution.getExecutionContext().get(executionContextUserSupport.getKey(PERSISTENT_USER_DATA_KEY)); } /* (non-Javadoc) @@ -101,7 +106,7 @@ public class StepContext implements javax.batch.runtime.context.StepContext { */ @Override public void setPersistentUserData(Serializable data) { - stepExecution.getExecutionContext().put("batch_jsr_persistentUserData", data); + stepExecution.getExecutionContext().put(executionContextUserSupport.getKey(PERSISTENT_USER_DATA_KEY), data); } /* (non-Javadoc) @@ -117,7 +122,7 @@ public class StepContext implements javax.batch.runtime.context.StepContext { */ @Override public String getExitStatus() { - return stepExecution.getExitStatus().getExitCode(); + return exitStatusSet.get() ? stepExecution.getExitStatus().getExitCode() : null; } /* (non-Javadoc) @@ -126,6 +131,7 @@ public class StepContext implements javax.batch.runtime.context.StepContext { @Override public void setExitStatus(String status) { stepExecution.setExitStatus(new ExitStatus(status)); + exitStatusSet.set(true); } /** diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/StepContextFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/StepContextFactoryBean.java index 4594fe496..74006466d 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/StepContextFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/StepContextFactoryBean.java @@ -17,10 +17,15 @@ package org.springframework.batch.core.jsr; import java.util.Properties; +import javax.batch.runtime.context.StepContext; + import org.springframework.batch.core.jsr.configuration.support.BatchPropertyContext; import org.springframework.batch.core.scope.context.StepSynchronizationManager; import org.springframework.beans.factory.FactoryBean; +import org.springframework.beans.factory.FactoryBeanNotInitializedException; +import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.util.Assert; /** * {@link FactoryBean} implementation used to create {@link javax.batch.runtime.context.StepContext} @@ -30,20 +35,53 @@ import org.springframework.beans.factory.annotation.Autowired; * @author Chris Schaefer * @since 3.0 */ -public class StepContextFactoryBean implements FactoryBean { +public class StepContextFactoryBean implements FactoryBean, InitializingBean { @Autowired private BatchPropertyContext batchPropertyContext; + private static final ThreadLocal contextHolder = new ThreadLocal(); + + protected void setBatchPropertyContext(BatchPropertyContext batchPropertyContext) { + this.batchPropertyContext = batchPropertyContext; + } + /* (non-Javadoc) * @see org.springframework.beans.factory.FactoryBean#getObject() */ @Override public StepContext getObject() throws Exception { - org.springframework.batch.core.scope.context.StepContext stepContext = StepSynchronizationManager.getContext(); + return getCurrent(); + } - Properties properties = batchPropertyContext.getStepProperties(stepContext.getStepName()); + private javax.batch.runtime.context.StepContext getCurrent() { + org.springframework.batch.core.StepExecution curStepExecution = null; - return new StepContext(stepContext.getStepExecution(), properties); + if(StepSynchronizationManager.getContext() != null) { + curStepExecution = StepSynchronizationManager.getContext().getStepExecution(); + } + + if(curStepExecution == null) { + throw new FactoryBeanNotInitializedException("A StepExecution is required"); + } + + StepContext context = contextHolder.get(); + + // If the current context applies to the current step, use it + if(context != null && context.getStepExecutionId() == curStepExecution.getId()) { + return context; + } + + Properties stepProperties = batchPropertyContext.getStepProperties(curStepExecution.getStepName()); + + if(stepProperties != null) { + context = new org.springframework.batch.core.jsr.StepContext(curStepExecution, stepProperties); + } else { + context = new org.springframework.batch.core.jsr.StepContext(curStepExecution, new Properties()); + } + + contextHolder.set(context); + + return context; } /* (non-Javadoc) @@ -61,4 +99,9 @@ public class StepContextFactoryBean implements FactoryBean { public boolean isSingleton() { return false; } + + @Override + public void afterPropertiesSet() throws Exception { + Assert.notNull(batchPropertyContext, "BatchPropertyContext is required"); + } } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/StepExecution.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/StepExecution.java index 586527ee9..3fc971565 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/StepExecution.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/StepExecution.java @@ -22,7 +22,9 @@ import javax.batch.runtime.BatchStatus; import javax.batch.runtime.Metric; import org.springframework.batch.core.ExitStatus; +import org.springframework.batch.item.util.ExecutionContextUserSupport; import org.springframework.util.Assert; +import org.springframework.util.ClassUtils; /** * Implementation of the StepExecution as defined in JSR-352. This implementation @@ -34,7 +36,10 @@ import org.springframework.util.Assert; */ public class StepExecution implements javax.batch.runtime.StepExecution{ + private final static String PERSISTENT_USER_DATA_KEY = "batch_jsr_persistentUserData"; private final org.springframework.batch.core.StepExecution stepExecution; + // The API for the persistent user data is handled by the StepContext which is why the name here is based on the StepContext. + private final ExecutionContextUserSupport executionContextUserSupport = new ExecutionContextUserSupport(ClassUtils.getShortName(StepContext.class)); /** * @param stepExecution The {@link org.springframework.batch.core.StepExecution} used @@ -105,7 +110,7 @@ public class StepExecution implements javax.batch.runtime.StepExecution{ */ @Override public Serializable getPersistentUserData() { - return (Serializable) stepExecution.getExecutionContext().get("batch_jsr_persistentUserData"); + return (Serializable) stepExecution.getExecutionContext().get(executionContextUserSupport.getKey(PERSISTENT_USER_DATA_KEY)); } /* (non-Javadoc) diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/support/JsrBeanScopBeanFactoryPostProcessor.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/support/JsrBeanScopBeanFactoryPostProcessor.java new file mode 100644 index 000000000..41b90bf46 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/support/JsrBeanScopBeanFactoryPostProcessor.java @@ -0,0 +1,103 @@ +/* + * Copyright 2013 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.jsr.configuration.support; + +import javax.batch.api.partition.PartitionAnalyzer; +import javax.batch.api.partition.PartitionMapper; +import javax.batch.api.partition.PartitionReducer; + +import org.springframework.batch.core.jsr.configuration.xml.StepFactoryBean; +import org.springframework.batch.core.jsr.partition.JsrPartitionHandler; +import org.springframework.beans.BeansException; +import org.springframework.beans.PropertyValue; +import org.springframework.beans.factory.config.BeanDefinition; +import org.springframework.beans.factory.config.BeanFactoryPostProcessor; +import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; +import org.springframework.beans.factory.config.RuntimeBeanReference; +import org.springframework.beans.factory.support.AbstractBeanDefinition; + +/** + * In order for property resolution to occur correctly within the scope of a JSR-352 + * batch job, initialization of job level artifacts must occur on the same thread that + * the job is executing. To allow this to occur, {@link PartitionMapper}, + * {@link PartitionReducer}, and {@link PartitionAnalyzer} are all configured to + * lazy initialization (equivalent to lazy-init="true"). + * + * @author Michael Minella + * @since 3.0 + */ +public class JsrBeanScopBeanFactoryPostProcessor implements BeanFactoryPostProcessor { + + private JobLevelBeanLazyInitializer initializer; + + /* (non-Javadoc) + * @see org.springframework.beans.factory.config.BeanFactoryPostProcessor#postProcessBeanFactory(org.springframework.beans.factory.config.ConfigurableListableBeanFactory) + */ + @Override + public void postProcessBeanFactory( + ConfigurableListableBeanFactory beanFactory) throws BeansException { + if (initializer == null) { + this.initializer = new JobLevelBeanLazyInitializer(beanFactory); + } + + String[] beanNames = beanFactory.getBeanDefinitionNames(); + + for (String curName : beanNames) { + initializer.visitBeanDefinition(beanFactory.getBeanDefinition(curName)); + } + } + + /** + * Looks for beans that may have dependencies that need to be lazily initialized and + * configures the corresponding {@link BeanDefinition} accordingly. + * + * @author Michael Minella + * @since 3.0 + */ + public static class JobLevelBeanLazyInitializer { + + private ConfigurableListableBeanFactory beanFactory; + + public JobLevelBeanLazyInitializer(ConfigurableListableBeanFactory beanFactory) { + this.beanFactory = beanFactory; + } + + public void visitBeanDefinition(BeanDefinition beanDefinition) { + String beanClassName = beanDefinition.getBeanClassName(); + + if(StepFactoryBean.class.getName().equals(beanClassName)) { + PropertyValue [] values = ((AbstractBeanDefinition) beanDefinition).getPropertyValues().getPropertyValues(); + for (PropertyValue propertyValue : values) { + if(propertyValue.getName().equalsIgnoreCase("partitionReducer")) { + RuntimeBeanReference ref = (RuntimeBeanReference) propertyValue.getValue(); + beanFactory.getBeanDefinition(ref.getBeanName()).setLazyInit(true); + } + } + } + + if(JsrPartitionHandler.class.getName().equals(beanClassName)) { + PropertyValue [] values = ((AbstractBeanDefinition) beanDefinition).getPropertyValues().getPropertyValues(); + for (PropertyValue propertyValue : values) { + String propertyName = propertyValue.getName(); + if(propertyName.equalsIgnoreCase("partitionMapper") || propertyName.equalsIgnoreCase("partitionAnalyzer")) { + RuntimeBeanReference ref = (RuntimeBeanReference) propertyValue.getValue(); + beanFactory.getBeanDefinition(ref.getBeanName()).setLazyInit(true); + } + } + } + } + } +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/FlowParser.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/FlowParser.java index e4f6574b1..6670ecacf 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/FlowParser.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/FlowParser.java @@ -27,6 +27,7 @@ import org.springframework.batch.core.configuration.xml.AbstractFlowParser; import org.springframework.batch.core.configuration.xml.SimpleFlowFactoryBean; import org.springframework.batch.core.job.flow.FlowExecutionStatus; import org.springframework.batch.core.job.flow.support.DefaultStateTransitionComparator; +import org.springframework.batch.core.jsr.job.flow.support.DefaultFlow; import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.config.RuntimeBeanReference; import org.springframework.beans.factory.support.BeanDefinitionBuilder; @@ -92,6 +93,7 @@ public class FlowParser extends AbstractFlowParser { protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) { builder.getRawBeanDefinition().setAttribute("flowName", flowName); builder.addPropertyValue("name", flowName); + builder.addPropertyValue("flowType", DefaultFlow.class); List stateTransitions = new ArrayList(); @@ -179,15 +181,10 @@ public class FlowParser extends AbstractFlowParser { list.addAll(createTransition(FlowExecutionStatus.UNKNOWN, FlowExecutionStatus.UNKNOWN.getName(), null, null, stateDef, parserContext, false)); if (!hasNextAttribute) { - list.addAll(createTransition(FlowExecutionStatus.COMPLETED, null, null, null, stateDef, parserContext, + list.addAll(createTransition(FlowExecutionStatus.COMPLETED, FlowExecutionStatus.COMPLETED.getName(), null, null, stateDef, parserContext, false)); } } - else if (hasNextAttribute) { - parserContext.getReaderContext().error( - "The <" + element.getNodeName() + "/> may not contain a 'next" - + "' attribute and a transition element", element); - } return list; } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/JsrNamespaceUtils.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/JsrNamespaceUtils.java index 7bcbc447f..c3281df09 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/JsrNamespaceUtils.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/JsrNamespaceUtils.java @@ -16,8 +16,10 @@ package org.springframework.batch.core.jsr.configuration.xml; import java.util.HashMap; + import org.springframework.batch.core.jsr.configuration.support.BatchPropertyBeanPostProcessor; import org.springframework.batch.core.jsr.configuration.support.JsrAutowiredAnnotationBeanPostProcessor; +import org.springframework.batch.core.jsr.configuration.support.JsrBeanScopBeanFactoryPostProcessor; import org.springframework.batch.core.jsr.configuration.support.ThreadLocalClassloaderBeanPostProcessor; import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.support.AbstractBeanDefinition; @@ -36,12 +38,19 @@ class JsrNamespaceUtils { private static final String JOB_PROPERTIES_BEAN_NAME = "jobProperties"; private static final String BATCH_PROPERTY_POST_PROCESSOR_BEAN_NAME = "batchPropertyPostProcessor"; private static final String THREAD_LOCAL_CLASS_LOADER_BEAN_POST_PROCESSOR_BEAN_NAME = "threadLocalClassloaderBeanPostProcessor"; + private static final String BEAN_SCOPE_POST_PROCESSOR_BEAN_NAME = "beanScopeBeanPostProcessor"; static void autoregisterJsrBeansForNamespace(ParserContext parserContext) { autoRegisterJobProperties(parserContext); autoRegisterBatchPostProcessor(parserContext); autoRegisterJsrAutowiredAnnotationBeanPostProcessor(parserContext); autoRegisterThreadLocalClassloaderBeanPostProcessor(parserContext); + autoRegisterBeanScopeBeanFactoryPostProcessor(parserContext); + } + + private static void autoRegisterBeanScopeBeanFactoryPostProcessor( + ParserContext parserContext) { + registerPostProcessor(parserContext, JsrBeanScopBeanFactoryPostProcessor.class, BeanDefinition.ROLE_INFRASTRUCTURE, BEAN_SCOPE_POST_PROCESSOR_BEAN_NAME); } private static void autoRegisterBatchPostProcessor(ParserContext parserContext) { diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/job/flow/support/DefaultFlow.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/job/flow/support/DefaultFlow.java new file mode 100644 index 000000000..9683048f9 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/job/flow/support/DefaultFlow.java @@ -0,0 +1,104 @@ +/* + * Copyright 2013 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.jsr.job.flow.support; + +import java.util.Set; + +import org.springframework.batch.core.JobExecutionException; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.job.flow.Flow; +import org.springframework.batch.core.job.flow.FlowExecutionException; +import org.springframework.batch.core.job.flow.FlowExecutionStatus; +import org.springframework.batch.core.job.flow.State; +import org.springframework.batch.core.job.flow.support.SimpleFlow; +import org.springframework.batch.core.job.flow.support.StateTransition; + +/** + * Implements JSR-352 specific logic around the execution of a flow. Specifically, this + * {@link Flow} implementation will attempt to find the next state based on the provided + * exit status. If none is found (the exit status isn't mapped), it will attempt to + * resolve the next state basing it on the last step's batch status. Only if both + * attempts fail, the flow will fail due to the inability to find the next state. + * + * @author Michael Minella + * @since 3.0 + */ +public class DefaultFlow extends SimpleFlow { + + /** + * @param name name of the flow + */ + public DefaultFlow(String name) { + super(name); + } + + /** + * @return the next {@link Step} (or null if this is the end) + * @throws JobExecutionException + */ + @Override + protected State nextState(String stateName, FlowExecutionStatus status, StepExecution stepExecution) throws FlowExecutionException { + Set set = getTransitionMap().get(stateName); + + if (set == null) { + throw new FlowExecutionException(String.format("No transitions found in flow=%s for state=%s", getName(), + stateName)); + } + + String next = null; + String exitCode = status.getName(); + for (StateTransition stateTransition : set) { + if (stateTransition.matches(exitCode) || (exitCode.equals("PENDING") && stateTransition.matches("STOPPED"))) { + if (stateTransition.isEnd()) { + // End of job + return null; + } + next = stateTransition.getNext(); + break; + } + } + + if (next == null) { + if(stepExecution != null) { + exitCode = stepExecution.getStatus().toString(); + + for (StateTransition stateTransition : set) { + if (stateTransition.matches(exitCode) || (exitCode.equals("PENDING") && stateTransition.matches("STOPPED"))) { + if (stateTransition.isEnd()) { + // End of job + return null; + } + next = stateTransition.getNext(); + break; + } + } + } + + if(next == null) { + throw new FlowExecutionException(String.format( + "Next state not found in flow=%s for state=%s with exit status=%s", getName(), stateName, status.getName())); + } + } + + if (!getStateMap().containsKey(next)) { + throw new FlowExecutionException(String.format("Next state not specified in flow=%s for next=%s", + getName(), next)); + } + + return getStateMap().get(next); + } +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/partition/support/JsrStepExecutionAggregator.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/partition/support/JsrStepExecutionAggregator.java new file mode 100644 index 000000000..a4e11e965 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/partition/support/JsrStepExecutionAggregator.java @@ -0,0 +1,56 @@ +/* + * Copyright 2013 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.jsr.partition.support; + +import java.util.Collection; + +import org.springframework.batch.core.BatchStatus; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.partition.support.StepExecutionAggregator; +import org.springframework.util.Assert; + +/** + * Aggregates {@link StepExecution}s based on the rules outlined in JSR-352. Specifically + * it aggregates all counts and determines the correct BatchStatus. However, the ExitStatus + * for each child StepExecution is ignored. + * + * @author Michael Minella + * @since 3.0 + */ +public class JsrStepExecutionAggregator implements StepExecutionAggregator { + + /* (non-Javadoc) + * @see org.springframework.batch.core.partition.support.StepExecutionAggregator#aggregate(org.springframework.batch.core.StepExecution, java.util.Collection) + */ + @Override + public void aggregate(StepExecution result, + Collection executions) { + Assert.notNull(result, "To aggregate into a result it must be non-null."); + if (executions == null) { + return; + } + for (StepExecution stepExecution : executions) { + BatchStatus status = stepExecution.getStatus(); + result.setStatus(BatchStatus.max(result.getStatus(), status)); + result.setCommitCount(result.getCommitCount() + stepExecution.getCommitCount()); + result.setRollbackCount(result.getRollbackCount() + stepExecution.getRollbackCount()); + result.setReadCount(result.getReadCount() + stepExecution.getReadCount()); + result.setReadSkipCount(result.getReadSkipCount() + stepExecution.getReadSkipCount()); + result.setWriteCount(result.getWriteCount() + stepExecution.getWriteCount()); + result.setWriteSkipCount(result.getWriteSkipCount() + stepExecution.getWriteSkipCount()); + } + } +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/PartitionStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/PartitionStep.java index b1d2fc949..8f0d0f547 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/PartitionStep.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/PartitionStep.java @@ -15,6 +15,8 @@ */ package org.springframework.batch.core.jsr.step; +import java.util.Collection; + import javax.batch.api.partition.PartitionReducer; import javax.batch.api.partition.PartitionReducer.PartitionStatus; @@ -22,8 +24,10 @@ import org.springframework.batch.core.BatchStatus; import org.springframework.batch.core.JobExecutionException; import org.springframework.batch.core.Step; import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.jsr.partition.support.JsrStepExecutionAggregator; import org.springframework.batch.core.partition.PartitionHandler; import org.springframework.batch.core.partition.StepExecutionSplitter; +import org.springframework.batch.core.partition.support.StepExecutionAggregator; import org.springframework.batch.item.ExecutionContext; /** @@ -38,6 +42,7 @@ public class PartitionStep extends org.springframework.batch.core.partition.supp private PartitionReducer reducer; private boolean hasReducer = false; + private StepExecutionAggregator stepExecutionAggregator = new JsrStepExecutionAggregator(); public void setPartitionReducer(PartitionReducer reducer) { this.reducer = reducer; @@ -65,13 +70,13 @@ public class PartitionStep extends org.springframework.batch.core.partition.supp } // Wait for task completion and then aggregate the results - getPartitionHandler().handle(getStepExecutionSplitter(), stepExecution); + Collection stepExecutions = getPartitionHandler().handle(getStepExecutionSplitter(), stepExecution); stepExecution.upgradeStatus(BatchStatus.COMPLETED); + stepExecutionAggregator.aggregate(stepExecution, stepExecutions); if (stepExecution.getStatus().isUnsuccessful()) { if (hasReducer) { reducer.rollbackPartitionedStep(); - reducer.beforePartitionedStepCompletion(); reducer.afterPartitionedStepCompletion(PartitionStatus.ROLLBACK); } throw new JobExecutionException("Partition handler returned an unsuccessful step"); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/StateSupport.java b/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/StateSupport.java index 317cf0d90..78ac1fb1d 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/StateSupport.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/StateSupport.java @@ -26,7 +26,7 @@ import org.springframework.batch.core.job.flow.support.state.AbstractState; */ public class StateSupport extends AbstractState { - private FlowExecutionStatus status; + protected FlowExecutionStatus status; public StateSupport(String name) { this(name, FlowExecutionStatus.COMPLETED); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/support/SimpleFlowTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/support/SimpleFlowTests.java index 17963ce93..19904b67a 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/support/SimpleFlowTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/support/SimpleFlowTests.java @@ -25,6 +25,7 @@ import java.util.ArrayList; import java.util.Collections; import java.util.List; +import org.junit.Before; import org.junit.Test; import org.springframework.batch.core.ExitStatus; import org.springframework.batch.core.job.flow.FlowExecution; @@ -41,9 +42,14 @@ import org.springframework.batch.core.job.flow.StateSupport; */ public class SimpleFlowTests { - private SimpleFlow flow = new SimpleFlow("job"); + protected SimpleFlow flow; - private FlowExecutor executor = new JobFlowExecutorSupport(); + protected FlowExecutor executor = new JobFlowExecutorSupport(); + + @Before + public void setUp() { + flow = new SimpleFlow("job"); + } @Test(expected = IllegalArgumentException.class) public void testEmptySteps() throws Exception { @@ -205,22 +211,13 @@ public class SimpleFlowTests { assertNull(state); } - private List collect(StateTransition s1, StateTransition s2) { + protected List collect(StateTransition... states) { List list = new ArrayList(); - list.add(s1); - list.add(s2); - return list; - } - private List collect(StateTransition s1, StateTransition s2, StateTransition s3) { - List list = collect(s1, s2); - list.add(s3); - return list; - } + for (StateTransition stateTransition : states) { + list.add(stateTransition); + } - private List collect(StateTransition s1, StateTransition s2, StateTransition s3, StateTransition s4) { - List list = collect(s1, s2, s3); - list.add(s4); return list; } @@ -228,7 +225,7 @@ public class SimpleFlowTests { * @author Dave Syer * */ - private static class StubState extends StateSupport { + protected static class StubState extends StateSupport { /** * @param string diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/StepContextFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/StepContextFactoryBeanTests.java new file mode 100644 index 000000000..1863b05e0 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/StepContextFactoryBeanTests.java @@ -0,0 +1,168 @@ +package org.springframework.batch.core.jsr; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.when; + +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Properties; +import java.util.Set; +import java.util.concurrent.Callable; +import java.util.concurrent.Future; + +import javax.batch.runtime.context.StepContext; + +import org.junit.Before; +import org.junit.Test; +import org.mockito.Mock; +import org.mockito.MockitoAnnotations; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.jsr.configuration.support.BatchPropertyContext; +import org.springframework.batch.core.scope.context.StepSynchronizationManager; +import org.springframework.beans.factory.FactoryBeanNotInitializedException; +import org.springframework.core.task.AsyncTaskExecutor; +import org.springframework.core.task.SimpleAsyncTaskExecutor; + +public class StepContextFactoryBeanTests { + + private StepContextFactoryBean factory; + @Mock + private BatchPropertyContext propertyContext; + + @Before + public void setUp() throws Exception { + MockitoAnnotations.initMocks(this); + factory = new StepContextFactoryBean(); + factory.setBatchPropertyContext(propertyContext); + } + + @Test(expected=FactoryBeanNotInitializedException.class) + public void testNoStepExecutionRegistered() throws Exception { + factory.getObject(); + } + + @Test + public void getObjectSingleThread() throws Exception { + StepSynchronizationManager.register(new StepExecution("step1", new JobExecution(5l), 3l)); + + StepContext context1 = factory.getObject(); + StepContext context2 = factory.getObject(); + + assertTrue(context1 == context2); + assertEquals(3l, context1.getStepExecutionId()); + + StepSynchronizationManager.close(); + + StepSynchronizationManager.register(new StepExecution("step2", new JobExecution(5l), 2l)); + + StepContext context3 = factory.getObject(); + StepContext context4 = factory.getObject(); + + assertTrue(context3 == context4); + assertTrue(context3 != context2); + assertEquals(2l, context3.getStepExecutionId()); + + StepSynchronizationManager.close(); + } + + @Test + public void getObjectSingleThreadWithProperties() throws Exception { + Properties props = new Properties(); + props.put("key1", "value1"); + + when(propertyContext.getStepProperties("step3")).thenReturn(props); + + StepSynchronizationManager.register(new StepExecution("step3", new JobExecution(5l), 3l)); + + StepContext context1 = factory.getObject(); + StepContext context2 = factory.getObject(); + + assertTrue(context1 == context2); + assertEquals(3l, context1.getStepExecutionId()); + assertEquals("value1", context1.getProperties().get("key1")); + + StepSynchronizationManager.close(); + } + + @Test + public void getObjectMultiThread() throws Exception { + List> stepContexts = new ArrayList>(); + + AsyncTaskExecutor executor = new SimpleAsyncTaskExecutor(); + + for(int i = 0; i < 4; i++) { + final long count = i; + stepContexts.add(executor.submit(new Callable() { + + @Override + public StepContext call() throws Exception { + try { + StepSynchronizationManager.register(new StepExecution("step" + count, new JobExecution(count))); + StepContext context = factory.getObject(); + Thread.sleep(1000l); + return context; + } catch (Throwable ignore) { + return null; + }finally { + StepSynchronizationManager.release(); + } + } + })); + } + + Set contexts = new HashSet(); + for (Future future : stepContexts) { + contexts.add(future.get()); + } + + assertEquals(4, contexts.size()); + } + + @Test + public void getObjectMultiThreadWithProperties() throws Exception { + for(int i = 0; i < 4; i++) { + Properties props = new Properties(); + props.put("step" + i, "step" + i + "value"); + + when(propertyContext.getStepProperties("step" + i)).thenReturn(props); + } + + List> stepContexts = new ArrayList>(); + + AsyncTaskExecutor executor = new SimpleAsyncTaskExecutor(); + + for(int i = 0; i < 4; i++) { + final long count = i; + stepContexts.add(executor.submit(new Callable() { + + @Override + public StepContext call() throws Exception { + try { + StepSynchronizationManager.register(new StepExecution("step" + count, new JobExecution(count))); + StepContext context = factory.getObject(); + Thread.sleep(1000l); + return context; + } catch (Throwable ignore) { + return null; + }finally { + StepSynchronizationManager.release(); + } + } + })); + } + + Set contexts = new HashSet(); + for (Future future : stepContexts) { + contexts.add(future.get()); + } + + assertEquals(4, contexts.size()); + + for (StepContext stepContext : contexts) { + assertEquals(stepContext.getStepName() + "value", stepContext.getProperties().get(stepContext.getStepName())); + } + } +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/StepContextTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/StepContextTests.java index 5873378b1..1b40bf8c4 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/StepContextTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/StepContextTests.java @@ -32,12 +32,15 @@ import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.JobParametersBuilder; import org.springframework.batch.core.StepExecution; import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.item.util.ExecutionContextUserSupport; +import org.springframework.util.ClassUtils; public class StepContextTests { private StepExecution stepExecution; private StepContext stepContext; private ExecutionContext executionContext; + private ExecutionContextUserSupport executionContextUserSupport = new ExecutionContextUserSupport(ClassUtils.getShortName(StepContext.class)); @Before public void setUp() throws Exception { @@ -68,6 +71,8 @@ public class StepContextTests { @Test public void testBasicProperties() { assertEquals(javax.batch.runtime.BatchStatus.STARTED, stepContext.getBatchStatus()); + assertEquals(null, stepContext.getExitStatus()); + stepContext.setExitStatus("customExitStatus"); assertEquals("customExitStatus", stepContext.getExitStatus()); assertEquals(5l, stepContext.getStepExecutionId()); assertEquals("testStep", stepContext.getStepName()); @@ -121,7 +126,7 @@ public class StepContextTests { String data = "saved data"; stepContext.setPersistentUserData(data); assertEquals(data, stepContext.getPersistentUserData()); - assertEquals(data, executionContext.get("batch_jsr_persistentUserData")); + assertEquals(data, executionContext.get(executionContextUserSupport.getKey("batch_jsr_persistentUserData"))); } @Test diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/StepExecutionTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/StepExecutionTests.java index 0a3a251a5..00055ca0e 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/StepExecutionTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/StepExecutionTests.java @@ -15,11 +15,15 @@ import org.springframework.batch.core.ExitStatus; import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.JobParametersBuilder; import org.springframework.batch.core.StepExecution; +import org.springframework.batch.item.util.ExecutionContextUserSupport; +import org.springframework.util.ClassUtils; public class StepExecutionTests { private StepExecution stepExecution; private javax.batch.runtime.StepExecution jsrStepExecution; + //The API that sets the persisted user data is on the StepContext so the key within the ExecutionContext is StepContext + private ExecutionContextUserSupport executionContextUserSupport = new ExecutionContextUserSupport(ClassUtils.getShortName(StepContext.class)); @Before public void setUp() throws Exception { @@ -39,7 +43,7 @@ public class StepExecutionTests { stepExecution.setWriteSkipCount(8); stepExecution.setStartTime(new Date(0)); stepExecution.setEndTime(new Date(10000000)); - stepExecution.getExecutionContext().put("batch_jsr_persistentUserData", "persisted data"); + stepExecution.getExecutionContext().put(executionContextUserSupport.getKey("batch_jsr_persistentUserData"), "persisted data"); jsrStepExecution = new org.springframework.batch.core.jsr.StepExecution(stepExecution); } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/configuration/xml/JobPropertyTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/configuration/xml/JobPropertyTests.java index 9542871d4..d05767962 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/configuration/xml/JobPropertyTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/configuration/xml/JobPropertyTests.java @@ -15,6 +15,8 @@ */ package org.springframework.batch.core.jsr.configuration.xml; +import static org.junit.Assert.assertEquals; + import java.io.Serializable; import java.util.List; @@ -38,7 +40,6 @@ import org.springframework.batch.core.launch.JobLauncher; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; -import static org.junit.Assert.assertEquals; /** *

diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/job/flow/support/DefaultFlowTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/job/flow/support/DefaultFlowTests.java new file mode 100644 index 000000000..f72326239 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/job/flow/support/DefaultFlowTests.java @@ -0,0 +1,74 @@ +/* + * Copyright 2013 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.jsr.job.flow.support; + +import static org.junit.Assert.assertEquals; + +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.StepExecution; +import org.springframework.batch.core.job.flow.FlowExecution; +import org.springframework.batch.core.job.flow.FlowExecutionStatus; +import org.springframework.batch.core.job.flow.State; +import org.springframework.batch.core.job.flow.StateSupport; +import org.springframework.batch.core.job.flow.support.JobFlowExecutorSupport; +import org.springframework.batch.core.job.flow.support.SimpleFlowTests; +import org.springframework.batch.core.job.flow.support.StateTransition; + +public class DefaultFlowTests extends SimpleFlowTests { + + @Override + @Before + public void setUp() { + flow = new DefaultFlow("flow1"); + } + + @Test + public void testNextBasedOnBatchStatus() throws Exception { + StepExecution stepExecution = new StepExecution("step1", new JobExecution(5l)); + stepExecution.setExitStatus(new ExitStatus("unmapped exit code")); + stepExecution.setStatus(BatchStatus.FAILED); + executor = new FlowExecutor(stepExecution); + + State startState = new StateSupport("step1", new FlowExecutionStatus("unmapped exit code")); + State endState = new StateSupport("failed", FlowExecutionStatus.FAILED); + + StateTransition failureTransition = StateTransition.createStateTransition(startState, "FAILED", "failed"); + StateTransition endTransition = StateTransition.createEndStateTransition(endState); + flow.setStateTransitions(collect(failureTransition, endTransition)); + flow.afterPropertiesSet(); + FlowExecution execution = flow.start(executor); + assertEquals(FlowExecutionStatus.FAILED, execution.getStatus()); + assertEquals("failed", execution.getName()); + } + + public static class FlowExecutor extends JobFlowExecutorSupport { + + private StepExecution stepExecution; + + public FlowExecutor(StepExecution stepExecution) { + this.stepExecution = stepExecution; + } + + @Override + public StepExecution getStepExecution() { + return stepExecution; + } + } +}