Added the implementation of a post processor to lazily init partition

related artifacts
* Added BeanFactoryPostProcessor
* Fixed StepContext creation (now creates one per thread)
* Fixed StepContext exit status processing (now keeps track of being set
 manualy or not
* Created new Flow implementation to handle JSR-352 state transitions
* Updated flow parsing to not default to completed if no other
 transitions are provided

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<StateTransition> stateTransitionComparator;
 
+	private Class<SimpleFlow> flowType;
+
+	/**
+	 * @param stateTransitionComparator {@link Comparator} implementation that addresses
+	 * the ordering of state evaluation
+	 */
 	public void setStateTransitionComparator(Comparator<StateTransition> stateTransitionComparator) {
 		this.stateTransitionComparator = stateTransitionComparator;
 	}
 
+	/**
+	 * @param flowType Used to inject the type of flow (regular Spring Batch or JSR-352)
+	 */
+	public void setFlowType(Class<SimpleFlow> 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);
 

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<String, Set<StateTransition>> 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<String, State> 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<StateTransition> 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.

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);
 	}
 
 	/**

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<StepContext> {
+public class StepContextFactoryBean implements FactoryBean<StepContext>, InitializingBean {
 	@Autowired
 	private BatchPropertyContext batchPropertyContext;
 
+	private static final ThreadLocal<javax.batch.runtime.context.StepContext> contextHolder = new ThreadLocal<javax.batch.runtime.context.StepContext>();
+
+	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<StepContext> {
 	public boolean isSingleton() {
 		return false;
 	}
+
+	@Override
+	public void afterPropertiesSet() throws Exception {
+		Assert.notNull(batchPropertyContext, "BatchPropertyContext is required");
+	}
 }

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)

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);
+					}
+				}
+			}
+		}
+	}
+}

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<BeanDefinition> stateTransitions = new ArrayList<BeanDefinition>();
 
@@ -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;
 	}

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) {

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<StateTransition> 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);
+	}
+}

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<StepExecution> 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());
+		}
+	}
+}

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<StepExecution> 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");

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);

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<StateTransition> collect(StateTransition s1, StateTransition s2) {
+	protected List<StateTransition> collect(StateTransition... states) {
 		List<StateTransition> list = new ArrayList<StateTransition>();
-		list.add(s1);
-		list.add(s2);
-		return list;
-	}
 
-	private List<StateTransition> collect(StateTransition s1, StateTransition s2, StateTransition s3) {
-		List<StateTransition> list = collect(s1, s2);
-		list.add(s3);
-		return list;
-	}
+		for (StateTransition stateTransition : states) {
+			list.add(stateTransition);
+		}
 
-	private List<StateTransition> collect(StateTransition s1, StateTransition s2, StateTransition s3, StateTransition s4) {
-		List<StateTransition> 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

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<Future<StepContext>> stepContexts = new ArrayList<Future<StepContext>>();
+
+		AsyncTaskExecutor executor = new SimpleAsyncTaskExecutor();
+
+		for(int i = 0; i < 4; i++) {
+			final long count = i;
+			stepContexts.add(executor.submit(new Callable<StepContext>() {
+
+				@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<StepContext> contexts = new HashSet<StepContext>();
+		for (Future<StepContext> 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<Future<StepContext>> stepContexts = new ArrayList<Future<StepContext>>();
+
+		AsyncTaskExecutor executor = new SimpleAsyncTaskExecutor();
+
+		for(int i = 0; i < 4; i++) {
+			final long count = i;
+			stepContexts.add(executor.submit(new Callable<StepContext>() {
+
+				@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<StepContext> contexts = new HashSet<StepContext>();
+		for (Future<StepContext> future : stepContexts) {
+			contexts.add(future.get());
+		}
+
+		assertEquals(4, contexts.size());
+
+		for (StepContext stepContext : contexts) {
+			assertEquals(stepContext.getStepName() + "value", stepContext.getProperties().get(stepContext.getStepName()));
+		}
+	}
+}

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

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);
 	}

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;
 
 /**
  * <p>

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;
+		}
+	}
+}