RESOLVED - issue BATCH-1458: Add FlowStep: a Step implementation that executes a flow

Javadocs + re-org update() methods

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

@@ -352,17 +352,6 @@ public abstract class AbstractJob implements Job, StepLocator, BeanNameAware, In
 
 	}
 
-	/**
-	 * Convenience method for subclasses so they can change the state of a
-	 * {@link StepExecution} if necessary. Use with care (and not at all
-	 * preferably) and only before or after a step is executed.
-	 * 
-	 * @param stepExecution
-	 */
-	protected void updateStepExecution(StepExecution stepExecution) {
-		stepHandler.updateStepExecution(stepExecution);
-	}
-
 	/**
 	 * Default mapping from throwable to {@link ExitStatus}.
 	 * 

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

@@ -32,22 +32,25 @@ import org.springframework.beans.factory.InitializingBean;
 import org.springframework.util.Assert;
 
 /**
+ * Implementation of {@link StepHandler} that manages repository and restart
+ * concerns.
+ * 
  * @author Dave Syer
- *
+ * 
  */
 public class SimpleStepHandler implements StepHandler, InitializingBean {
 
 	private static final Log logger = LogFactory.getLog(SimpleStepHandler.class);
 
 	private JobRepository jobRepository;
-	
+
 	/**
 	 * Convenient default constructor for configuration usage.
 	 */
 	public SimpleStepHandler() {
 		this(null);
 	}
-	
+
 	/**
 	 * @param jobRepository
 	 */
@@ -55,14 +58,14 @@ public class SimpleStepHandler implements StepHandler, InitializingBean {
 		super();
 		this.jobRepository = jobRepository;
 	}
-	
+
 	/**
 	 * Check mandatory properties (jobRepository).
 	 * 
 	 * @see InitializingBean#afterPropertiesSet()
 	 */
 	public void afterPropertiesSet() throws Exception {
-		Assert.state(jobRepository!=null, "A JobRepository must be provided");
+		Assert.state(jobRepository != null, "A JobRepository must be provided");
 	}
 
 	/**
@@ -82,10 +85,11 @@ public class SimpleStepHandler implements StepHandler, InitializingBean {
 
 		StepExecution lastStepExecution = jobRepository.getLastStepExecution(jobInstance, step.getName());
 		if (stepExecutionPartOfExistingJobExecution(execution, lastStepExecution)) {
-			// If the last execution of this step was in the same job, it's probably
-			// intentional so we want to run it again...
-			logger.info(String.format("Duplicate step [%s] detected in execution of job=[%s]. " +
-					"If either step fails, both will be executed again on restart.", step.getName(), jobInstance.getJobName()));
+			// If the last execution of this step was in the same job, it's
+			// probably intentional so we want to run it again...
+			logger.info(String.format("Duplicate step [%s] detected in execution of job=[%s]. "
+					+ "If either step fails, both will be executed again on restart.", step.getName(), jobInstance
+					.getJobName()));
 			lastStepExecution = null;
 		}
 		StepExecution currentStepExecution = lastStepExecution;
@@ -135,10 +139,6 @@ public class SimpleStepHandler implements StepHandler, InitializingBean {
 		return currentStepExecution;
 	}
 
-	public void updateStepExecution(StepExecution stepExecution) {
-		jobRepository.update(stepExecution);
-	}
-
 	/**
 	 * Detect whether a step execution belongs to this job execution.
 	 * @param jobExecution the current job execution

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

@@ -16,6 +16,7 @@
 
 package org.springframework.batch.core.job;
 
+import org.springframework.batch.core.Job;
 import org.springframework.batch.core.JobExecution;
 import org.springframework.batch.core.JobInterruptedException;
 import org.springframework.batch.core.StartLimitExceededException;
@@ -23,11 +24,33 @@ import org.springframework.batch.core.Step;
 import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.repository.JobRestartException;
 
+/**
+ * Strategy interface for handling a {@link Step} on behalf of a {@link Job}.
+ * 
+ * @author Dave Syer
+ * 
+ */
 public interface StepHandler {
 
-	StepExecution handleStep(Step step, JobExecution execution) throws JobInterruptedException, JobRestartException,
+	/**
+	 * Handle a step and return the execution for it. Does not save the
+	 * {@link JobExecution}, but should manage the persistence of the
+	 * {@link StepExecution} if required (e.g. at least it needs to be added to
+	 * a repository before the step can eb executed).
+	 * 
+	 * @param step a {@link Step}
+	 * @param jobExecution a {@link JobExecution}
+	 * @return an execution of the step
+	 * 
+	 * @throws JobInterruptedException if there is an interruption
+	 * @throws JobRestartException if there is a problem restarting a failed
+	 * step
+	 * @throws StartLimitExceededException if the step exceeds its start limit
+	 * 
+	 * @see Job#execute(JobExecution)
+	 * @see Step#execute(StepExecution)
+	 */
+	StepExecution handleStep(Step step, JobExecution jobExecution) throws JobInterruptedException, JobRestartException,
 			StartLimitExceededException;
-	
-	void updateStepExecution(StepExecution stepExecution);
 
 }

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

@@ -94,7 +94,7 @@ public class FlowJob extends AbstractJob {
 	@Override
 	protected void doExecute(final JobExecution execution) throws JobExecutionException {
 		try {
-			JobFlowExecutor executor = new JobFlowExecutor(new SimpleStepHandler(getJobRepository()),
+			JobFlowExecutor executor = new JobFlowExecutor(getJobRepository(), new SimpleStepHandler(getJobRepository()),
 					execution);
 			executor.updateJobExecutionStatus(flow.start(executor).getStatus());
 		}

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

@@ -4,13 +4,42 @@ import org.springframework.batch.core.JobExecutionException;
 import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.job.SimpleStepHandler;
 import org.springframework.batch.core.job.StepHandler;
+import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.core.step.AbstractStep;
 import org.springframework.util.Assert;
 
+import com.sun.org.apache.xerces.internal.impl.xpath.XPath.Step;
+
+/**
+ * A {@link Step} implementation that delegates to a {@link Flow}. Useful for
+ * logical grouping of steps, and especially for partitioning with multiple
+ * steps per execution. If the flow has steps then when the {@link FlowStep}
+ * executes, all steps including the parent {@link FlowStep} will have
+ * executions in the {@link JobRepository} (one for the parent and one each for
+ * the flow steps).
+ * 
+ * @author Dave Syer
+ * 
+ */
 public class FlowStep extends AbstractStep {
-	
+
 	private Flow flow;
-	
+
+	/**
+	 * Default constructor convenient for configuration purposes.
+	 */
+	public FlowStep() {
+		super(null);
+	}
+
+	/**
+	 * Constructor for a {@link FlowStep} that sets the flow and of the step
+	 * explicitly.
+	 */
+	public FlowStep(Flow flow) {
+		super(flow.getName());
+	}
+
 	/**
 	 * Public setter for the flow.
 	 * 
@@ -19,7 +48,7 @@ public class FlowStep extends AbstractStep {
 	public void setFlow(Flow flow) {
 		this.flow = flow;
 	}
-	
+
 	/**
 	 * Ensure that the flow is set.
 	 * @see AbstractStep#afterPropertiesSet()
@@ -27,14 +56,19 @@ public class FlowStep extends AbstractStep {
 	@Override
 	public void afterPropertiesSet() throws Exception {
 		super.afterPropertiesSet();
-		Assert.state(flow!=null, "A Flow must be provided");
+		Assert.state(flow != null, "A Flow must be provided");
 	}
 
+	/**
+	 * Delegate to the flow provided for the execution of the step.
+	 * 
+	 * @see AbstractStep#doExecute(StepExecution)
+	 */
 	@Override
 	protected void doExecute(StepExecution stepExecution) throws Exception {
 		try {
 			StepHandler stepHandler = new SimpleStepHandler(getJobRepository());
-			FlowExecutor executor = new JobFlowExecutor(stepHandler, stepExecution.getJobExecution());
+			FlowExecutor executor = new JobFlowExecutor(getJobRepository(), stepHandler, stepExecution.getJobExecution());
 			executor.updateJobExecutionStatus(flow.start(executor).getStatus());
 		}
 		catch (FlowExecutionException e) {

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

@@ -24,6 +24,7 @@ import org.springframework.batch.core.StartLimitExceededException;
 import org.springframework.batch.core.Step;
 import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.job.StepHandler;
+import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.core.repository.JobRestartException;
 
 /**
@@ -43,10 +44,13 @@ public class JobFlowExecutor implements FlowExecutor {
 
 	private final StepHandler stepHandler;
 
+	private final JobRepository jobRepository;
+
 	/**
 	 * @param execution
 	 */
-	public JobFlowExecutor(StepHandler stepHandler, JobExecution execution) {
+	public JobFlowExecutor(JobRepository jobRepository, StepHandler stepHandler, JobExecution execution) {
+		this.jobRepository = jobRepository;
 		this.stepHandler = stepHandler;
 		this.execution = execution;
 		stepExecutionHolder.set(null);
@@ -63,7 +67,7 @@ public class JobFlowExecutor implements FlowExecutor {
 		StepExecution lastStepExecution = stepExecutionHolder.get();
 		if (lastStepExecution != null && lastStepExecution.getStatus().isGreaterThan(BatchStatus.STOPPING)) {
 			lastStepExecution.upgradeStatus(BatchStatus.ABANDONED);
-			stepHandler.updateStepExecution(lastStepExecution);
+			jobRepository.update(lastStepExecution);
 		}
 	}
 

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

@@ -16,7 +16,7 @@
 
 package org.springframework.batch.core.job;
 
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertEquals;
 
 import org.junit.Before;
 import org.junit.Test;
@@ -27,7 +27,6 @@ import org.springframework.batch.core.JobInterruptedException;
 import org.springframework.batch.core.JobParameters;
 import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.repository.JobRepository;
-import org.springframework.batch.core.repository.dao.StepExecutionDao;
 import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean;
 import org.springframework.batch.core.step.StepSupport;
 
@@ -43,13 +42,10 @@ public class SimpleStepHandlerTests {
 
 	private SimpleStepHandler stepHandler;
 
-	private StepExecutionDao stepExecutionDao;
-
 	@Before
 	public void setUp() throws Exception {
 		MapJobRepositoryFactoryBean jobRepositoryFactoryBean = new MapJobRepositoryFactoryBean();
 		jobRepository = jobRepositoryFactoryBean.getJobRepository();
-		stepExecutionDao = jobRepositoryFactoryBean.getStepExecutionDao();
 		jobExecution = jobRepository.createJobExecution("job", new JobParameters());
 		stepHandler = new SimpleStepHandler(jobRepository);
 		stepHandler.afterPropertiesSet();
@@ -75,20 +71,6 @@ public class SimpleStepHandlerTests {
 		assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus());
 	}
 
-	/**
-	 * Test method for
-	 * {@link SimpleStepHandler#updateStepExecution(org.springframework.batch.core.StepExecution)}
-	 * 
-	 */
-	@Test
-	public void testUpdateStepExecution() {
-		StepExecution stepExecution = jobExecution.createStepExecution("step");
-		jobRepository.add(stepExecution);
-		stepExecution.setStatus(BatchStatus.FAILED);
-		stepHandler.updateStepExecution(stepExecution);
-		assertEquals(stepExecution, stepExecutionDao.getStepExecution(jobExecution, stepExecution.getId()));
-	}
-
 	private class StubStep extends StepSupport {
 
 		private StubStep(String name) {