BATCH-2008: Completed implementaiton of JSR-352 JobInstance, JobExecution and StepExecution

* Spring Batch's native JobInstance now implements the JSR's JobInstance interface.
* SB's JobExecution can be wrapped by a JSR's JobExecution.
* SB's StepExecution can be wrappedy by a JSR's StepExecution.
* A new JsrFlowJob and JsrFlowExecutor have been introduced to correctly handle the promotion of ExitStatuses during step execution.
* The JsrJobOperator has been fixed to correctly retrieve StepExecutions (previously was not returning fully hydrated verions).

spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/CoreNamespacePostProcessor.java

@@ -15,6 +15,7 @@
  */
 package org.springframework.batch.core.configuration.xml;
 
+import org.springframework.batch.core.jsr.configuration.xml.JobFactoryBean;
 import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.core.step.AbstractStep;
 import org.springframework.beans.BeansException;
@@ -126,6 +127,12 @@ public class CoreNamespacePostProcessor implements BeanPostProcessor, BeanFactor
 			if (jobRepository == null) {
 				fb.setJobRepository((JobRepository) applicationContext.getBean(DEFAULT_JOB_REPOSITORY_NAME));
 			}
+		} else if(bean instanceof JobFactoryBean) {
+			JobFactoryBean fb = (JobFactoryBean) bean;
+			JobRepository jobRepository = fb.getJobRepository();
+			if (jobRepository == null) {
+				fb.setJobRepository((JobRepository) applicationContext.getBean(DEFAULT_JOB_REPOSITORY_NAME));
+			}
 		}
 		else if (bean instanceof StepParserStepFactoryBean) {
 			StepParserStepFactoryBean<?, ?> fb = (StepParserStepFactoryBean<?, ?>) bean;

spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowJob.java

@@ -33,13 +33,13 @@ import org.springframework.batch.core.step.StepLocator;
  * steps, rather than requiring sequential execution. In general, this job
  * implementation was designed to be used behind a parser, allowing for a
  * namespace to abstract away details.
- * 
+ *
  * @author Dave Syer
  * @since 2.0
  */
 public class FlowJob extends AbstractJob {
 
-	private Flow flow;
+	protected Flow flow;
 
 	private Map<String, Step> stepMap = new ConcurrentHashMap<String, Step>();
 
@@ -61,7 +61,7 @@ public class FlowJob extends AbstractJob {
 
 	/**
 	 * Public setter for the flow.
-	 * 
+	 *
 	 * @param flow the flow to set
 	 */
 	public void setFlow(Flow flow) {

spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/JobFlowExecutor.java

@@ -41,7 +41,7 @@ public class JobFlowExecutor implements FlowExecutor {
 
 	private final JobExecution execution;
 
-	private ExitStatus exitStatus = ExitStatus.EXECUTING;
+	protected ExitStatus exitStatus = ExitStatus.EXECUTING;
 
 	private final StepHandler stepHandler;
 
@@ -137,7 +137,7 @@ public class JobFlowExecutor implements FlowExecutor {
 	 * @param status
 	 * @return
 	 */
-	private BatchStatus findBatchStatus(FlowExecutionStatus status) {
+	protected BatchStatus findBatchStatus(FlowExecutionStatus status) {
 		for (BatchStatus batchStatus : BatchStatus.values()) {
 			if (status.getName().startsWith(batchStatus.toString())) {
 				return batchStatus;

spring-batch-core/src/main/java/org/springframework/batch/core/jsr/StepContext.java

@@ -72,6 +72,7 @@ public class StepContext implements javax.batch.runtime.context.StepContext {
 
 	@Override
 	public void setPersistentUserData(Serializable data) {
+		stepExecution.getExecutionContext().put("batch_jsr_persistentUserData", data);
 	}
 
 	@Override

spring-batch-core/src/main/java/org/springframework/batch/core/jsr/StepExecution.java

@@ -25,6 +25,9 @@ import org.springframework.batch.core.ExitStatus;
 import org.springframework.util.Assert;
 
 /**
+ * Implementation of the StepExecution as defined in JSR-352.  This implementation
+ * wraps a {@link org.springframework.batch.core.StepExecution} as it's source of
+ * data.
  *
  * @author Michael Minella
  * @since 3.0
@@ -33,37 +36,59 @@ public class StepExecution implements javax.batch.runtime.StepExecution{
 
 	private final org.springframework.batch.core.StepExecution stepExecution;
 
+	/**
+	 * @param stepExecution The {@link org.springframework.batch.core.StepExecution} used
+	 * as the basis for the data.
+	 */
 	public StepExecution(org.springframework.batch.core.StepExecution stepExecution) {
 		Assert.notNull(stepExecution, "A StepExecution is required");
 
 		this.stepExecution = stepExecution;
 	}
 
+	/* (non-Javadoc)
+	 * @see javax.batch.runtime.StepExecution#getStepExecutionId()
+	 */
 	@Override
 	public long getStepExecutionId() {
 		return stepExecution.getId();
 	}
 
+	/* (non-Javadoc)
+	 * @see javax.batch.runtime.StepExecution#getStepName()
+	 */
 	@Override
 	public String getStepName() {
 		return stepExecution.getStepName();
 	}
 
+	/* (non-Javadoc)
+	 * @see javax.batch.runtime.StepExecution#getBatchStatus()
+	 */
 	@Override
 	public BatchStatus getBatchStatus() {
 		return stepExecution.getStatus().getBatchStatus();
 	}
 
+	/* (non-Javadoc)
+	 * @see javax.batch.runtime.StepExecution#getStartTime()
+	 */
 	@Override
 	public Date getStartTime() {
 		return stepExecution.getStartTime();
 	}
 
+	/* (non-Javadoc)
+	 * @see javax.batch.runtime.StepExecution#getEndTime()
+	 */
 	@Override
 	public Date getEndTime() {
 		return stepExecution.getEndTime();
 	}
 
+	/* (non-Javadoc)
+	 * @see javax.batch.runtime.StepExecution#getExitStatus()
+	 */
 	@Override
 	public String getExitStatus() {
 		ExitStatus status = stepExecution.getExitStatus();
@@ -75,12 +100,17 @@ public class StepExecution implements javax.batch.runtime.StepExecution{
 		}
 	}
 
-	//TODO: Implement this
+	/* (non-Javadoc)
+	 * @see javax.batch.runtime.StepExecution#getPersistentUserData()
+	 */
 	@Override
 	public Serializable getPersistentUserData() {
-		return null;
+		return (Serializable) stepExecution.getExecutionContext().get("batch_jsr_persistentUserData");
 	}
 
+	/* (non-Javadoc)
+	 * @see javax.batch.runtime.StepExecution#getMetrics()
+	 */
 	@Override
 	public Metric[] getMetrics() {
 		Metric[] metrics = new Metric[8];

spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/JobFactoryBean.java

@@ -18,10 +18,17 @@ package org.springframework.batch.core.jsr.configuration.xml;
 import javax.batch.api.listener.JobListener;
 
 import org.springframework.batch.core.JobExecutionListener;
-import org.springframework.batch.core.configuration.xml.JobParserJobFactoryBean;
+import org.springframework.batch.core.JobParametersIncrementer;
+import org.springframework.batch.core.JobParametersValidator;
+import org.springframework.batch.core.job.flow.Flow;
 import org.springframework.batch.core.job.flow.FlowJob;
 import org.springframework.batch.core.jsr.JobListenerAdapter;
+import org.springframework.batch.core.jsr.job.flow.JsrFlowJob;
+import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.beans.factory.FactoryBean;
+import org.springframework.beans.factory.SmartFactoryBean;
+import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
 
 /**
  * This {@link FactoryBean} is used by the JSR-352 namespace parser to create
@@ -31,10 +38,101 @@ import org.springframework.beans.factory.FactoryBean;
  * @author Michael Minella
  * @since 3.0
  */
-public class JobFactoryBean extends JobParserJobFactoryBean {
+public class JobFactoryBean implements SmartFactoryBean<FlowJob> {
+
+	private String name;
+
+	private Boolean restartable;
+
+	private JobRepository jobRepository;
+
+	private JobParametersValidator jobParametersValidator;
+
+	private JobExecutionListener[] jobExecutionListeners;
+
+	private JobParametersIncrementer jobParametersIncrementer;
+
+	private Flow flow;
 
 	public JobFactoryBean(String name) {
-		super(name);
+		this.name = name;
+	}
+
+	@Override
+	public final FlowJob getObject() throws Exception {
+		Assert.isTrue(StringUtils.hasText(name), "The job must have an id.");
+		FlowJob flowJob = new JsrFlowJob(name);
+
+		if (restartable != null) {
+			flowJob.setRestartable(restartable);
+		}
+
+		if (jobRepository != null) {
+			flowJob.setJobRepository(jobRepository);
+		}
+
+		if (jobParametersValidator != null) {
+			flowJob.setJobParametersValidator(jobParametersValidator);
+		}
+
+		if (jobExecutionListeners != null) {
+			flowJob.setJobExecutionListeners(jobExecutionListeners);
+		}
+
+		if (jobParametersIncrementer != null) {
+			flowJob.setJobParametersIncrementer(jobParametersIncrementer);
+		}
+
+		if (flow != null) {
+			flowJob.setFlow(flow);
+		}
+
+		flowJob.afterPropertiesSet();
+		return flowJob;
+	}
+
+	public void setRestartable(Boolean restartable) {
+		this.restartable = restartable;
+	}
+
+	public void setJobRepository(JobRepository jobRepository) {
+		this.jobRepository = jobRepository;
+	}
+
+	public void setJobParametersValidator(JobParametersValidator jobParametersValidator) {
+		this.jobParametersValidator = jobParametersValidator;
+	}
+
+	public JobRepository getJobRepository() {
+		return this.jobRepository;
+	}
+
+	public void setJobParametersIncrementer(JobParametersIncrementer jobParametersIncrementer) {
+		this.jobParametersIncrementer = jobParametersIncrementer;
+	}
+
+	public void setFlow(Flow flow) {
+		this.flow = flow;
+	}
+
+	@Override
+	public Class<FlowJob> getObjectType() {
+		return FlowJob.class;
+	}
+
+	@Override
+	public boolean isSingleton() {
+		return true;
+	}
+
+	@Override
+	public boolean isEagerInit() {
+		return true;
+	}
+
+	@Override
+	public boolean isPrototype() {
+		return false;
 	}
 
 	/**
@@ -56,7 +154,7 @@ public class JobFactoryBean extends JobParserJobFactoryBean {
 				}
 			}
 
-			super.setJobExecutionListeners(listeners);
+			this.jobExecutionListeners = listeners;
 		}
 	}
 }

spring-batch-core/src/main/java/org/springframework/batch/core/jsr/job/flow/JsrFlowExecutor.java

@@ -0,0 +1,62 @@
+/*
+ * 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;
+
+import org.springframework.batch.core.ExitStatus;
+import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.core.job.StepHandler;
+import org.springframework.batch.core.job.flow.FlowExecutionStatus;
+import org.springframework.batch.core.job.flow.JobFlowExecutor;
+import org.springframework.batch.core.repository.JobRepository;
+
+/**
+ * JSR-352 specific {@link JobFlowExecutor}.  Unlike the regular {@link JobFlowExecutor},
+ * this extension does not promote an {@link ExitStatus} from a step to the job level if
+ * a custom exit status has been set on the job.
+ *
+ * @author Michael Minella
+ * @since 3.0
+ */
+public class JsrFlowExecutor extends JobFlowExecutor {
+
+	public JsrFlowExecutor(JobRepository jobRepository,
+			StepHandler stepHandler, JobExecution execution) {
+		super(jobRepository, stepHandler, execution);
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.batch.core.job.flow.JobFlowExecutor#updateJobExecutionStatus(org.springframework.batch.core.job.flow.FlowExecutionStatus)
+	 */
+	@Override
+	public void updateJobExecutionStatus(FlowExecutionStatus status) {
+		JobExecution execution = super.getJobExecution();
+
+		execution.setStatus(findBatchStatus(status));
+
+		ExitStatus curStatus = execution.getExitStatus();
+		if(curStatus == null ||
+				curStatus.getExitCode() == null ||
+				curStatus.getExitCode().equals(ExitStatus.COMPLETED.getExitCode()) ||
+				curStatus.getExitCode().equals(ExitStatus.EXECUTING.getExitCode()) ||
+				curStatus.getExitCode().equals(ExitStatus.FAILED.getExitCode()) ||
+				curStatus.getExitCode().equals(ExitStatus.NOOP.getExitCode()) ||
+				curStatus.getExitCode().equals(ExitStatus.STOPPED.getExitCode()) ||
+				curStatus.getExitCode().equals(ExitStatus.UNKNOWN.getExitCode())) {
+			exitStatus = exitStatus.and(new ExitStatus(status.getName()));
+			execution.setExitStatus(exitStatus);
+		}
+	}
+}

spring-batch-core/src/main/java/org/springframework/batch/core/jsr/job/flow/JsrFlowJob.java

@@ -0,0 +1,67 @@
+/*
+ * 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;
+
+import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.core.JobExecutionException;
+import org.springframework.batch.core.job.AbstractJob;
+import org.springframework.batch.core.job.SimpleStepHandler;
+import org.springframework.batch.core.job.flow.FlowExecutionException;
+import org.springframework.batch.core.job.flow.FlowJob;
+import org.springframework.batch.core.job.flow.JobFlowExecutor;
+
+/**
+ * JSR-352 specific extension of the {@link FlowJob}.
+ *
+ * @author Michael Minella
+ * @since 3.0
+ */
+public class JsrFlowJob extends FlowJob {
+
+	/**
+	 * No arg constructor (invalid state)
+	 */
+	public JsrFlowJob() {
+		super();
+	}
+
+	/**
+	 * Main constructor
+	 *
+	 * @param name of the flow
+	 */
+	public JsrFlowJob(String name) {
+		super(name);
+	}
+
+	/**
+	 * @see AbstractJob#doExecute(JobExecution)
+	 */
+	@Override
+	protected void doExecute(final JobExecution execution) throws JobExecutionException {
+		try {
+			JobFlowExecutor executor = new JsrFlowExecutor(getJobRepository(),
+					new SimpleStepHandler(getJobRepository()), execution);
+			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);
+		}
+	}
+}

spring-batch-core/src/main/java/org/springframework/batch/core/jsr/launch/JsrJobOperator.java

@@ -352,7 +352,7 @@ public class JsrJobOperator implements JobOperator {
 
 		if(executions != null) {
 			for (org.springframework.batch.core.StepExecution stepExecution : executions) {
-				batchExecutions.add(new org.springframework.batch.core.jsr.StepExecution(stepExecution));
+				batchExecutions.add(new org.springframework.batch.core.jsr.StepExecution(jobExplorer.getStepExecution(executionId, stepExecution.getId())));
 			}
 		}
 

spring-batch-core/src/main/resources/org/springframework/batch/core/schema-db2.sql

@@ -16,10 +16,10 @@ CREATE TABLE BATCH_JOB_EXECUTION  (
 	START_TIME TIMESTAMP DEFAULT NULL ,
 	END_TIME TIMESTAMP DEFAULT NULL ,
 	STATUS VARCHAR(10) ,
-	EXIT_CODE VARCHAR(100) ,
+	EXIT_CODE VARCHAR(2500) ,
 	EXIT_MESSAGE VARCHAR(2500) ,
 	LAST_UPDATED TIMESTAMP,
-	JOB_CONFIGURATION_LOCATION VARCHAR(500) NULL,
+	JOB_CONFIGURATION_LOCATION VARCHAR(2500) NULL,
 	constraint JOB_INST_EXEC_FK foreign key (JOB_INSTANCE_ID)
 	references BATCH_JOB_INSTANCE(JOB_INSTANCE_ID)
 ) ;
@@ -53,7 +53,7 @@ CREATE TABLE BATCH_STEP_EXECUTION  (
 	WRITE_SKIP_COUNT BIGINT ,
 	PROCESS_SKIP_COUNT BIGINT ,
 	ROLLBACK_COUNT BIGINT ,
-	EXIT_CODE VARCHAR(100) ,
+	EXIT_CODE VARCHAR(2500) ,
 	EXIT_MESSAGE VARCHAR(2500) ,
 	LAST_UPDATED TIMESTAMP,
 	constraint JOB_EXEC_STEP_FK foreign key (JOB_EXECUTION_ID)

spring-batch-core/src/main/resources/org/springframework/batch/core/schema-derby.sql

@@ -16,10 +16,10 @@ CREATE TABLE BATCH_JOB_EXECUTION  (
 	START_TIME TIMESTAMP DEFAULT NULL ,
 	END_TIME TIMESTAMP DEFAULT NULL ,
 	STATUS VARCHAR(10) ,
-	EXIT_CODE VARCHAR(100) ,
+	EXIT_CODE VARCHAR(2500) ,
 	EXIT_MESSAGE VARCHAR(2500) ,
 	LAST_UPDATED TIMESTAMP,
-	JOB_CONFIGURATION_LOCATION VARCHAR(500) NULL,
+	JOB_CONFIGURATION_LOCATION VARCHAR(2500) NULL,
 	constraint JOB_INST_EXEC_FK foreign key (JOB_INSTANCE_ID)
 	references BATCH_JOB_INSTANCE(JOB_INSTANCE_ID)
 ) ;
@@ -53,7 +53,7 @@ CREATE TABLE BATCH_STEP_EXECUTION  (
 	WRITE_SKIP_COUNT BIGINT ,
 	PROCESS_SKIP_COUNT BIGINT ,
 	ROLLBACK_COUNT BIGINT ,
-	EXIT_CODE VARCHAR(100) ,
+	EXIT_CODE VARCHAR(2500) ,
 	EXIT_MESSAGE VARCHAR(2500) ,
 	LAST_UPDATED TIMESTAMP,
 	constraint JOB_EXEC_STEP_FK foreign key (JOB_EXECUTION_ID)

spring-batch-core/src/main/resources/org/springframework/batch/core/schema-h2.sql

@@ -16,10 +16,10 @@ CREATE TABLE BATCH_JOB_EXECUTION  (
 	START_TIME TIMESTAMP DEFAULT NULL ,
 	END_TIME TIMESTAMP DEFAULT NULL ,
 	STATUS VARCHAR(10) ,
-	EXIT_CODE VARCHAR(100) ,
+	EXIT_CODE VARCHAR(2500) ,
 	EXIT_MESSAGE VARCHAR(2500) ,
 	LAST_UPDATED TIMESTAMP,
-	JOB_CONFIGURATION_LOCATION VARCHAR(500) NULL,
+	JOB_CONFIGURATION_LOCATION VARCHAR(2500) NULL,
 	constraint JOB_INST_EXEC_FK foreign key (JOB_INSTANCE_ID)
 	references BATCH_JOB_INSTANCE(JOB_INSTANCE_ID)
 ) ;
@@ -53,7 +53,7 @@ CREATE TABLE BATCH_STEP_EXECUTION  (
 	WRITE_SKIP_COUNT BIGINT ,
 	PROCESS_SKIP_COUNT BIGINT ,
 	ROLLBACK_COUNT BIGINT ,
-	EXIT_CODE VARCHAR(100) ,
+	EXIT_CODE VARCHAR(2500) ,
 	EXIT_MESSAGE VARCHAR(2500) ,
 	LAST_UPDATED TIMESTAMP,
 	constraint JOB_EXEC_STEP_FK foreign key (JOB_EXECUTION_ID)

spring-batch-core/src/main/resources/org/springframework/batch/core/schema-hsqldb.sql

@@ -16,10 +16,10 @@ CREATE TABLE BATCH_JOB_EXECUTION  (
 	START_TIME TIMESTAMP DEFAULT NULL ,
 	END_TIME TIMESTAMP DEFAULT NULL ,
 	STATUS VARCHAR(10) ,
-	EXIT_CODE VARCHAR(100) ,
+	EXIT_CODE VARCHAR(2500) ,
 	EXIT_MESSAGE VARCHAR(2500) ,
 	LAST_UPDATED TIMESTAMP,
-	JOB_CONFIGURATION_LOCATION VARCHAR(500) NULL,
+	JOB_CONFIGURATION_LOCATION VARCHAR(2500) NULL,
 	constraint JOB_INST_EXEC_FK foreign key (JOB_INSTANCE_ID)
 	references BATCH_JOB_INSTANCE(JOB_INSTANCE_ID)
 ) ;
@@ -53,7 +53,7 @@ CREATE TABLE BATCH_STEP_EXECUTION  (
 	WRITE_SKIP_COUNT BIGINT ,
 	PROCESS_SKIP_COUNT BIGINT ,
 	ROLLBACK_COUNT BIGINT ,
-	EXIT_CODE VARCHAR(100) ,
+	EXIT_CODE VARCHAR(2500) ,
 	EXIT_MESSAGE VARCHAR(2500) ,
 	LAST_UPDATED TIMESTAMP,
 	constraint JOB_EXEC_STEP_FK foreign key (JOB_EXECUTION_ID)

spring-batch-core/src/main/resources/org/springframework/batch/core/schema-mysql.sql

@@ -16,10 +16,10 @@ CREATE TABLE BATCH_JOB_EXECUTION  (
 	START_TIME DATETIME DEFAULT NULL ,
 	END_TIME DATETIME DEFAULT NULL ,
 	STATUS VARCHAR(10) ,
-	EXIT_CODE VARCHAR(100) ,
+	EXIT_CODE VARCHAR(2500) ,
 	EXIT_MESSAGE VARCHAR(2500) ,
 	LAST_UPDATED DATETIME,
-	JOB_CONFIGURATION_LOCATION VARCHAR(500) NULL,
+	JOB_CONFIGURATION_LOCATION VARCHAR(2500) NULL,
 	constraint JOB_INST_EXEC_FK foreign key (JOB_INSTANCE_ID)
 	references BATCH_JOB_INSTANCE(JOB_INSTANCE_ID)
 ) ENGINE=InnoDB;
@@ -53,7 +53,7 @@ CREATE TABLE BATCH_STEP_EXECUTION  (
 	WRITE_SKIP_COUNT BIGINT ,
 	PROCESS_SKIP_COUNT BIGINT ,
 	ROLLBACK_COUNT BIGINT ,
-	EXIT_CODE VARCHAR(100) ,
+	EXIT_CODE VARCHAR(2500) ,
 	EXIT_MESSAGE VARCHAR(2500) ,
 	LAST_UPDATED DATETIME,
 	constraint JOB_EXEC_STEP_FK foreign key (JOB_EXECUTION_ID)

spring-batch-core/src/main/resources/org/springframework/batch/core/schema-oracle10g.sql

@@ -16,10 +16,10 @@ CREATE TABLE BATCH_JOB_EXECUTION  (
 	START_TIME TIMESTAMP DEFAULT NULL ,
 	END_TIME TIMESTAMP DEFAULT NULL ,
 	STATUS VARCHAR2(10) ,
-	EXIT_CODE VARCHAR2(100) ,
+	EXIT_CODE VARCHAR2(2500) ,
 	EXIT_MESSAGE VARCHAR2(2500) ,
 	LAST_UPDATED TIMESTAMP,
-	JOB_CONFIGURATION_LOCATION VARCHAR(500) NULL,
+	JOB_CONFIGURATION_LOCATION VARCHAR(2500) NULL,
 	constraint JOB_INST_EXEC_FK foreign key (JOB_INSTANCE_ID)
 	references BATCH_JOB_INSTANCE(JOB_INSTANCE_ID)
 ) ;
@@ -53,7 +53,7 @@ CREATE TABLE BATCH_STEP_EXECUTION  (
 	WRITE_SKIP_COUNT NUMBER(19,0) ,
 	PROCESS_SKIP_COUNT NUMBER(19,0) ,
 	ROLLBACK_COUNT NUMBER(19,0) ,
-	EXIT_CODE VARCHAR2(100) ,
+	EXIT_CODE VARCHAR2(2500) ,
 	EXIT_MESSAGE VARCHAR2(2500) ,
 	LAST_UPDATED TIMESTAMP,
 	constraint JOB_EXEC_STEP_FK foreign key (JOB_EXECUTION_ID)

spring-batch-core/src/main/resources/org/springframework/batch/core/schema-postgresql.sql

@@ -16,10 +16,10 @@ CREATE TABLE BATCH_JOB_EXECUTION  (
 	START_TIME TIMESTAMP DEFAULT NULL ,
 	END_TIME TIMESTAMP DEFAULT NULL ,
 	STATUS VARCHAR(10) ,
-	EXIT_CODE VARCHAR(100) ,
+	EXIT_CODE VARCHAR(2500) ,
 	EXIT_MESSAGE VARCHAR(2500) ,
 	LAST_UPDATED TIMESTAMP,
-	JOB_CONFIGURATION_LOCATION VARCHAR(500) NULL,
+	JOB_CONFIGURATION_LOCATION VARCHAR(2500) NULL,
 	constraint JOB_INST_EXEC_FK foreign key (JOB_INSTANCE_ID)
 	references BATCH_JOB_INSTANCE(JOB_INSTANCE_ID)
 ) ;
@@ -53,7 +53,7 @@ CREATE TABLE BATCH_STEP_EXECUTION  (
 	WRITE_SKIP_COUNT BIGINT ,
 	PROCESS_SKIP_COUNT BIGINT ,
 	ROLLBACK_COUNT BIGINT ,
-	EXIT_CODE VARCHAR(100) ,
+	EXIT_CODE VARCHAR(2500) ,
 	EXIT_MESSAGE VARCHAR(2500) ,
 	LAST_UPDATED TIMESTAMP,
 	constraint JOB_EXEC_STEP_FK foreign key (JOB_EXECUTION_ID)

spring-batch-core/src/main/resources/org/springframework/batch/core/schema-sqlf.sql

@@ -16,10 +16,10 @@ CREATE TABLE BATCH_JOB_EXECUTION  (
 	START_TIME TIMESTAMP DEFAULT NULL ,
 	END_TIME TIMESTAMP DEFAULT NULL ,
 	STATUS VARCHAR(10) ,
-	EXIT_CODE VARCHAR(100) ,
+	EXIT_CODE VARCHAR(2500) ,
 	EXIT_MESSAGE VARCHAR(2500) ,
 	LAST_UPDATED TIMESTAMP,
-	JOB_CONFIGURATION_LOCATION VARCHAR(500) NULL,
+	JOB_CONFIGURATION_LOCATION VARCHAR(2500) NULL,
 	constraint JOB_INST_EXEC_FK foreign key (JOB_INSTANCE_ID)
 	references BATCH_JOB_INSTANCE(JOB_INSTANCE_ID)
 ) ;
@@ -53,7 +53,7 @@ CREATE TABLE BATCH_STEP_EXECUTION  (
 	WRITE_SKIP_COUNT BIGINT ,
 	PROCESS_SKIP_COUNT BIGINT ,
 	ROLLBACK_COUNT BIGINT ,
-	EXIT_CODE VARCHAR(100) ,
+	EXIT_CODE VARCHAR(2500) ,
 	EXIT_MESSAGE VARCHAR(2500) ,
 	LAST_UPDATED TIMESTAMP,
 	constraint JOB_EXEC_STEP_FK foreign key (JOB_EXECUTION_ID)

spring-batch-core/src/main/resources/org/springframework/batch/core/schema-sqlserver.sql

@@ -16,10 +16,10 @@ CREATE TABLE BATCH_JOB_EXECUTION  (
 	START_TIME DATETIME DEFAULT NULL ,
 	END_TIME DATETIME DEFAULT NULL ,
 	STATUS VARCHAR(10) ,
-	EXIT_CODE VARCHAR(100) ,
+	EXIT_CODE VARCHAR(2500) ,
 	EXIT_MESSAGE VARCHAR(2500) ,
 	LAST_UPDATED DATETIME,
-	JOB_CONFIGURATION_LOCATION VARCHAR(500) NULL,
+	JOB_CONFIGURATION_LOCATION VARCHAR(2500) NULL,
 	constraint JOB_INST_EXEC_FK foreign key (JOB_INSTANCE_ID)
 	references BATCH_JOB_INSTANCE(JOB_INSTANCE_ID)
 ) ;
@@ -53,7 +53,7 @@ CREATE TABLE BATCH_STEP_EXECUTION  (
 	WRITE_SKIP_COUNT BIGINT ,
 	PROCESS_SKIP_COUNT BIGINT ,
 	ROLLBACK_COUNT BIGINT ,
-	EXIT_CODE VARCHAR(100) ,
+	EXIT_CODE VARCHAR(2500) ,
 	EXIT_MESSAGE VARCHAR(2500) ,
 	LAST_UPDATED DATETIME,
 	constraint JOB_EXEC_STEP_FK foreign key (JOB_EXECUTION_ID)

spring-batch-core/src/main/resources/org/springframework/batch/core/schema-sybase.sql

@@ -16,10 +16,10 @@ CREATE TABLE BATCH_JOB_EXECUTION  (
 	START_TIME DATETIME DEFAULT NULL NULL,
 	END_TIME DATETIME DEFAULT NULL NULL,
 	STATUS VARCHAR(10) NULL,
-	EXIT_CODE VARCHAR(100) NULL,
+	EXIT_CODE VARCHAR(2500) NULL,
 	EXIT_MESSAGE VARCHAR(2500) NULL,
 	LAST_UPDATED DATETIME,
-	JOB_CONFIGURATION_LOCATION VARCHAR(500) NULL,
+	JOB_CONFIGURATION_LOCATION VARCHAR(2500) NULL,
 	constraint JOB_INST_EXEC_FK foreign key (JOB_INSTANCE_ID)
 	references BATCH_JOB_INSTANCE(JOB_INSTANCE_ID)
 ) ;
@@ -53,7 +53,7 @@ CREATE TABLE BATCH_STEP_EXECUTION  (
 	WRITE_SKIP_COUNT BIGINT NULL,
 	PROCESS_SKIP_COUNT BIGINT NULL,
 	ROLLBACK_COUNT BIGINT NULL,
-	EXIT_CODE VARCHAR(100) NULL,
+	EXIT_CODE VARCHAR(2500) NULL,
 	EXIT_MESSAGE VARCHAR(2500) NULL,
 	LAST_UPDATED DATETIME,
 	constraint JOB_EXEC_STEP_FK foreign key (JOB_EXECUTION_ID)

spring-batch-core/src/main/sql/schema.sql.vpp

@@ -1,33 +1,34 @@
 -- Autogenerated: do not edit this file
 
 CREATE TABLE BATCH_JOB_INSTANCE  (
-	JOB_INSTANCE_ID ${BIGINT} $!{IDENTITY} NOT NULL PRIMARY KEY $!{GENERATED},  
-	VERSION ${BIGINT} $!{NULL},  
-	JOB_NAME ${VARCHAR}(100) NOT NULL, 
+	JOB_INSTANCE_ID ${BIGINT} $!{IDENTITY} NOT NULL PRIMARY KEY $!{GENERATED},
+	VERSION ${BIGINT} $!{NULL},
+	JOB_NAME ${VARCHAR}(100) NOT NULL,
 	JOB_KEY ${VARCHAR}(32) NOT NULL,
 	constraint JOB_INST_UN unique (JOB_NAME, JOB_KEY)
 ) $!{VOODOO};
 
 CREATE TABLE BATCH_JOB_EXECUTION  (
 	JOB_EXECUTION_ID ${BIGINT} $!{IDENTITY} NOT NULL PRIMARY KEY $!{GENERATED},
-	VERSION ${BIGINT}  $!{NULL},  
+	VERSION ${BIGINT}  $!{NULL},
 	JOB_INSTANCE_ID ${BIGINT} NOT NULL,
 	CREATE_TIME ${TIMESTAMP} NOT NULL,
-	START_TIME ${TIMESTAMP} DEFAULT NULL $!{NULL}, 
+	START_TIME ${TIMESTAMP} DEFAULT NULL $!{NULL},
 	END_TIME ${TIMESTAMP} DEFAULT NULL $!{NULL},
 	STATUS ${VARCHAR}(10) $!{NULL},
-	EXIT_CODE ${VARCHAR}(100) $!{NULL},
+	EXIT_CODE ${VARCHAR}(2500) $!{NULL},
 	EXIT_MESSAGE ${VARCHAR}(2500) $!{NULL},
 	LAST_UPDATED ${TIMESTAMP},
+	JOB_CONFIGURATION_LOCATION ${VARCHAR}(2500) $!{NULL},
 	constraint JOB_INST_EXEC_FK foreign key (JOB_INSTANCE_ID)
 	references BATCH_JOB_INSTANCE(JOB_INSTANCE_ID)
 ) $!{VOODOO};
-	
+
 CREATE TABLE BATCH_JOB_EXECUTION_PARAMS  (
 	JOB_EXECUTION_ID ${BIGINT} NOT NULL ,
 	TYPE_CD ${VARCHAR}(6) NOT NULL ,
-	KEY_NAME ${VARCHAR}(100) NOT NULL , 
-	STRING_VAL ${VARCHAR}(250) $!{NULL}, 
+	KEY_NAME ${VARCHAR}(100) NOT NULL ,
+	STRING_VAL ${VARCHAR}(250) $!{NULL},
 	DATE_VAL ${TIMESTAMP} DEFAULT NULL $!{NULL},
 	LONG_VAL ${BIGINT} $!{NULL},
 	DOUBLE_VAL ${DOUBLE} $!{NULL},
@@ -35,24 +36,24 @@ CREATE TABLE BATCH_JOB_EXECUTION_PARAMS  (
 	constraint JOB_EXEC_PARAMS_FK foreign key (JOB_EXECUTION_ID)
 	references BATCH_JOB_EXECUTION(JOB_EXECUTION_ID)
 ) $!{VOODOO};
-	
+
 CREATE TABLE BATCH_STEP_EXECUTION  (
 	STEP_EXECUTION_ID ${BIGINT} $!{IDENTITY} NOT NULL PRIMARY KEY $!{GENERATED},
-	VERSION ${BIGINT} NOT NULL,  
+	VERSION ${BIGINT} NOT NULL,
 	STEP_NAME ${VARCHAR}(100) NOT NULL,
 	JOB_EXECUTION_ID ${BIGINT} NOT NULL,
-	START_TIME ${TIMESTAMP} NOT NULL , 
-	END_TIME ${TIMESTAMP} DEFAULT NULL $!{NULL},  
+	START_TIME ${TIMESTAMP} NOT NULL ,
+	END_TIME ${TIMESTAMP} DEFAULT NULL $!{NULL},
 	STATUS ${VARCHAR}(10) $!{NULL},
-	COMMIT_COUNT ${BIGINT} $!{NULL}, 
+	COMMIT_COUNT ${BIGINT} $!{NULL},
 	READ_COUNT ${BIGINT} $!{NULL},
 	FILTER_COUNT ${BIGINT} $!{NULL},
 	WRITE_COUNT ${BIGINT} $!{NULL},
 	READ_SKIP_COUNT ${BIGINT} $!{NULL},
 	WRITE_SKIP_COUNT ${BIGINT} $!{NULL},
 	PROCESS_SKIP_COUNT ${BIGINT} $!{NULL},
-	ROLLBACK_COUNT ${BIGINT} $!{NULL}, 
-	EXIT_CODE ${VARCHAR}(100) $!{NULL},
+	ROLLBACK_COUNT ${BIGINT} $!{NULL},
+	EXIT_CODE ${VARCHAR}(2500) $!{NULL},
 	EXIT_MESSAGE ${VARCHAR}(2500) $!{NULL},
 	LAST_UPDATED ${TIMESTAMP},
 	constraint JOB_EXEC_STEP_FK foreign key (JOB_EXECUTION_ID)
@@ -62,7 +63,7 @@ CREATE TABLE BATCH_STEP_EXECUTION  (
 CREATE TABLE BATCH_STEP_EXECUTION_CONTEXT  (
 	STEP_EXECUTION_ID ${BIGINT} NOT NULL PRIMARY KEY,
 	SHORT_CONTEXT ${VARCHAR}(2500) NOT NULL,
-	SERIALIZED_CONTEXT ${CLOB} $!{NULL}, 
+	SERIALIZED_CONTEXT ${CLOB} $!{NULL},
 	constraint STEP_EXEC_CTX_FK foreign key (STEP_EXECUTION_ID)
 	references BATCH_STEP_EXECUTION(STEP_EXECUTION_ID)
 ) $!{VOODOO};
@@ -70,7 +71,7 @@ CREATE TABLE BATCH_STEP_EXECUTION_CONTEXT  (
 CREATE TABLE BATCH_JOB_EXECUTION_CONTEXT  (
 	JOB_EXECUTION_ID ${BIGINT} NOT NULL PRIMARY KEY,
 	SHORT_CONTEXT ${VARCHAR}(2500) NOT NULL,
-	SERIALIZED_CONTEXT ${CLOB} $!{NULL}, 
+	SERIALIZED_CONTEXT ${CLOB} $!{NULL},
 	constraint JOB_EXEC_CTX_FK foreign key (JOB_EXECUTION_ID)
 	references BATCH_JOB_EXECUTION(JOB_EXECUTION_ID)
 ) $!{VOODOO};

spring-batch-core/src/test/java/org/springframework/batch/core/jsr/StepExecutionTests.java

@@ -39,6 +39,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");
 
 		jsrStepExecution = new org.springframework.batch.core.jsr.StepExecution(stepExecution);
 	}
@@ -63,6 +64,7 @@ public class StepExecutionTests {
 		assertEquals(new Date(0), jsrStepExecution.getStartTime());
 		assertEquals(new Date(10000000), jsrStepExecution.getEndTime());
 		assertEquals("customExitStatus", jsrStepExecution.getExitStatus());
+		assertEquals("persisted data", jsrStepExecution.getPersistentUserData());
 
 		Metric[] metrics = jsrStepExecution.getMetrics();
 

spring-batch-core/src/test/java/org/springframework/batch/core/jsr/job/flow/JsrFlowJobTests.java

@@ -0,0 +1,721 @@
+/*
+ * Copyright 2006-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;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.fail;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+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.JobInstance;
+import org.springframework.batch.core.JobInterruptedException;
+import org.springframework.batch.core.JobParameters;
+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.FlowExecutionStatus;
+import org.springframework.batch.core.job.flow.FlowExecutor;
+import org.springframework.batch.core.job.flow.JobExecutionDecider;
+import org.springframework.batch.core.job.flow.State;
+import org.springframework.batch.core.job.flow.StateSupport;
+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.DecisionState;
+import org.springframework.batch.core.job.flow.support.state.EndState;
+import org.springframework.batch.core.job.flow.support.state.FlowState;
+import org.springframework.batch.core.job.flow.support.state.SplitState;
+import org.springframework.batch.core.job.flow.support.state.StepState;
+import org.springframework.batch.core.repository.JobRepository;
+import org.springframework.batch.core.repository.dao.JobExecutionDao;
+import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean;
+import org.springframework.batch.core.step.StepSupport;
+
+/**
+ * @author Dave Syer
+ * @author Michael Minella
+ */
+public class JsrFlowJobTests {
+
+	private JsrFlowJob job = new JsrFlowJob();
+
+	private JobExecution jobExecution;
+
+	private JobRepository jobRepository;
+
+	private boolean fail = false;
+
+	private JobExecutionDao jobExecutionDao;
+
+	@Before
+	public void setUp() throws Exception {
+		MapJobRepositoryFactoryBean factory = new MapJobRepositoryFactoryBean();
+		factory.afterPropertiesSet();
+		jobExecutionDao = factory.getJobExecutionDao();
+		jobRepository = (JobRepository) factory.getObject();
+		job.setJobRepository(jobRepository);
+		jobExecution = jobRepository.createJobExecution("job", new JobParameters());
+	}
+
+	@Test
+	public void testGetSteps() throws Exception {
+		SimpleFlow flow = new SimpleFlow("job");
+		List<StateTransition> transitions = new ArrayList<StateTransition>();
+		transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("step1")), "step2"));
+		transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("step2")), "end0"));
+		transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.COMPLETED, "end0")));
+		flow.setStateTransitions(transitions);
+		flow.afterPropertiesSet();
+		job.setFlow(flow);
+		job.afterPropertiesSet();
+		assertEquals(2, job.getStepNames().size());
+	}
+
+	@Test
+	public void testTwoSteps() throws Exception {
+		SimpleFlow flow = new SimpleFlow("job");
+		List<StateTransition> transitions = new ArrayList<StateTransition>();
+		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);
+		job.setFlow(flow);
+		job.afterPropertiesSet();
+		job.doExecute(jobExecution);
+		StepExecution stepExecution = getStepExecution(jobExecution, "step2");
+		assertEquals(ExitStatus.COMPLETED, stepExecution.getExitStatus());
+		assertEquals(2, jobExecution.getStepExecutions().size());
+	}
+
+	@Test
+	public void testFailedStep() throws Exception {
+		SimpleFlow flow = new SimpleFlow("job");
+		List<StateTransition> transitions = new ArrayList<StateTransition>();
+		transitions.add(StateTransition.createStateTransition(new StateSupport("step1", FlowExecutionStatus.FAILED),
+				"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);
+		job.setFlow(flow);
+		job.afterPropertiesSet();
+		job.doExecute(jobExecution);
+		StepExecution stepExecution = getStepExecution(jobExecution, "step2");
+		assertEquals(ExitStatus.COMPLETED, stepExecution.getExitStatus());
+		assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus());
+		assertEquals(2, jobExecution.getStepExecutions().size());
+	}
+
+	@Test
+	public void testFailedStepRestarted() throws Exception {
+		SimpleFlow flow = new SimpleFlow("job");
+		List<StateTransition> transitions = new ArrayList<StateTransition>();
+		transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("step1")), "step2"));
+		State step2State = new StateSupport("step2") {
+			@Override
+			public FlowExecutionStatus handle(FlowExecutor executor) throws Exception {
+				JobExecution jobExecution = executor.getJobExecution();
+				jobExecution.createStepExecution(getName());
+				if (fail) {
+					return FlowExecutionStatus.FAILED;
+				}
+				else {
+					return FlowExecutionStatus.COMPLETED;
+				}
+			}
+		};
+		transitions.add(StateTransition.createStateTransition(step2State, ExitStatus.COMPLETED.getExitCode(), "end0"));
+		transitions.add(StateTransition.createStateTransition(step2State, ExitStatus.FAILED.getExitCode(), "end1"));
+		transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.COMPLETED, "end0")));
+		transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.FAILED, "end1")));
+		flow.setStateTransitions(transitions);
+		job.setFlow(flow);
+		job.afterPropertiesSet();
+		fail = true;
+		job.execute(jobExecution);
+		assertEquals(ExitStatus.FAILED, jobExecution.getExitStatus());
+		assertEquals(2, jobExecution.getStepExecutions().size());
+		jobRepository.update(jobExecution);
+		jobExecution = jobRepository.createJobExecution("job", new JobParameters());
+		fail = false;
+		job.execute(jobExecution);
+		assertEquals(ExitStatus.COMPLETED, jobExecution.getExitStatus());
+		assertEquals(1, jobExecution.getStepExecutions().size());
+	}
+
+	@Test
+	public void testStoppingStep() throws Exception {
+		SimpleFlow flow = new SimpleFlow("job");
+		List<StateTransition> transitions = new ArrayList<StateTransition>();
+		transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("step1")), "step2"));
+		State state2 = new StateSupport("step2", FlowExecutionStatus.FAILED);
+		transitions.add(StateTransition.createStateTransition(state2, ExitStatus.FAILED.getExitCode(), "end0"));
+		transitions.add(StateTransition.createStateTransition(state2, ExitStatus.COMPLETED.getExitCode(), "end1"));
+		transitions.add(StateTransition.createStateTransition(new EndState(FlowExecutionStatus.STOPPED, "end0"),
+				"step3"));
+		transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.COMPLETED, "end1")));
+		transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("step3")), "end2"));
+		transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.COMPLETED, "end2")));
+		flow.setStateTransitions(transitions);
+		job.setFlow(flow);
+		job.afterPropertiesSet();
+		job.doExecute(jobExecution);
+		assertEquals(2, jobExecution.getStepExecutions().size());
+		assertEquals(BatchStatus.STOPPED, jobExecution.getStatus());
+	}
+
+	@Test
+	public void testInterrupted() throws Exception {
+		SimpleFlow flow = new SimpleFlow("job");
+		List<StateTransition> transitions = new ArrayList<StateTransition>();
+		transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("step1") {
+			@Override
+			public void execute(StepExecution stepExecution) throws JobInterruptedException {
+				stepExecution.setStatus(BatchStatus.STOPPING);
+				jobRepository.update(stepExecution);
+			}
+		}), "end0"));
+		transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.COMPLETED, "end0")));
+		flow.setStateTransitions(transitions);
+		flow.afterPropertiesSet();
+		job.setFlow(flow);
+		job.afterPropertiesSet();
+		job.execute(jobExecution);
+		assertEquals(BatchStatus.STOPPED, jobExecution.getStatus());
+		checkRepository(BatchStatus.STOPPED, ExitStatus.STOPPED);
+		assertEquals(1, jobExecution.getAllFailureExceptions().size());
+		assertEquals(JobInterruptedException.class, jobExecution.getFailureExceptions().get(0).getClass());
+	}
+
+	@Test
+	public void testUnknownStatusStopsJob() throws Exception {
+		SimpleFlow flow = new SimpleFlow("job");
+		List<StateTransition> transitions = new ArrayList<StateTransition>();
+		transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("step1") {
+			@Override
+			public void execute(StepExecution stepExecution) throws JobInterruptedException {
+				stepExecution.setStatus(BatchStatus.UNKNOWN);
+				stepExecution.setTerminateOnly();
+				jobRepository.update(stepExecution);
+			}
+		}), "step2"));
+		transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("step2")), "end0"));
+		transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.COMPLETED, "end0")));
+		flow.setStateTransitions(transitions);
+		flow.afterPropertiesSet();
+		job.setFlow(flow);
+		job.afterPropertiesSet();
+		job.execute(jobExecution);
+		assertEquals(BatchStatus.UNKNOWN, jobExecution.getStatus());
+		checkRepository(BatchStatus.UNKNOWN, ExitStatus.STOPPED);
+		assertEquals(1, jobExecution.getAllFailureExceptions().size());
+		assertEquals(JobInterruptedException.class, jobExecution.getFailureExceptions().get(0).getClass());
+	}
+
+	@Test
+	public void testInterruptedSplit() throws Exception {
+		SimpleFlow flow = new SimpleFlow("job");
+		SimpleFlow flow1 = new SimpleFlow("flow1");
+		SimpleFlow flow2 = new SimpleFlow("flow2");
+
+		List<StateTransition> transitions = new ArrayList<StateTransition>();
+		transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("step1") {
+			@Override
+			public void execute(StepExecution stepExecution) throws JobInterruptedException {
+				if (!stepExecution.getJobExecution().getExecutionContext().containsKey("STOPPED")) {
+					stepExecution.getJobExecution().getExecutionContext().put("STOPPED", true);
+					stepExecution.setStatus(BatchStatus.STOPPED);
+					jobRepository.update(stepExecution);
+				}
+				else {
+					fail("The Job should have stopped by now");
+				}
+			}
+		}), "end0"));
+		transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.COMPLETED, "end0")));
+		flow1.setStateTransitions(new ArrayList<StateTransition>(transitions));
+		flow1.afterPropertiesSet();
+		flow2.setStateTransitions(new ArrayList<StateTransition>(transitions));
+		flow2.afterPropertiesSet();
+
+		transitions = new ArrayList<StateTransition>();
+		transitions.add(StateTransition.createStateTransition(new SplitState(Arrays.<Flow> asList(flow1, flow2),
+				"split"), "end0"));
+		transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.COMPLETED, "end0")));
+		flow.setStateTransitions(transitions);
+		flow.afterPropertiesSet();
+
+		job.setFlow(flow);
+		job.afterPropertiesSet();
+		job.execute(jobExecution);
+		assertEquals(BatchStatus.STOPPED, jobExecution.getStatus());
+		checkRepository(BatchStatus.STOPPED, ExitStatus.STOPPED);
+		assertEquals(1, jobExecution.getAllFailureExceptions().size());
+		assertEquals(JobInterruptedException.class, jobExecution.getFailureExceptions().get(0).getClass());
+		assertEquals(1, jobExecution.getStepExecutions().size());
+		for (StepExecution stepExecution : jobExecution.getStepExecutions()) {
+			assertEquals(BatchStatus.STOPPED, stepExecution.getStatus());
+		}
+	}
+
+	@Test
+	public void testInterruptedException() throws Exception {
+		SimpleFlow flow = new SimpleFlow("job");
+		List<StateTransition> transitions = new ArrayList<StateTransition>();
+		transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("step1") {
+			@Override
+			public void execute(StepExecution stepExecution) throws JobInterruptedException {
+				throw new JobInterruptedException("Stopped");
+			}
+		}), "end0"));
+		transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.COMPLETED, "end0")));
+		flow.setStateTransitions(transitions);
+		flow.afterPropertiesSet();
+		job.setFlow(flow);
+		job.afterPropertiesSet();
+		job.execute(jobExecution);
+		assertEquals(BatchStatus.STOPPED, jobExecution.getStatus());
+		checkRepository(BatchStatus.STOPPED, ExitStatus.STOPPED);
+		assertEquals(1, jobExecution.getAllFailureExceptions().size());
+		assertEquals(JobInterruptedException.class, jobExecution.getFailureExceptions().get(0).getClass());
+	}
+
+	@Test
+	public void testInterruptedSplitException() throws Exception {
+		SimpleFlow flow = new SimpleFlow("job");
+		SimpleFlow flow1 = new SimpleFlow("flow1");
+		SimpleFlow flow2 = new SimpleFlow("flow2");
+
+		List<StateTransition> transitions = new ArrayList<StateTransition>();
+		transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("step1") {
+			@Override
+			public void execute(StepExecution stepExecution) throws JobInterruptedException {
+				throw new JobInterruptedException("Stopped");
+			}
+		}), "end0"));
+		transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.COMPLETED, "end0")));
+		flow1.setStateTransitions(new ArrayList<StateTransition>(transitions));
+		flow1.afterPropertiesSet();
+		flow2.setStateTransitions(new ArrayList<StateTransition>(transitions));
+		flow2.afterPropertiesSet();
+
+		transitions = new ArrayList<StateTransition>();
+		transitions.add(StateTransition.createStateTransition(new SplitState(Arrays.<Flow> asList(flow1, flow2),
+				"split"), "end0"));
+		transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.COMPLETED, "end0")));
+		flow.setStateTransitions(transitions);
+		flow.afterPropertiesSet();
+
+		job.setFlow(flow);
+		job.afterPropertiesSet();
+		job.execute(jobExecution);
+		assertEquals(BatchStatus.STOPPED, jobExecution.getStatus());
+		checkRepository(BatchStatus.STOPPED, ExitStatus.STOPPED);
+		assertEquals(1, jobExecution.getAllFailureExceptions().size());
+		assertEquals(JobInterruptedException.class, jobExecution.getFailureExceptions().get(0).getClass());
+	}
+
+	@Test
+	public void testEndStateStopped() throws Exception {
+		SimpleFlow flow = new SimpleFlow("job");
+		List<StateTransition> transitions = new ArrayList<StateTransition>();
+		transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("step1")), "end"));
+		transitions.add(StateTransition
+				.createStateTransition(new EndState(FlowExecutionStatus.STOPPED, "end"), "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);
+		job.setFlow(flow);
+		job.afterPropertiesSet();
+		job.doExecute(jobExecution);
+		assertEquals(1, jobExecution.getStepExecutions().size());
+		assertEquals(BatchStatus.STOPPED, jobExecution.getStatus());
+	}
+
+	public void testEndStateFailed() throws Exception {
+		SimpleFlow flow = new SimpleFlow("job");
+		List<StateTransition> transitions = new ArrayList<StateTransition>();
+		transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("step1")), "end"));
+		transitions
+		.add(StateTransition.createStateTransition(new EndState(FlowExecutionStatus.FAILED, "end"), "step2"));
+		transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("step2")), ExitStatus.FAILED
+				.getExitCode(), "end0"));
+		transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("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);
+		job.setFlow(flow);
+		job.afterPropertiesSet();
+		job.doExecute(jobExecution);
+		assertEquals(BatchStatus.FAILED, jobExecution.getStatus());
+		assertEquals(1, jobExecution.getStepExecutions().size());
+	}
+
+	@Test
+	public void testEndStateStoppedWithRestart() throws Exception {
+		SimpleFlow flow = new SimpleFlow("job");
+		List<StateTransition> transitions = new ArrayList<StateTransition>();
+		transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("step1")), "end"));
+		transitions.add(StateTransition
+				.createStateTransition(new EndState(FlowExecutionStatus.STOPPED, "end"), "step2"));
+		StepState step2 = new StepState(new StubStep("step2"));
+		transitions.add(StateTransition.createStateTransition(step2, ExitStatus.COMPLETED.getExitCode(), "end0"));
+		transitions.add(StateTransition.createStateTransition(step2, ExitStatus.FAILED.getExitCode(), "end1"));
+		transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.COMPLETED, "end0")));
+		transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.FAILED, "end1")));
+		flow.setStateTransitions(transitions);
+		job.setFlow(flow);
+		job.afterPropertiesSet();
+
+		// To test a restart we have to use the AbstractJob.execute()...
+		job.execute(jobExecution);
+		assertEquals(BatchStatus.STOPPED, jobExecution.getStatus());
+		assertEquals(1, jobExecution.getStepExecutions().size());
+
+		jobExecution = jobRepository.createJobExecution("job", new JobParameters());
+		job.execute(jobExecution);
+		assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus());
+		assertEquals(1, jobExecution.getStepExecutions().size());
+
+	}
+
+	@Test
+	public void testBranching() throws Exception {
+		SimpleFlow flow = new SimpleFlow("job");
+		List<StateTransition> transitions = new ArrayList<StateTransition>();
+		StepState step1 = new StepState(new StubStep("step1"));
+		transitions.add(StateTransition.createStateTransition(step1, "step2"));
+		transitions.add(StateTransition.createStateTransition(step1, "COMPLETED", "step3"));
+		StepState step2 = new StepState(new StubStep("step2"));
+		transitions.add(StateTransition.createStateTransition(step2, ExitStatus.COMPLETED.getExitCode(), "end0"));
+		transitions.add(StateTransition.createStateTransition(step2, ExitStatus.FAILED.getExitCode(), "end1"));
+		transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.COMPLETED, "end0")));
+		transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.FAILED, "end1")));
+		StepState step3 = new StepState(new StubStep("step3"));
+		transitions.add(StateTransition.createStateTransition(step3, ExitStatus.FAILED.getExitCode(), "end2"));
+		transitions.add(StateTransition.createStateTransition(step3, ExitStatus.COMPLETED.getExitCode(), "end3"));
+		transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.FAILED, "end2")));
+		transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.COMPLETED, "end3")));
+		flow.setStateTransitions(transitions);
+		job.setFlow(flow);
+		job.afterPropertiesSet();
+		job.doExecute(jobExecution);
+		StepExecution stepExecution = getStepExecution(jobExecution, "step3");
+		assertEquals(ExitStatus.COMPLETED, stepExecution.getExitStatus());
+		assertEquals(2, jobExecution.getStepExecutions().size());
+	}
+
+	@Test
+	public void testBasicFlow() throws Throwable {
+		SimpleFlow flow = new SimpleFlow("job");
+		List<StateTransition> transitions = new ArrayList<StateTransition>();
+		transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("step")), "end0"));
+		transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.COMPLETED, "end0")));
+		flow.setStateTransitions(transitions);
+		job.setFlow(flow);
+		job.execute(jobExecution);
+		if (!jobExecution.getAllFailureExceptions().isEmpty()) {
+			throw jobExecution.getAllFailureExceptions().get(0);
+		}
+		assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus());
+	}
+
+	@Test
+	public void testDecisionFlow() throws Throwable {
+
+		SimpleFlow flow = new SimpleFlow("job");
+		JobExecutionDecider decider = new JobExecutionDecider() {
+			@Override
+			public FlowExecutionStatus decide(JobExecution jobExecution, StepExecution stepExecution) {
+				assertNotNull(stepExecution);
+				return new FlowExecutionStatus("SWITCH");
+			}
+		};
+
+		List<StateTransition> transitions = new ArrayList<StateTransition>();
+		transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("step1")), "decision"));
+		DecisionState decision = new DecisionState(decider, "decision");
+		transitions.add(StateTransition.createStateTransition(decision, "step2"));
+		transitions.add(StateTransition.createStateTransition(decision, "SWITCH", "step3"));
+		StepState step2 = new StepState(new StubStep("step2"));
+		transitions.add(StateTransition.createStateTransition(step2, ExitStatus.COMPLETED.getExitCode(), "end0"));
+		transitions.add(StateTransition.createStateTransition(step2, ExitStatus.FAILED.getExitCode(), "end1"));
+		transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.COMPLETED, "end0")));
+		transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.FAILED, "end1")));
+		StepState step3 = new StepState(new StubStep("step3"));
+		transitions.add(StateTransition.createStateTransition(step3, ExitStatus.FAILED.getExitCode(), "end2"));
+		transitions.add(StateTransition.createStateTransition(step3, ExitStatus.COMPLETED.getExitCode(), "end3"));
+		transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.FAILED, "end2")));
+		transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.COMPLETED, "end3")));
+		flow.setStateTransitions(transitions);
+
+		job.setFlow(flow);
+		job.doExecute(jobExecution);
+		StepExecution stepExecution = getStepExecution(jobExecution, "step3");
+		if (!jobExecution.getAllFailureExceptions().isEmpty()) {
+			throw jobExecution.getAllFailureExceptions().get(0);
+		}
+
+		assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus());
+		assertEquals(2, jobExecution.getStepExecutions().size());
+
+	}
+
+	@Test
+	public void testDecisionFlowWithExceptionInDecider() throws Throwable {
+
+		SimpleFlow flow = new SimpleFlow("job");
+		JobExecutionDecider decider = new JobExecutionDecider() {
+			@Override
+			public FlowExecutionStatus decide(JobExecution jobExecution, StepExecution stepExecution) {
+				assertNotNull(stepExecution);
+				throw new RuntimeException("Foo");
+			}
+		};
+
+		List<StateTransition> transitions = new ArrayList<StateTransition>();
+		transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("step1")), "decision"));
+		DecisionState decision = new DecisionState(decider, "decision");
+		transitions.add(StateTransition.createStateTransition(decision, "step2"));
+		transitions.add(StateTransition.createStateTransition(decision, "SWITCH", "step3"));
+		StepState step2 = new StepState(new StubStep("step2"));
+		transitions.add(StateTransition.createStateTransition(step2, ExitStatus.COMPLETED.getExitCode(), "end0"));
+		transitions.add(StateTransition.createStateTransition(step2, ExitStatus.FAILED.getExitCode(), "end1"));
+		transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.COMPLETED, "end0")));
+		transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.FAILED, "end1")));
+		StepState step3 = new StepState(new StubStep("step3"));
+		transitions.add(StateTransition.createStateTransition(step3, ExitStatus.FAILED.getExitCode(), "end2"));
+		transitions.add(StateTransition.createStateTransition(step3, ExitStatus.COMPLETED.getExitCode(), "end3"));
+		transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.FAILED, "end2")));
+		transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.COMPLETED, "end3")));
+		flow.setStateTransitions(transitions);
+
+		job.setFlow(flow);
+		try {
+			job.execute(jobExecution);
+		}
+		finally {
+
+			assertEquals(BatchStatus.FAILED, jobExecution.getStatus());
+			assertEquals(1, jobExecution.getStepExecutions().size());
+
+			assertEquals(1, jobExecution.getAllFailureExceptions().size());
+			assertEquals("Foo", jobExecution.getAllFailureExceptions().get(0).getCause().getCause().getMessage());
+
+		}
+	}
+
+	@Test
+	public void testGetStepExists() throws Exception {
+		SimpleFlow flow = new SimpleFlow("job");
+		List<StateTransition> transitions = new ArrayList<StateTransition>();
+		transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("step1")), "step2"));
+		transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("step2")), "end0"));
+		transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.COMPLETED, "end0")));
+		flow.setStateTransitions(transitions);
+		flow.afterPropertiesSet();
+		job.setFlow(flow);
+		job.afterPropertiesSet();
+
+		Step step = job.getStep("step2");
+		assertNotNull(step);
+		assertEquals("step2", step.getName());
+	}
+
+	@Test
+	public void testGetStepExistsWithPrefix() throws Exception {
+		SimpleFlow flow = new SimpleFlow("job");
+		List<StateTransition> transitions = new ArrayList<StateTransition>();
+		transitions.add(StateTransition.createStateTransition(new StepState("job.step", new StubStep("step")), "end0"));
+		transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.COMPLETED, "end0")));
+		flow.setStateTransitions(transitions);
+		flow.afterPropertiesSet();
+		job.setFlow(flow);
+		job.setName(flow.getName());
+		job.afterPropertiesSet();
+
+		Step step = job.getStep("step");
+		assertNotNull(step);
+		assertEquals("step", step.getName());
+	}
+
+	@Test
+	public void testGetStepNamesWithPrefix() throws Exception {
+		SimpleFlow flow = new SimpleFlow("job");
+		List<StateTransition> transitions = new ArrayList<StateTransition>();
+		transitions.add(StateTransition.createStateTransition(new StepState("job.step", new StubStep("step")), "end0"));
+		transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.COMPLETED, "end0")));
+		flow.setStateTransitions(transitions);
+		flow.afterPropertiesSet();
+		job.setFlow(flow);
+		job.setName(flow.getName());
+		job.afterPropertiesSet();
+
+		assertEquals("[step]", job.getStepNames().toString());
+	}
+
+	@Test
+	public void testGetStepNotExists() throws Exception {
+		SimpleFlow flow = new SimpleFlow("job");
+		List<StateTransition> transitions = new ArrayList<StateTransition>();
+		transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("step1")), "step2"));
+		transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("step2")), "end0"));
+		transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.COMPLETED, "end0")));
+		flow.setStateTransitions(transitions);
+		flow.afterPropertiesSet();
+		job.setFlow(flow);
+		job.afterPropertiesSet();
+
+		Step step = job.getStep("foo");
+		assertNull(step);
+	}
+
+	@Test
+	public void testGetStepNotStepState() throws Exception {
+		SimpleFlow flow = new SimpleFlow("job");
+		List<StateTransition> transitions = new ArrayList<StateTransition>();
+		transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("step1")), "step2"));
+		transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("step2")), "end0"));
+		transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.COMPLETED, "end0")));
+		flow.setStateTransitions(transitions);
+		flow.afterPropertiesSet();
+		job.setFlow(flow);
+		job.afterPropertiesSet();
+
+		Step step = job.getStep("end0");
+		assertNull(step);
+	}
+
+	@Test
+	public void testGetStepNestedFlow() throws Exception {
+		SimpleFlow nested = new SimpleFlow("nested");
+		List<StateTransition> transitions = new ArrayList<StateTransition>();
+		transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("step2")), "end1"));
+		transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.COMPLETED, "end1")));
+		nested.setStateTransitions(transitions);
+		nested.afterPropertiesSet();
+
+		SimpleFlow flow = new SimpleFlow("job");
+		transitions = new ArrayList<StateTransition>();
+		transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("step1")), "nested"));
+		transitions.add(StateTransition.createStateTransition(new FlowState(nested, "nested"), "end0"));
+		transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.COMPLETED, "end0")));
+		flow.setStateTransitions(transitions);
+		flow.afterPropertiesSet();
+		job.setFlow(flow);
+		job.afterPropertiesSet();
+
+		List<String> names = new ArrayList<String>(job.getStepNames());
+		Collections.sort(names);
+		assertEquals("[step1, step2]", names.toString());
+	}
+
+	@Test
+	public void testGetStepSplitFlow() throws Exception {
+		SimpleFlow flow = new SimpleFlow("job");
+		SimpleFlow flow1 = new SimpleFlow("flow1");
+		SimpleFlow flow2 = new SimpleFlow("flow2");
+
+		List<StateTransition> transitions = new ArrayList<StateTransition>();
+		transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("step1")), "end0"));
+		transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.COMPLETED, "end0")));
+		flow1.setStateTransitions(new ArrayList<StateTransition>(transitions));
+		flow1.afterPropertiesSet();
+		transitions = new ArrayList<StateTransition>();
+		transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("step2")), "end1"));
+		transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.COMPLETED, "end1")));
+		flow2.setStateTransitions(new ArrayList<StateTransition>(transitions));
+		flow2.afterPropertiesSet();
+
+		transitions = new ArrayList<StateTransition>();
+		transitions.add(StateTransition.createStateTransition(new SplitState(Arrays.<Flow> asList(flow1, flow2),
+				"split"), "end2"));
+		transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.COMPLETED, "end2")));
+		flow.setStateTransitions(transitions);
+		flow.afterPropertiesSet();
+
+		job.setFlow(flow);
+		job.afterPropertiesSet();
+		List<String> names = new ArrayList<String>(job.getStepNames());
+		Collections.sort(names);
+		assertEquals("[step1, step2]", names.toString());
+	}
+
+	/**
+	/**
+	 * @author Dave Syer
+	 *
+	 */
+	private class StubStep extends StepSupport {
+
+		private StubStep(String name) {
+			super(name);
+		}
+
+		@Override
+		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;
+	}
+
+	private void checkRepository(BatchStatus status, ExitStatus exitStatus) {
+		// because map dao stores in memory, it can be checked directly
+		JobInstance jobInstance = jobExecution.getJobInstance();
+		JobExecution other = jobExecutionDao.findJobExecutions(jobInstance).get(0);
+		assertEquals(jobInstance.getId(), other.getJobId());
+		assertEquals(status, other.getStatus());
+		if (exitStatus != null) {
+			assertEquals(exitStatus.getExitCode(), other.getExitStatus().getExitCode());
+		}
+	}
+
+}

spring-batch-core/src/test/java/org/springframework/batch/core/jsr/launch/JsrJobOperatorTests.java

@@ -273,11 +273,13 @@ public class JsrJobOperatorTests {
 	public void testGetStepExecutionsRoseyScenario() {
 		JobExecution jobExecution = new JobExecution(5l);
 		List<StepExecution> stepExecutions = new ArrayList<StepExecution>();
-		stepExecutions.add(new StepExecution("step1", jobExecution));
-		stepExecutions.add(new StepExecution("step2", jobExecution));
+		stepExecutions.add(new StepExecution("step1", jobExecution, 1l));
+		stepExecutions.add(new StepExecution("step2", jobExecution, 2l));
 		jobExecution.addStepExecutions(stepExecutions);
 
 		when(jobExplorer.getJobExecution(5l)).thenReturn(jobExecution);
+		when(jobExplorer.getStepExecution(5l, 1l)).thenReturn(new StepExecution("step1", jobExecution, 1l));
+		when(jobExplorer.getStepExecution(5l, 2l)).thenReturn(new StepExecution("step2", jobExecution, 2l));
 
 		List<javax.batch.runtime.StepExecution> results = jsrJobOperator.getStepExecutions(5l);
 

spring-batch-core/src/test/resources/org/springframework/batch/core/repository/dao/schema-prefix-hsqldb.sql

@@ -28,7 +28,7 @@ CREATE TABLE PREFIX_JOB_EXECUTION  (
 	EXIT_CODE VARCHAR(20) ,
 	EXIT_MESSAGE VARCHAR(2500) ,
 	LAST_UPDATED TIMESTAMP,
-	JOB_CONFIGURATION_LOCATION VARCHAR(500) NULL,
+	JOB_CONFIGURATION_LOCATION VARCHAR(2500) NULL,
 	constraint PREFIX_JOB_INST_EXEC_FK foreign key (JOB_INSTANCE_ID)
 	references PREFIX_JOB_INSTANCE(JOB_INSTANCE_ID)
 ) ;