BATCH-1871: Add flow and job builders

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

@@ -15,6 +15,15 @@
  */
 package org.springframework.batch.core.configuration.xml;
 
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
 import org.springframework.batch.core.job.flow.FlowExecutionStatus;
 import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.beans.factory.parsing.CompositeComponentDefinition;
@@ -28,8 +37,6 @@ import org.w3c.dom.Element;
 import org.w3c.dom.Node;
 import org.w3c.dom.NodeList;
 
-import java.util.*;
-
 /**
  * @author Dave Syer
  * 
@@ -182,14 +189,12 @@ public abstract class AbstractFlowParser extends AbstractSingleBeanDefinitionPar
 			reachableElements.add(nextAttribute);
 		}
 
-		@SuppressWarnings("unchecked")
 		List<Element> nextElements = DomUtils.getChildElementsByTagName(element, NEXT_ELE);
 		for (Element nextElement : nextElements) {
 			String toAttribute = nextElement.getAttribute(TO_ATTR);
 			reachableElements.add(toAttribute);
 		}
 
-		@SuppressWarnings("unchecked")
 		List<Element> stopElements = DomUtils.getChildElementsByTagName(element, STOP_ELE);
 		for (Element stopElement : stopElements) {
 			String restartAttribute = stopElement.getAttribute(RESTART_ATTR);
@@ -257,7 +262,6 @@ public abstract class AbstractFlowParser extends AbstractSingleBeanDefinitionPar
 		boolean transitionElementExists = false;
 		List<String> patterns = new ArrayList<String>();
 		for (String transitionName : new String[] { NEXT_ELE, STOP_ELE, END_ELE, FAIL_ELE }) {
-			@SuppressWarnings("unchecked")
 			List<Element> transitionElements = DomUtils.getChildElementsByTagName(element, transitionName);
 			for (Element transitionElement : transitionElements) {
 				verifyUniquePattern(transitionElement, patterns, element, parserContext);

spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/FlowBuilder.java

@@ -0,0 +1,557 @@
+/*
+ * Copyright 2012-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.job.builder;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.springframework.batch.core.ExitStatus;
+import org.springframework.batch.core.Step;
+import org.springframework.batch.core.job.flow.Flow;
+import org.springframework.batch.core.job.flow.FlowExecutionStatus;
+import org.springframework.batch.core.job.flow.JobExecutionDecider;
+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;
+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.core.task.TaskExecutor;
+
+/**
+ * A builder for a flow of steps that can be executed as a job or as part of a job. Steps can be linked together with
+ * conditional transitions that depend on the exit status of the previous step.
+ * 
+ * @author Dave Syer
+ * 
+ * @since 2.2
+ * 
+ * @param <Q> the type of object returned by the builder (by default a Flow)
+ * 
+ */
+public class FlowBuilder<Q> {
+
+	private String name;
+
+	private String prefix;
+
+	private List<StateTransition> transitions = new ArrayList<StateTransition>();
+
+	private Map<String, State> tos = new HashMap<String, State>();
+
+	private State currentState;
+
+	private EndState failedState;
+
+	private EndState completedState;
+
+	private int decisionCounter = 0;
+
+	private int splitCounter = 0;
+
+	private Map<Object, State> states = new HashMap<Object, State>();
+
+	private SimpleFlow flow;
+
+	private boolean dirty = true;
+
+	public FlowBuilder(String name) {
+		this.name = name;
+		this.prefix = name + ".";
+		this.failedState = new EndState(FlowExecutionStatus.FAILED, prefix + "FAILED");
+		this.completedState = new EndState(FlowExecutionStatus.COMPLETED, prefix + "COMPLETED");
+	}
+
+	/**
+	 * Validate the current state of the builder and build a flow. Subclasses may override this to build an object of a
+	 * different type that itself depends on the flow.
+	 * 
+	 * @return a flow
+	 */
+	public Q build() {
+		@SuppressWarnings("unchecked")
+		Q result = (Q) flow();
+		return result;
+	}
+
+	/**
+	 * Transition to the next step on successful completion of the current step. All other outcomes are treated as
+	 * failures.
+	 * 
+	 * @param step the next step
+	 * @return this to enable chaining
+	 */
+	public FlowBuilder<Q> next(Step step) {
+		doNext(step);
+		return this;
+	}
+
+	/**
+	 * Start a flow. If some steps are already registered, just a synonym for {@link #from(Step)}.
+	 * 
+	 * @param step the step to start with
+	 * @return this to enable chaining
+	 */
+	public FlowBuilder<Q> start(Step step) {
+		doStart(step);
+		return this;
+	}
+
+	/**
+	 * Go back to a previously registered step and start a new path. If no steps are registered yet just a synonym for
+	 * {@link #start(Step)}.
+	 * 
+	 * @param step the step to start from (already registered)
+	 * @return this to enable chaining
+	 */
+	public FlowBuilder<Q> from(Step step) {
+		doFrom(step);
+		return this;
+	}
+
+	/**
+	 * Transition to the decider on successful completion of the current step. All other outcomes are treated as
+	 * failures.
+	 * 
+	 * @param step the next step
+	 * @return this to enable chaining
+	 */
+	public UnterminatedFlowBuilder<Q> next(JobExecutionDecider decider) {
+		doNext(decider);
+		return new UnterminatedFlowBuilder<Q>(this);
+	}
+
+	/**
+	 * If a flow should start with a decision use this as the first state.
+	 * 
+	 * @param decider the to start from
+	 * @return a builder to enable chaining
+	 */
+	public UnterminatedFlowBuilder<Q> start(JobExecutionDecider decider) {
+		doStart(decider);
+		return new UnterminatedFlowBuilder<Q>(this);
+	}
+
+	/**
+	 * Start again from a decision that was already registered.
+	 * 
+	 * @param decider the decider to start from (already registered)
+	 * @return a builder to enable chaining
+	 */
+	public UnterminatedFlowBuilder<Q> from(JobExecutionDecider decider) {
+		doFrom(decider);
+		return new UnterminatedFlowBuilder<Q>(this);
+	}
+
+	/**
+	 * Go next on successful completion to a subflow.
+	 * 
+	 * @param flow the flow to go to
+	 * @return a builder to enable chaining
+	 */
+	public FlowBuilder<Q> next(Flow flow) {
+		doNext(flow);
+		return this;
+	}
+
+	/**
+	 * Start again from a subflow that was already registered.
+	 * 
+	 * @param flow the flow to start from (already registered)
+	 * @return a builder to enable chaining
+	 */
+	public FlowBuilder<Q> from(Flow flow) {
+		doFrom(flow);
+		return this;
+	}
+
+	/**
+	 * If a flow should start with a subflow use this as the first state.
+	 * 
+	 * @param flow the flow to start from
+	 * @return a builder to enable chaining
+	 */
+	public FlowBuilder<Q> start(Flow flow) {
+		doStart(flow);
+		return this;
+	}
+
+	/**
+	 * @param executor a task executor to execute the split flows
+	 * @return a builder to enable fluent chaining
+	 */
+	public SplitBuilder<Q> split(TaskExecutor executor) {
+		return new SplitBuilder<Q>(this, executor);
+	}
+
+	/**
+	 * Start a transition to a new state if the exit status from the previous state matches the pattern given.
+	 * Successful completion normally results in an exit status equal to (or starting with by convention) "COMPLETED".
+	 * See {@link ExitStatus} for commonly used values.
+	 * 
+	 * @param pattern the pattern of exit status on which to take this transition
+	 * @return a builder to enable fluent chaining
+	 */
+	public TransitionBuilder<Q> on(String pattern) {
+		return new TransitionBuilder<Q>(this, pattern);
+	}
+
+	/**
+	 * A synonym for {@link #build()} which callers might find useful. Subclasses can override build to create an object
+	 * of the desired type (e.g. a parent builder or an actual flow).
+	 * 
+	 * @return the result of the builder
+	 * 
+	 * @param Q the type of the result
+	 */
+	public final Q end() {
+		return build();
+	}
+
+	protected Flow flow() {
+		if (!dirty) {
+			// optimization in case this method is called consecutively
+			return flow;
+		}
+		flow = new SimpleFlow(name);
+		// optimization for flows that only have one state that itself is a flow:
+		if (currentState instanceof FlowState && states.size() == 1) {
+			return ((FlowState) currentState).getFlows().iterator().next();
+		}
+		addDanglingEndStates();
+		flow.setStateTransitions(transitions);
+		dirty = false;
+		return flow;
+	}
+
+	private void doNext(Object input) {
+		if (this.currentState == null) {
+			doStart(input);
+		}
+		State next = createState(input);
+		addTransition("COMPLETED", next);
+		addTransition("*", failedState);
+		this.currentState = next;
+	}
+
+	private void doStart(Object input) {
+		if (this.currentState != null) {
+			doFrom(input);
+		}
+		this.currentState = createState(input);
+	}
+
+	private void doFrom(Object input) {
+		if (currentState == null) {
+			doStart(input);
+		}
+		State state = createState(input);
+		tos.put(currentState.getName(), currentState);
+		this.currentState = state;
+	}
+
+	private State createState(Object input) {
+		State result;
+		if (input instanceof Step) {
+			if (!states.containsKey(input)) {
+				Step step = (Step) input;
+				states.put(input, new StepState(prefix + step.getName(), step));
+			}
+			result = states.get(input);
+		}
+		else if (input instanceof JobExecutionDecider) {
+			if (!states.containsKey(input)) {
+				states.put(input, new DecisionState((JobExecutionDecider) input, prefix + "decision"
+						+ (decisionCounter++)));
+			}
+			result = states.get(input);
+		}
+		else if (input instanceof Flow) {
+			if (!states.containsKey(input)) {
+				states.put(input, new FlowState((Flow) input, prefix + ((Flow) input).getName()));
+			}
+			result = states.get(input);
+		}
+		else {
+			throw new FlowBuilderException("No state can be created for: " + input);
+		}
+		dirty = true;
+		return result;
+	}
+
+	private SplitState createState(Collection<Flow> flows, TaskExecutor executor) {
+		if (!states.containsKey(flows)) {
+			states.put(flows, new SplitState(flows, prefix + "split" + (splitCounter++)));
+		}
+		SplitState result = (SplitState) states.get(flows);
+		if (executor != null) {
+			result.setTaskExecutor(executor);
+		}
+		dirty = true;
+		return result;
+	}
+
+	private void addDanglingEndStates() {
+		Set<String> froms = new HashSet<String>();
+		for (StateTransition transition : transitions) {
+			froms.add(transition.getState().getName());
+		}
+		if (tos.isEmpty() && currentState != null) {
+			tos.put(currentState.getName(), currentState);
+		}
+		Map<String, State> copy = new HashMap<String, State>(tos);
+		// Find all the states that are really end states but not explicitly declared as such
+		for (String to : copy.keySet()) {
+			if (!froms.contains(to)) {
+				currentState = copy.get(to);
+				if (currentState != completedState) {
+					addTransition("COMPLETED", completedState);
+				}
+				if (currentState != failedState) {
+					addTransition("*", failedState);
+				}
+			}
+		}
+		copy = new HashMap<String, State>(tos);
+		// Then find the states that do not have a default transition
+		for (String from : copy.keySet()) {
+			currentState = copy.get(from);
+			if (currentState != failedState) {
+				if (!hasFail(from)) {
+					addTransition("*", failedState);
+				}
+			}
+			if (currentState != completedState) {
+				if (!hasCompleted(from)) {
+					addTransition("*", completedState);
+				}
+			}
+		}
+	}
+
+	private boolean hasFail(String from) {
+		return matches(from, "FAILED");
+	}
+
+	private boolean hasCompleted(String from) {
+		return matches(from, "COMPLETED");
+	}
+
+	private boolean matches(String from, String status) {
+		for (StateTransition transition : transitions) {
+			if (from.equals(transition.getState().getName()) && transition.matches(status)) {
+				return true;
+			}
+		}
+		return false;
+	}
+
+	private void addTransition(String pattern, State next) {
+		tos.put(next.getName(), next);
+		transitions.add(StateTransition.createStateTransition(currentState, pattern, next.getName()));
+		if (transitions.size() == 1) {
+			transitions.add(StateTransition.createEndStateTransition(failedState));
+			transitions.add(StateTransition.createEndStateTransition(completedState));
+		}
+		dirty = true;
+	}
+
+	private void end(String pattern) {
+		addTransition(pattern, completedState);
+	}
+
+	private void fail(String pattern) {
+		addTransition(pattern, failedState);
+	}
+
+	/**
+	 * A builder for continuing a flow from a decision state.
+	 * 
+	 * @author Dave Syer
+	 * 
+	 * @param <Q> the result of the builder's build()
+	 */
+	public static class UnterminatedFlowBuilder<Q> {
+
+		private final FlowBuilder<Q> parent;
+
+		public UnterminatedFlowBuilder(FlowBuilder<Q> parent) {
+			this.parent = parent;
+		}
+
+		/**
+		 * Start a transition to a new state if the exit status from the previous state matches the pattern given.
+		 * Successful completion normally results in an exit status equal to (or starting with by convention)
+		 * "COMPLETED". See {@link ExitStatus} for commonly used values.
+		 * 
+		 * @param pattern the pattern of exit status on which to take this transition
+		 * @return
+		 */
+		public TransitionBuilder<Q> on(String pattern) {
+			return new TransitionBuilder<Q>(parent, pattern);
+		}
+
+	}
+
+	/**
+	 * A builder for transitions within a flow.
+	 * 
+	 * @author Dave Syer
+	 * 
+	 * @param <Q> the result of the parent builder's build()
+	 */
+	public static class TransitionBuilder<Q> {
+
+		private final FlowBuilder<Q> parent;
+
+		private final String pattern;
+
+		public TransitionBuilder(FlowBuilder<Q> parent, String pattern) {
+			this.parent = parent;
+			this.pattern = pattern;
+		}
+
+		/**
+		 * Specify the next step.
+		 * 
+		 * @param step the next step after this transition
+		 * @return a FlowBuilder
+		 */
+		public FlowBuilder<Q> to(Step step) {
+			State next = parent.createState(step);
+			parent.addTransition(pattern, next);
+			parent.currentState = next;
+			return parent;
+		}
+
+		/**
+		 * Specify the next state as a complete flow.
+		 * 
+		 * @param step the next step after this transition
+		 * @return a FlowBuilder
+		 */
+		public FlowBuilder<Q> to(Flow flow) {
+			State next = parent.createState(flow);
+			parent.addTransition(pattern, next);
+			parent.currentState = next;
+			return parent;
+		}
+
+		/**
+		 * Specify the next state as a decision.
+		 * 
+		 * @param step the next step after this transition
+		 * @return a FlowBuilder
+		 */
+		public FlowBuilder<Q> to(JobExecutionDecider decider) {
+			State next = parent.createState(decider);
+			parent.addTransition(pattern, next);
+			parent.currentState = next;
+			return parent;
+		}
+
+		/**
+		 * Signal the successful end of the flow.
+		 * 
+		 * @return a FlowBuilder
+		 */
+		public FlowBuilder<Q> end() {
+			parent.end(pattern);
+			return parent;
+		}
+
+		/**
+		 * Signal the end of the flow with an error condition.
+		 * 
+		 * @return a FlowBuilder
+		 */
+		public FlowBuilder<Q> fail() {
+			parent.fail(pattern);
+			return parent;
+		}
+	}
+
+	/**
+	 * A builder for building a split state. Example (<code>builder</code> is a {@link FlowBuilder}):
+	 * 
+	 * <pre>
+	 * Flow splitFlow = builder.start(flow1).split(new SyncTaskExecutor()).add(flow2).build();
+	 * </pre>
+	 * 
+	 * where <code>flow1</code> and <code>flow2</code> will be executed (one after the other because of the task
+	 * executor that was added). Another example
+	 * 
+	 * <pre>
+	 * Flow splitFlow = builder.start(step1).split(new SimpleAsyncTaskExecutor()).add(flow).build();
+	 * </pre>
+	 * 
+	 * In this example, a flow consisting of <code>step1</code> will be executed in parallel with <code>flow</code>.
+	 * 
+	 * @author Dave Syer
+	 * 
+	 * @param <Q> the result of the parent builder's build()
+	 */
+	public static class SplitBuilder<Q> {
+
+		private final FlowBuilder<Q> parent;
+
+		private TaskExecutor executor;
+
+		/**
+		 * @param parent the parent builder
+		 * @param executor the task executor to use in the split
+		 */
+		public SplitBuilder(FlowBuilder<Q> parent, TaskExecutor executor) {
+			this.parent = parent;
+			this.executor = executor;
+		}
+
+		/**
+		 * Add flows to the split, in addition to the current state already present in the parent builder.
+		 * 
+		 * @param flows more flows to add to the split
+		 * @return the parent builder
+		 */
+		public FlowBuilder<Q> add(Flow... flows) {
+			Collection<Flow> list = new ArrayList<Flow>(Arrays.asList(flows));
+			String name = "split" + (parent.splitCounter++);
+			int counter = 0;
+			State one = parent.currentState;
+			Flow flow = null;
+			if (!(one instanceof FlowState)) {
+				FlowBuilder<Flow> stateBuilder = new FlowBuilder<Flow>(name + "_" + (counter++));
+				stateBuilder.currentState = one;
+				flow = stateBuilder.build();
+			}
+			if (flow != null) {
+				list.add(flow);
+			}
+			State next = parent.createState(list, executor);
+			parent.currentState = next;
+			return parent;
+		}
+
+	}
+
+}

spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/FlowBuilderException.java

@@ -0,0 +1,38 @@
+/*
+ * Copyright 2012-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.job.builder;
+
+/**
+ * @author Dave Syer
+ * 
+ * @since 2.2
+ *
+ */
+public class FlowBuilderException extends RuntimeException {
+
+	public FlowBuilderException(String msg, Exception e) {
+		super(msg, e);
+	}
+
+	public FlowBuilderException(Exception e) {
+		super(e);
+	}
+
+	public FlowBuilderException(String msg) {
+		super(msg);
+	}
+
+}

spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/FlowJobBuilder.java

@@ -0,0 +1,95 @@
+/*
+ * Copyright 2006-2011 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.core.job.builder;
+
+import org.springframework.batch.core.Job;
+import org.springframework.batch.core.Step;
+import org.springframework.batch.core.job.flow.Flow;
+import org.springframework.batch.core.job.flow.FlowJob;
+import org.springframework.batch.core.step.builder.StepBuilderException;
+
+/**
+ * A job builder for {@link FlowJob} instances. A flow job delegates processing to a nested flow composed of steps and
+ * conditional transitions between steps.
+ * 
+ * @author Dave Syer
+ * 
+ * @since 2.2
+ */
+public class FlowJobBuilder extends JobBuilderHelper<FlowJobBuilder> {
+
+	private Flow flow;
+
+	/**
+	 * Create a new builder initialized with any properties in the parent. The parent is copied, so it can be re-used.
+	 * 
+	 * @param parent a parent helper containing common job properties
+	 */
+	public FlowJobBuilder(JobBuilderHelper<?> parent) {
+		super(parent);
+	}
+
+	/**
+	 * Start a job with this flow, but expect to transition from there to other flows or steps.
+	 * 
+	 * @param flow the flow to start with
+	 * @return a builder to enable fluent chaining
+	 */
+	public JobFlowBuilder start(Flow flow) {
+		return new JobFlowBuilder(this, flow);
+	}
+
+	/**
+	 * Start a job with this step, but expect to transition from there to other flows or steps.
+	 * 
+	 * @param step the step to start with
+	 * @return a builder to enable fluent chaining
+	 */
+	public JobFlowBuilder start(Step step) {
+		return new JobFlowBuilder(this, step);
+	}
+
+	/**
+	 * Provide a single flow to execute as the job.
+	 * 
+	 * @param flow the flow to execute
+	 * @return this for fluent chaining
+	 */
+	protected FlowJobBuilder flow(Flow flow) {
+		this.flow = flow;
+		return this;
+	}
+
+	/**
+	 * Build a job that executes the flow provided, normally composed of other steps.
+	 * 
+	 * @return a flow job
+	 */
+	public Job build() {
+		FlowJob job = new FlowJob();
+		job.setName(getName());
+		job.setFlow(flow);
+		super.enhance(job);
+		try {
+			job.afterPropertiesSet();
+		}
+		catch (Exception e) {
+			throw new StepBuilderException(e);
+		}
+		return job;
+	}
+
+}

spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/JobBuilder.java

@@ -0,0 +1,71 @@
+/*
+ * Copyright 2006-2011 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.core.job.builder;
+
+import org.springframework.batch.core.Step;
+import org.springframework.batch.core.job.flow.Flow;
+
+/**
+ * Convenience for building jobs of various kinds.
+ * 
+ * @author Dave Syer
+ * 
+ * @since 2.2
+ * 
+ */
+public class JobBuilder extends JobBuilderHelper<JobBuilder> {
+
+	/**
+	 * Create a new builder for a job with the given name.
+	 * 
+	 * @param name the name of the job
+	 */
+	public JobBuilder(String name) {
+		super(name);
+	}
+
+	/**
+	 * Create a new job builder that will execute a step or sequence of steps.
+	 * 
+	 * @param step a step to execute
+	 * @return a {@link SimpleJobBuilder}
+	 */
+	public SimpleJobBuilder start(Step step) {
+		return new SimpleJobBuilder(this).start(step);
+	}
+
+	/**
+	 * Create a new job builder that will execute a step or sequence of steps.
+	 * 
+	 * @param step a step to execute
+	 * @return a {@link SimpleJobBuilder}
+	 */
+	public JobFlowBuilder start(Flow flow) {
+		return new FlowJobBuilder(this).start(flow);
+	}
+
+	/**
+	 * Create a new job builder that will execute a step or sequence of steps.
+	 * 
+	 * @param step a step to execute
+	 * @return a {@link SimpleJobBuilder}
+	 */
+	public JobFlowBuilder flow(Step step) {
+		return new FlowJobBuilder(this).start(step);
+	}
+
+
+}

spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/JobBuilderException.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2012-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.job.builder;
+
+/**
+ * @author Dave Syer
+ * 
+ * @since 2.2
+ *
+ */
+public class JobBuilderException extends RuntimeException {
+
+	public JobBuilderException(Exception e) {
+		super(e);
+	}
+
+}

spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/JobBuilderHelper.java

@@ -0,0 +1,242 @@
+/*
+ * Copyright 2006-2011 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.core.job.builder;
+
+import java.util.ArrayList;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Set;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.batch.core.Job;
+import org.springframework.batch.core.JobExecutionListener;
+import org.springframework.batch.core.JobParametersIncrementer;
+import org.springframework.batch.core.JobParametersValidator;
+import org.springframework.batch.core.job.AbstractJob;
+import org.springframework.batch.core.repository.JobRepository;
+
+/**
+ * A base class and utility for other job builders providing access to common properties like job repository.
+ * 
+ * @author Dave Syer
+ * 
+ * @since 2.2
+ */
+public abstract class JobBuilderHelper<B extends JobBuilderHelper<B>> {
+
+	protected final Log logger = LogFactory.getLog(getClass());
+
+	private final CommonStepProperties properties;
+
+	public JobBuilderHelper(String name) {
+		this.properties = new CommonStepProperties();
+		properties.name = name;
+	}
+
+	/**
+	 * Create a new builder initialized with any properties in the parent. The parent is copied, so it can be re-used.
+	 * 
+	 * @param parent a parent helper containing common step properties
+	 */
+	protected JobBuilderHelper(JobBuilderHelper<?> parent) {
+		this.properties = new CommonStepProperties(parent.properties);
+	}
+
+	/**
+	 * Add a job parameters validator.
+	 * 
+	 * @param jobParametersValidator a job parameters validator
+	 * @return this to enable fluent chaining
+	 */
+	public B validator(JobParametersValidator jobParametersValidator) {
+		properties.jobParametersValidator = jobParametersValidator;
+		@SuppressWarnings("unchecked")
+		B result = (B) this;
+		return result;
+	}
+
+	/**
+	 * Add a job parameters incrementer.
+	 * 
+	 * @param jobParametersIncrementer a job parameters incrementer
+	 * @return this to enable fluent chaining
+	 */
+	public B incrementer(JobParametersIncrementer jobParametersIncrementer) {
+		properties.jobParametersIncrementer = jobParametersIncrementer;
+		@SuppressWarnings("unchecked")
+		B result = (B) this;
+		return result;
+	}
+
+	/**
+	 * Sets the job repository for the job.
+	 * 
+	 * @param jobRepository the job repository (mandatory)
+	 * @return this to enable fluent chaining
+	 */
+	public B repository(JobRepository jobRepository) {
+		properties.jobRepository = jobRepository;
+		@SuppressWarnings("unchecked")
+		B result = (B) this;
+		return result;
+	}
+
+	/**
+	 * Register a job execution listener.
+	 * 
+	 * @param listener a job execution listener
+	 * @return this to enable fluent chaining
+	 */
+	public B listener(JobExecutionListener listener) {
+		properties.addJobExecutionListener(listener);
+		@SuppressWarnings("unchecked")
+		B result = (B) this;
+		return result;
+	}
+
+	/**
+	 * Set a flag to prevent restart an execution of this job even if it has failed.
+	 * 
+	 * @return this to enable fluent chaining
+	 */
+	public B preventRestart() {
+		properties.restartable = false;
+		@SuppressWarnings("unchecked")
+		B result = (B) this;
+		return result;
+	}
+
+	protected String getName() {
+		return properties.name;
+	}
+
+	protected JobRepository getJobRepository() {
+		return properties.jobRepository;
+	}
+
+	protected boolean isRestartable() {
+		return properties.restartable;
+	}
+
+	protected void enhance(Job target) {
+
+		if (target instanceof AbstractJob) {
+
+			AbstractJob job = (AbstractJob) target;
+			job.setJobRepository(properties.getJobRepository());
+
+			JobParametersIncrementer jobParametersIncrementer = properties.getJobParametersIncrementer();
+			if (jobParametersIncrementer != null) {
+				job.setJobParametersIncrementer(jobParametersIncrementer);
+			}
+			JobParametersValidator jobParametersValidator = properties.getJobParametersValidator();
+			if (jobParametersValidator != null) {
+				job.setJobParametersValidator(jobParametersValidator);
+			}
+
+			Boolean restartable = properties.getRestartable();
+			if (restartable != null) {
+				job.setRestartable(restartable);
+			}
+
+			List<JobExecutionListener> listeners = properties.getJobExecutionListeners();
+			if (!listeners.isEmpty()) {
+				job.setJobExecutionListeners(listeners.toArray(new JobExecutionListener[0]));
+			}
+
+		}
+
+	}
+
+	public static class CommonStepProperties {
+
+		private Set<JobExecutionListener> jobExecutionListeners = new LinkedHashSet<JobExecutionListener>();
+
+		private boolean restartable = true;
+
+		private JobRepository jobRepository;
+
+		private JobParametersIncrementer jobParametersIncrementer;
+
+		private JobParametersValidator jobParametersValidator;
+
+		public CommonStepProperties() {
+		}
+
+		public CommonStepProperties(CommonStepProperties properties) {
+			this.name = properties.name;
+			this.restartable = properties.restartable;
+			this.jobRepository = properties.jobRepository;
+		}
+
+		public JobParametersIncrementer getJobParametersIncrementer() {
+			return jobParametersIncrementer;
+		}
+
+		public void setJobParametersIncrementer(JobParametersIncrementer jobParametersIncrementer) {
+			this.jobParametersIncrementer = jobParametersIncrementer;
+		}
+
+		public JobParametersValidator getJobParametersValidator() {
+			return jobParametersValidator;
+		}
+
+		public void setJobParametersValidator(JobParametersValidator jobParametersValidator) {
+			this.jobParametersValidator = jobParametersValidator;
+		}
+
+		public JobRepository getJobRepository() {
+			return jobRepository;
+		}
+
+		public void setJobRepository(JobRepository jobRepository) {
+			this.jobRepository = jobRepository;
+		}
+
+		public String getName() {
+			return name;
+		}
+
+		public void setName(String name) {
+			this.name = name;
+		}
+
+		public List<JobExecutionListener> getJobExecutionListeners() {
+			return new ArrayList<JobExecutionListener>(jobExecutionListeners);
+		}
+
+		public void addStepExecutionListeners(List<JobExecutionListener> jobExecutionListeners) {
+			this.jobExecutionListeners.addAll(jobExecutionListeners);
+		}
+
+		public void addJobExecutionListener(JobExecutionListener jobExecutionListener) {
+			this.jobExecutionListeners.add(jobExecutionListener);
+		}
+
+		public boolean getRestartable() {
+			return restartable;
+		}
+
+		public void setRestartable(boolean restartable) {
+			this.restartable = restartable;
+		}
+
+		private String name;
+
+	}
+
+}

spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/JobFlowBuilder.java

@@ -0,0 +1,65 @@
+/*
+ * Copyright 2012-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.job.builder;
+
+import org.springframework.batch.core.Step;
+import org.springframework.batch.core.job.flow.Flow;
+import org.springframework.batch.core.job.flow.JobExecutionDecider;
+
+/**
+ * @author Dave Syer
+ * 
+ */
+public class JobFlowBuilder extends FlowBuilder<FlowJobBuilder> {
+
+	private FlowJobBuilder parent;
+
+	public JobFlowBuilder(FlowJobBuilder parent) {
+		super(parent.getName());
+		this.parent = parent;
+	}
+
+	public JobFlowBuilder(FlowJobBuilder parent, Step step) {
+		super(parent.getName());
+		this.parent = parent;
+		start(step);
+	}
+
+	public JobFlowBuilder(FlowJobBuilder parent, JobExecutionDecider decider) {
+		super(parent.getName());
+		this.parent = parent;
+		start(decider);
+	}
+
+	public JobFlowBuilder(FlowJobBuilder parent, Flow flow) {
+		super(parent.getName());
+		this.parent = parent;
+		start(flow);
+	}
+
+	/**
+	 * Build a flow and inject it into the parent builder. The parent builder is then returned so it can be enhanced
+	 * before building an actual job.  Normally called explicitly via {@link #end()}.
+	 * 
+	 * @see org.springframework.batch.core.job.builder.FlowBuilder#build()
+	 */
+	public FlowJobBuilder build() {
+		Flow flow = flow();
+		parent.flow(flow);
+		return parent;
+	}
+
+}

spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/SimpleJobBuilder.java

@@ -0,0 +1,178 @@
+/*
+ * Copyright 2012-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.job.builder;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.springframework.batch.core.Job;
+import org.springframework.batch.core.Step;
+import org.springframework.batch.core.job.SimpleJob;
+import org.springframework.batch.core.job.flow.JobExecutionDecider;
+import org.springframework.core.task.TaskExecutor;
+import org.springframework.util.Assert;
+
+/**
+ * @author Dave Syer
+ * 
+ * @since 2.2
+ * 
+ */
+public class SimpleJobBuilder extends JobBuilderHelper<SimpleJobBuilder> {
+
+	private List<Step> steps = new ArrayList<Step>();
+
+	private JobFlowBuilder builder;
+
+	/**
+	 * Create a new builder initialized with any properties in the parent. The parent is copied, so it can be re-used.
+	 * 
+	 * @param parent the parent to use
+	 */
+	public SimpleJobBuilder(JobBuilderHelper<?> parent) {
+		super(parent);
+	}
+
+	public Job build() {
+		if (builder != null) {
+			return builder.end().build();
+		}
+		SimpleJob job = new SimpleJob();
+		super.enhance(job);
+		job.setSteps(steps);
+		try {
+			job.afterPropertiesSet();
+		}
+		catch (Exception e) {
+			throw new JobBuilderException(e);
+		}
+		return job;
+	}
+
+	/**
+	 * Start the job with this step.
+	 * 
+	 * @param step a step to start with
+	 * @return this for fluent chaining
+	 */
+	public SimpleJobBuilder start(Step step) {
+		if (steps.isEmpty()) {
+			steps.add(step);
+		}
+		else {
+			steps.set(0, step);
+		}
+		return this;
+	}
+
+	/**
+	 * Branch into a flow conditional on the outcome of the current step.
+	 * 
+	 * @param pattern a pattern for the exit status of the current step
+	 * @return a builder for fluent chaining
+	 */
+	public FlowBuilder.TransitionBuilder<FlowJobBuilder> on(String pattern) {
+		Assert.state(steps.size() > 0, "You have to start a job with a step");
+		for (Step step : steps) {
+			if (builder == null) {
+				builder = new JobFlowBuilder(new FlowJobBuilder(this), step);
+			}
+			else {
+				builder.next(step);
+			}
+		}
+		return builder.on(pattern);
+	}
+
+	/**
+	 * Start with this decider. Returns a flow builder and when the flow is ended a job builder will be returned to
+	 * continue the job configuration if needed.
+	 * 
+	 * @param decider a decider to execute first
+	 * @return builder for fluent chaining
+	 */
+	public JobFlowBuilder start(JobExecutionDecider decider) {
+		if (builder == null) {
+			builder = new JobFlowBuilder(new FlowJobBuilder(this), decider);
+		}
+		else {
+			builder.start(decider);
+		}
+		if (!steps.isEmpty()) {
+			steps.remove(0);
+		}
+		for (Step step : steps) {
+			builder.next(step);
+		}
+		return builder;
+	}
+
+	/**
+	 * Continue with this decider if the previous step was successful. Returns a flow builder and when the flow is ended
+	 * a job builder will be returned to continue the job configuration if needed.
+	 * 
+	 * @param decider a decider to execute next
+	 * @return builder for fluent chaining
+	 */
+	public JobFlowBuilder next(JobExecutionDecider decider) {
+		for (Step step : steps) {
+			if (builder == null) {
+				builder = new JobFlowBuilder(new FlowJobBuilder(this), step);
+			}
+			else {
+				builder.next(step);
+			}
+		}
+		if (builder == null) {
+			builder = new JobFlowBuilder(new FlowJobBuilder(this), decider);
+		}
+		else {
+			builder.next(decider);
+		}
+		return builder;
+	}
+
+	/**
+	 * Continue or end a job with this step if the previous step was successful.
+	 * 
+	 * @param step a step to execute next
+	 * @return this for fluent chaining
+	 */
+	public SimpleJobBuilder next(Step step) {
+		steps.add(step);
+		return this;
+	}
+
+	/**
+	 * @param executor
+	 * @return builder for fluent chaining
+	 */
+	public JobFlowBuilder.SplitBuilder<FlowJobBuilder> split(TaskExecutor executor) {
+		for (Step step : steps) {
+			if (builder == null) {
+				builder = new JobFlowBuilder(new FlowJobBuilder(this), step);
+			}
+			else {
+				builder.next(step);
+			}
+		}
+		if (builder == null) {
+			builder = new JobFlowBuilder(new FlowJobBuilder(this));
+		}
+		return builder.split(executor);
+	}
+
+}

spring-batch-core/src/test/java/org/springframework/batch/core/job/builder/FlowBuilderTests.java

@@ -0,0 +1,50 @@
+/*
+ * Copyright 2012-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.job.builder;
+
+import org.junit.Test;
+import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.core.JobInterruptedException;
+import org.springframework.batch.core.JobParameters;
+import org.springframework.batch.core.StepExecution;
+import org.springframework.batch.core.UnexpectedJobExecutionException;
+import org.springframework.batch.core.job.SimpleStepHandler;
+import org.springframework.batch.core.job.flow.Flow;
+import org.springframework.batch.core.job.flow.JobFlowExecutor;
+import org.springframework.batch.core.repository.JobRepository;
+import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean;
+import org.springframework.batch.core.step.StepSupport;
+
+/**
+ * @author Dave Syer
+ * 
+ */
+public class FlowBuilderTests {
+
+	@Test
+	public void test() throws Exception {
+		FlowBuilder<Flow> builder = new FlowBuilder<Flow>("flow");
+		JobRepository jobRepository = new MapJobRepositoryFactoryBean().getJobRepository();
+		JobExecution execution = jobRepository.createJobExecution("foo", new JobParameters());
+		builder.start(new StepSupport("step") {
+			@Override
+			public void execute(StepExecution stepExecution) throws JobInterruptedException,
+					UnexpectedJobExecutionException {
+			}
+		}).end().start(new JobFlowExecutor(jobRepository, new SimpleStepHandler(jobRepository), execution));
+	}
+
+}

spring-batch-core/src/test/java/org/springframework/batch/core/job/builder/FlowJobBuilderTests.java

@@ -0,0 +1,167 @@
+/*
+ * Copyright 2012-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.job.builder;
+
+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.JobInterruptedException;
+import org.springframework.batch.core.JobParameters;
+import org.springframework.batch.core.StepExecution;
+import org.springframework.batch.core.UnexpectedJobExecutionException;
+import org.springframework.batch.core.job.flow.Flow;
+import org.springframework.batch.core.job.flow.FlowExecutionStatus;
+import org.springframework.batch.core.job.flow.JobExecutionDecider;
+import org.springframework.batch.core.repository.JobRepository;
+import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean;
+import org.springframework.batch.core.step.StepSupport;
+import org.springframework.core.task.SimpleAsyncTaskExecutor;
+
+/**
+ * @author Dave Syer
+ * 
+ */
+public class FlowJobBuilderTests {
+
+	private JobRepository jobRepository;
+	
+	private JobExecution execution;
+
+	private StepSupport step1 = new StepSupport("step1") {
+		@Override
+		public void execute(StepExecution stepExecution) throws JobInterruptedException,
+				UnexpectedJobExecutionException {
+			stepExecution.upgradeStatus(BatchStatus.COMPLETED);
+			stepExecution.setExitStatus(ExitStatus.COMPLETED);
+			jobRepository.update(stepExecution);
+		}
+	};
+
+	private StepSupport step2 = new StepSupport("step2") {
+		@Override
+		public void execute(StepExecution stepExecution) throws JobInterruptedException,
+				UnexpectedJobExecutionException {
+			stepExecution.upgradeStatus(BatchStatus.COMPLETED);
+			stepExecution.setExitStatus(ExitStatus.COMPLETED);
+			jobRepository.update(stepExecution);
+		}
+	};
+
+	private StepSupport step3 = new StepSupport("step3") {
+		@Override
+		public void execute(StepExecution stepExecution) throws JobInterruptedException,
+				UnexpectedJobExecutionException {
+			stepExecution.upgradeStatus(BatchStatus.COMPLETED);
+			stepExecution.setExitStatus(ExitStatus.COMPLETED);
+			jobRepository.update(stepExecution);
+		}
+	};
+
+	@Before
+	public void init() throws Exception {
+		jobRepository = new MapJobRepositoryFactoryBean().getJobRepository();
+		execution = jobRepository.createJobExecution("flow", new JobParameters());
+	}
+
+	@Test
+	public void testBuildOnOneLine() throws Exception {
+		FlowJobBuilder builder = new JobBuilder("flow").repository(jobRepository).start(step1).on("COMPLETED")
+				.to(step2).end().preventRestart();
+		builder.build().execute(execution);
+		assertEquals(BatchStatus.COMPLETED, execution.getStatus());
+		assertEquals(2, execution.getStepExecutions().size());
+	}
+
+	@Test
+	public void testBuildSingleFlow() throws Exception {
+		Flow flow = new FlowBuilder<Flow>("subflow").from(step1).next(step2).build();
+		FlowJobBuilder builder = new JobBuilder("flow").repository(jobRepository).start(flow).end().preventRestart();
+		builder.build().execute(execution);
+		assertEquals(BatchStatus.COMPLETED, execution.getStatus());
+		assertEquals(2, execution.getStepExecutions().size());
+	}
+
+	@Test
+	public void testBuildOverTwoLines() throws Exception {
+		FlowJobBuilder builder = new JobBuilder("flow").repository(jobRepository).start(step1).on("COMPLETED")
+				.to(step2).end();
+		builder.preventRestart();
+		builder.build().execute(execution);
+		assertEquals(BatchStatus.COMPLETED, execution.getStatus());
+		assertEquals(2, execution.getStepExecutions().size());
+	}
+
+	@Test
+	public void testBuildSubflow() throws Exception {
+		Flow flow = new FlowBuilder<Flow>("subflow").from(step1).end();
+		JobFlowBuilder builder = new JobBuilder("flow").repository(jobRepository).start(flow);
+		builder.on("COMPLETED").to(step2);
+		builder.end().preventRestart().build().execute(execution);
+		assertEquals(BatchStatus.COMPLETED, execution.getStatus());
+		assertEquals(2, execution.getStepExecutions().size());
+	}
+
+	@Test
+	public void testBuildSplit() throws Exception {
+		Flow flow = new FlowBuilder<Flow>("subflow").from(step1).end();
+		SimpleJobBuilder builder = new JobBuilder("flow").repository(jobRepository).start(step2);
+		builder.split(new SimpleAsyncTaskExecutor()).add(flow).end();
+		builder.preventRestart().build().execute(execution);
+		assertEquals(BatchStatus.COMPLETED, execution.getStatus());
+		assertEquals(2, execution.getStepExecutions().size());
+	}
+
+	@Test
+	public void testBuildDecision() throws Exception {
+		JobExecutionDecider decider = new JobExecutionDecider() {
+			private int count = 0;
+			public FlowExecutionStatus decide(JobExecution jobExecution, StepExecution stepExecution) {
+				count++;
+				return count<2 ? new FlowExecutionStatus("ONGOING") : FlowExecutionStatus.COMPLETED;
+			}
+		};
+		step1.setAllowStartIfComplete(true);
+		SimpleJobBuilder builder = new JobBuilder("flow").repository(jobRepository).start(step1);
+		builder.next(decider).on("COMPLETED").end().from(decider).on("*").to(step1).end();
+		builder.preventRestart().build().execute(execution);
+		assertEquals(BatchStatus.COMPLETED, execution.getStatus());
+		assertEquals(2, execution.getStepExecutions().size());
+	}
+
+	@Test
+	public void testBuildWithIntermediateSimpleJob() throws Exception {
+		SimpleJobBuilder builder = new JobBuilder("flow").repository(jobRepository).start(step1);
+		builder.on("COMPLETED").to(step2).end();
+		builder.preventRestart();
+		builder.build().execute(execution);
+		assertEquals(BatchStatus.COMPLETED, execution.getStatus());
+		assertEquals(2, execution.getStepExecutions().size());
+	}
+
+	@Test
+	public void testBuildWithIntermediateSimpleJobTwoSteps() throws Exception {
+		SimpleJobBuilder builder = new JobBuilder("flow").repository(jobRepository).start(step1).next(step2);
+		builder.on("FAILED").to(step3).end();
+		builder.build().execute(execution);
+		assertEquals(BatchStatus.COMPLETED, execution.getStatus());
+		assertEquals(2, execution.getStepExecutions().size());
+	}
+
+}