OPEN - issue BATCH-90: StepExecution and StepExecutionContext are parallel domains, and StepExecution is by comparison anaemic

http://opensource.atlassian.com/projects/spring/browse/BATCH-90

Remove *ExecutionContext by merging with *Execution

execution/src/main/java/org/springframework/batch/execution/bootstrap/TaskExecutorJobLauncher.java

@@ -23,7 +23,6 @@ import javax.management.Notification;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.springframework.batch.core.configuration.NoSuchJobConfigurationException;
-import org.springframework.batch.core.runtime.JobExecutionContext;
 import org.springframework.batch.core.runtime.JobIdentifier;
 import org.springframework.batch.execution.facade.JobExecutorFacade;
 import org.springframework.batch.execution.facade.NoSuchJobExecutionException;

execution/src/main/java/org/springframework/batch/execution/facade/SimpleJobExecutorFacade.java

@@ -22,10 +22,10 @@ import java.util.Properties;
 import org.springframework.batch.core.configuration.JobConfiguration;
 import org.springframework.batch.core.configuration.JobConfigurationLocator;
 import org.springframework.batch.core.configuration.NoSuchJobConfigurationException;
+import org.springframework.batch.core.domain.JobExecution;
 import org.springframework.batch.core.domain.JobInstance;
 import org.springframework.batch.core.executor.JobExecutor;
 import org.springframework.batch.core.repository.JobRepository;
-import org.springframework.batch.core.runtime.JobExecutionContext;
 import org.springframework.batch.core.runtime.JobExecutionRegistry;
 import org.springframework.batch.core.runtime.JobIdentifier;
 import org.springframework.batch.execution.job.DefaultJobExecutor;
@@ -127,7 +127,7 @@ public class SimpleJobExecutorFacade implements JobExecutorFacade, StatisticsPro
 				.getJobConfiguration(jobRuntimeInformation.getName());
 
 		final JobInstance job = jobRepository.findOrCreateJob(jobConfiguration, jobRuntimeInformation);
-		JobExecutionContext jobExecutionContext = jobExecutionRegistry.register(job);
+		JobExecution jobExecutionContext = jobExecutionRegistry.register(job);
 		
 		ExitStatus exitStatus = ExitStatus.FAILED;
 		try {
@@ -153,7 +153,7 @@ public class SimpleJobExecutorFacade implements JobExecutorFacade, StatisticsPro
 	 * @see org.springframework.batch.container.BatchContainer#stop(org.springframework.batch.container.common.runtime.JobRuntimeInformation)
 	 */
 	public void stop(JobIdentifier runtimeInformation) throws NoSuchJobExecutionException {
-		JobExecutionContext jobExecutionContext = (JobExecutionContext) jobExecutionRegistry.get(runtimeInformation);
+		JobExecution jobExecutionContext = jobExecutionRegistry.get(runtimeInformation);
 		if (jobExecutionContext == null) {
 			throw new NoSuchJobExecutionException("No such Job is executing: [" + runtimeInformation + "]");
 		}
@@ -193,7 +193,7 @@ public class SimpleJobExecutorFacade implements JobExecutorFacade, StatisticsPro
 		int i = 0;
 		Properties props = new Properties();
 		for (Iterator iter = jobExecutionRegistry.findAll().iterator(); iter.hasNext();) {
-			JobExecutionContext element = (JobExecutionContext) iter.next();
+			JobExecution element = (JobExecution) iter.next();
 			i++;
 			String runtime = "job" + i;
 			props.setProperty(runtime, "" + element.getJobIdentifier());

execution/src/main/java/org/springframework/batch/execution/facade/VolatileJobExecutionRegistry.java

@@ -22,8 +22,8 @@ import java.util.Iterator;
 import java.util.Map;
 import java.util.Set;
 
+import org.springframework.batch.core.domain.JobExecution;
 import org.springframework.batch.core.domain.JobInstance;
-import org.springframework.batch.core.runtime.JobExecutionContext;
 import org.springframework.batch.core.runtime.JobExecutionRegistry;
 import org.springframework.batch.core.runtime.JobIdentifier;
 
@@ -74,10 +74,10 @@ public class VolatileJobExecutionRegistry implements JobExecutionRegistry {
 	 * (non-Javadoc)
 	 * @see org.springframework.batch.container.common.executor.JobExecutionRegistry#findByRuntimeInformation(org.springframework.batch.container.common.runtime.JobRuntimeInformation)
 	 */
-	public JobExecutionContext get(JobIdentifier runtimeInformation) {
+	public JobExecution get(JobIdentifier runtimeInformation) {
 
 		synchronized (this.contexts) {
-			return (JobExecutionContext) contexts.get(runtimeInformation);
+			return (JobExecution) contexts.get(runtimeInformation);
 		}
 
 	}
@@ -98,14 +98,14 @@ public class VolatileJobExecutionRegistry implements JobExecutionRegistry {
 	 * @see org.springframework.batch.container.common.executor.JobExecutionRegistry#register(org.springframework.batch.container.common.runtime.JobRuntimeInformation,
 	 * org.springframework.batch.container.common.domain.JobExecution)
 	 */
-	public JobExecutionContext register(JobInstance job) {
+	public JobExecution register(JobInstance job) {
 		
 		JobIdentifier jobIdentifier = job.getIdentifier();
 		
 		if (isRegistered(jobIdentifier)) {
 			return get(jobIdentifier);
 		}
-		JobExecutionContext context = new JobExecutionContext(jobIdentifier, job);
+		JobExecution context = new JobExecution(job);
 
 		synchronized (this.contexts) {
 			contexts.put(jobIdentifier, context);

execution/src/main/java/org/springframework/batch/execution/job/DefaultJobExecutor.java

@@ -25,6 +25,7 @@ import org.springframework.batch.core.configuration.StepConfiguration;
 import org.springframework.batch.core.domain.BatchStatus;
 import org.springframework.batch.core.domain.JobExecution;
 import org.springframework.batch.core.domain.JobInstance;
+import org.springframework.batch.core.domain.StepExecution;
 import org.springframework.batch.core.domain.StepInstance;
 import org.springframework.batch.core.executor.ExitCodeExceptionClassifier;
 import org.springframework.batch.core.executor.JobExecutor;
@@ -32,8 +33,6 @@ import org.springframework.batch.core.executor.StepExecutor;
 import org.springframework.batch.core.executor.StepExecutorFactory;
 import org.springframework.batch.core.executor.StepInterruptedException;
 import org.springframework.batch.core.repository.JobRepository;
-import org.springframework.batch.core.runtime.JobExecutionContext;
-import org.springframework.batch.core.runtime.StepExecutionContext;
 import org.springframework.batch.execution.step.SimpleStepExecutorFactory;
 import org.springframework.batch.execution.step.simple.SimpleExitCodeExceptionClassifier;
 import org.springframework.batch.io.exception.BatchCriticalException;
@@ -57,12 +56,11 @@ public class DefaultJobExecutor implements JobExecutor {
 	
 	private ExitCodeExceptionClassifier exceptionClassifier = new SimpleExitCodeExceptionClassifier();
 
-	public ExitStatus run(JobConfiguration configuration, JobExecutionContext jobExecutionContext)
+	public ExitStatus run(JobConfiguration configuration, JobExecution jobExecution)
 			throws BatchCriticalException {
 
-		JobInstance job = jobExecutionContext.getJob();
-		JobExecution jobExecution = jobExecutionContext.getJobExecution();
-		updateStatus(jobExecutionContext, BatchStatus.STARTING);
+		JobInstance job = jobExecution.getJob();
+		updateStatus(jobExecution, BatchStatus.STARTING);
 
 		List steps = job.getSteps();
 
@@ -74,22 +72,22 @@ public class DefaultJobExecutor implements JobExecutor {
 				StepInstance step = (StepInstance) i.next();
 				StepConfiguration stepConfiguration = (StepConfiguration) j.next();
 				if (shouldStart(step, stepConfiguration)) {
-					updateStatus(jobExecutionContext, BatchStatus.STARTED);
+					updateStatus(jobExecution, BatchStatus.STARTED);
 					StepExecutor stepExecutor = stepExecutorFactory.getExecutor(stepConfiguration);
-					StepExecutionContext stepExecutionContext = new StepExecutionContext(jobExecutionContext, step);
-					status = stepExecutor.process(stepConfiguration, stepExecutionContext);
+					StepExecution stepExecution = new StepExecution(step, jobExecution);
+					status = stepExecutor.process(stepConfiguration, stepExecution);
 				}
 			}
 
-			updateStatus(jobExecutionContext, BatchStatus.COMPLETED);
+			updateStatus(jobExecution, BatchStatus.COMPLETED);
 		}
 		catch (StepInterruptedException e) {
-			updateStatus(jobExecutionContext, BatchStatus.STOPPED);
+			updateStatus(jobExecution, BatchStatus.STOPPED);
 			status = exceptionClassifier.classifyForExitCode(e);
 			rethrow(e);
 		}
 		catch (Throwable t) {
-			updateStatus(jobExecutionContext, BatchStatus.FAILED);
+			updateStatus(jobExecution, BatchStatus.FAILED);
 			status = exceptionClassifier.classifyForExitCode(t);
 			rethrow(t);
 		}
@@ -102,14 +100,13 @@ public class DefaultJobExecutor implements JobExecutor {
 		return status;
 	}
 
-	private void updateStatus(JobExecutionContext jobExecutionContext, BatchStatus status) {
-		JobInstance job = jobExecutionContext.getJob();
-		JobExecution jobExecution = jobExecutionContext.getJobExecution();
+	private void updateStatus(JobExecution jobExecution, BatchStatus status) {
+		JobInstance job = jobExecution.getJob();
 		jobExecution.setStatus(status);
 		job.setStatus(status);
 		jobRepository.update(job);
 		jobRepository.saveOrUpdate(jobExecution);
-		for (Iterator iter = jobExecutionContext.getStepContexts().iterator(); iter.hasNext();) {
+		for (Iterator iter = jobExecution.getStepContexts().iterator(); iter.hasNext();) {
 			RepeatContext context = (RepeatContext) iter.next();
 			context.setAttribute("JOB_STATUS", status);
 		}

execution/src/main/java/org/springframework/batch/execution/repository/dao/HibernateStepDao.java

@@ -30,10 +30,11 @@ import org.springframework.util.Assert;
 
 /**
  * It represents an implementation of {@link StepDao} functionality based
- * on the Hibernate ORM framework. Its advantage is the independency of implementation
+ * on the Hibernate ORM framework. Its advantage is the independence of implementation
  * on the underlying database.
  *
- * @author tomas.slanina
+ * @author Tomas Slanina
+ * @author Dave Syer
  */
 public class HibernateStepDao extends HibernateDaoSupport implements StepDao {
 	
@@ -45,9 +46,7 @@ public class HibernateStepDao extends HibernateDaoSupport implements StepDao {
 		Assert.notNull(job, "Job cannot be null.");
 		Assert.notNull(stepName, "StepName cannot be null.");
 		
-		StepInstance step = new StepInstance();
-		step.setName(stepName);
-		step.setJob(job);
+		StepInstance step = new StepInstance(job, stepName);
 		
 		Long stepId = (Long)getHibernateTemplate().save(step);
 		
@@ -78,20 +77,20 @@ public class HibernateStepDao extends HibernateDaoSupport implements StepDao {
 	}
 
 	/**
-	 * @see StepDao#findSteps(Long)
+	 * @see StepDao#findSteps(JobInstance)
 	 * 
 	 * Hibernate is asked to get all jobs that matches criteria. Afterwards, result is mapped into domain objects.
 	 * It should be noted that restart data must be requested separately.
 	 * 
 	 */
-	public List findSteps(final Long jobId) {
+	public List findSteps(final JobInstance job) {
 		
-		Assert.notNull(jobId, "JobId cannot be null.");
+		Assert.notNull(job, "JobId cannot be null.");
 		
 		List list = this.getHibernateTemplate().executeFind(new HibernateCallback() {
             public Object doInHibernate(Session session) {
                 Criteria criteria = session.createCriteria(StepInstance.class);
-                criteria.add(Expression.eq("job.id", jobId));
+                criteria.add(Expression.eq("job.id", job.getId()));
                 return criteria.list();
             }
         });
@@ -105,7 +104,7 @@ public class HibernateStepDao extends HibernateDaoSupport implements StepDao {
 	public int getStepExecutionCount(final Long stepId) {
 		Long result = (Long) this.getHibernateTemplate().execute(new HibernateCallback() {
             public Object doInHibernate(Session session) {
-                return session.createQuery("select count(id) from StepExecution s where s.stepId = :stepId")
+                return session.createQuery("select count(id) from StepExecution s where s.step.id = :stepId")
                 		.setLong("stepId", stepId.longValue())
                 		.uniqueResult();
             }
@@ -151,17 +150,14 @@ public class HibernateStepDao extends HibernateDaoSupport implements StepDao {
 		getHibernateTemplate().update(stepExecution);
 	}
 	
-	public List findStepExecutions(StepInstance step) {
+	public List findStepExecutions(final StepInstance step) {
 		
 		Assert.notNull(step, "Step cannot be null.");
-		Assert.notNull(step.getId(), "Step id cannot be null.");
-		
-		final Long stepId = step.getId();
-		
+				
 		List results = this.getHibernateTemplate().executeFind(new HibernateCallback() {
             public Object doInHibernate(Session session) {
                 Criteria criteria = session.createCriteria(StepExecution.class);
-                criteria.add(Expression.eq("stepId", stepId));
+                criteria.add(Expression.eq("step.id", step.getId()));
                 return criteria.list();
             }
         });

execution/src/main/java/org/springframework/batch/execution/repository/dao/MapStepDao.java

@@ -48,9 +48,7 @@ public class MapStepDao implements StepDao {
 	}
 
 	public StepInstance createStep(JobInstance job, String stepName) {
-		StepInstance step = new StepInstance(new Long(currentId++));
-		step.setName(stepName);
-		step.setJob(job);
+		StepInstance step = new StepInstance(job, stepName, new Long(currentId++));
 		Set steps = (Set) stepsByJobId.get(job.getId());
 		if (steps==null) {
 			steps = TransactionAwareProxyFactory.createTransactionalSet();
@@ -74,8 +72,8 @@ public class MapStepDao implements StepDao {
 		return null;
 	}
 
-	public List findSteps(Long jobId) {
-		Set steps = (Set) stepsByJobId.get(jobId);
+	public List findSteps(JobInstance job) {
+		Set steps = (Set) stepsByJobId.get(job.getId());
 		if (steps==null) {
 			return new ArrayList();
 		}

execution/src/main/java/org/springframework/batch/execution/repository/dao/SqlJobDao.java

@@ -71,14 +71,11 @@ public class SqlJobDao implements JobDao, InitializingBean {
 	private static final String UPDATE_JOB_EXECUTION = "UPDATE %PREFIX%JOB_EXECUTION set START_TIME = ?, END_TIME = ?, "
 			+ " STATUS = ?, EXIT_CODE = ? where ID = ?";
 
-	private static final String SAVE_JOB_EXECUTION = "INSERT into %PREFIX%JOB_EXECUTION(ID, JOB_ID, START_TIME, " +
-			"END_TIME, STATUS, EXIT_CODE) values (?, ?, ?, ?, ?, ?)";
+	private static final String SAVE_JOB_EXECUTION = "INSERT into %PREFIX%JOB_EXECUTION(ID, JOB_ID, START_TIME, "
+			+ "END_TIME, STATUS, EXIT_CODE) values (?, ?, ?, ?, ?, ?)";
 
 	private static final String CHECK_JOB_EXECUTION_EXISTS = "SELECT COUNT(*) FROM %PREFIX%JOB_EXECUTION WHERE ID=?";
 
-	private static final String FIND_JOB_EXECUTIONS = "SELECT ID, START_TIME, END_TIME, STATUS from %PREFIX%JOB_EXECUTION"
-			+ " where JOB_ID = ?";
-
 	private JdbcTemplate jdbcTemplate;
 
 	private DataFieldMaxValueIncrementer jobIncrementer;
@@ -144,7 +141,8 @@ public class SqlJobDao implements JobDao, InitializingBean {
 		RowMapper rowMapper = new RowMapper() {
 			public Object mapRow(ResultSet rs, int rowNum) throws SQLException {
 
-				JobInstance job = new JobInstance(jobIdentifier, new Long(rs.getLong(1)));
+				JobInstance job = new JobInstance(jobIdentifier, new Long(rs
+						.getLong(1)));
 				job.setStatus(BatchStatus.getStatus(rs.getString(2)));
 
 				return job;
@@ -189,7 +187,7 @@ public class SqlJobDao implements JobDao, InitializingBean {
 		Object[] parameters = new Object[] { jobExecution.getId(),
 				jobExecution.getJobId(), jobExecution.getStartTime(),
 				jobExecution.getEndTime(), jobExecution.getStatus().toString(),
-				jobExecution.getExitCode()};
+				jobExecution.getExitCode() };
 		jdbcTemplate.update(getSaveJobExecutionQuery(), parameters);
 	}
 
@@ -239,71 +237,51 @@ public class SqlJobDao implements JobDao, InitializingBean {
 
 		Object[] parameters = new Object[] { jobId };
 
-		return jdbcTemplate.queryForInt(getJobExecutionCountQuery(), parameters);
+		return jdbcTemplate
+				.queryForInt(getJobExecutionCountQuery(), parameters);
 	}
 
-	public List findJobExecutions(JobInstance job) {
+	public List findJobExecutions(final JobInstance job) {
 
 		Assert.notNull(job, "Job cannot be null.");
 		Assert.notNull(job.getId(), "Job Id cannot be null.");
 
-		final Long jobId = job.getId();
-
-		RowMapper rowMapper = new RowMapper() {
-
-			public Object mapRow(ResultSet rs, int rowNum) throws SQLException {
-
-				JobExecution jobExecution = new JobExecution(jobId);
-				jobExecution.setId(new Long(rs.getLong(1)));
-				jobExecution.setStartTime(rs.getTimestamp(2));
-				jobExecution.setEndTime(rs.getTimestamp(3));
-				jobExecution.setStatus(BatchStatus.getStatus(rs.getString(4)));
-
-				return jobExecution;
-			}
-
-		};
-
-		return jdbcTemplate.query(getFindJobExecutionsQuery(), new Object[] { jobId },
-				rowMapper);
+		return jdbcTemplate.query(getQuery(JobExecutionRowMapper.FIND_JOB_EXECUTIONS),
+				new Object[] { job.getId() }, new JobExecutionRowMapper(job));
 	}
 
 	private String getQuery(String base) {
 		return StringUtils.replace(base, "%PREFIX%", tablePrefix);
 	}
-	
+
 	private String getCreateJobQuery() {
 		return getQuery(CREATE_JOB);
 	}
-	
+
 	private String getFindJobsQuery() {
 		return getQuery(FIND_JOBS);
 	}
-	
+
 	private String getUpdateJobQuery() {
 		return getQuery(UPDATE_JOB);
 	}
-	
+
 	private String getSaveJobExecutionQuery() {
 		return getQuery(SAVE_JOB_EXECUTION);
 	}
-	
+
 	private String getUpdateJobExecutionQuery() {
 		return getQuery(UPDATE_JOB_EXECUTION);
 	}
-	
+
 	private String getCheckJobExecutionExistsQuery() {
 		return getQuery(CHECK_JOB_EXECUTION_EXISTS);
 	}
-	
+
 	private String getJobExecutionCountQuery() {
 		return getQuery(GET_JOB_EXECUTION_COUNT);
 	}
 
-	private String getFindJobExecutionsQuery() {
-		return getQuery(FIND_JOB_EXECUTIONS);
-	}
-
 	public void setJdbcTemplate(JdbcTemplate jdbcTemplate) {
 		this.jdbcTemplate = jdbcTemplate;
 	}
@@ -386,4 +364,35 @@ public class SqlJobDao implements JobDao, InitializingBean {
 		return new ScheduledJobIdentifier(jobIdentifier.getName());
 	}
 
+	/**
+	 * Re-usable mapper for {@link JobExecution} instances.
+	 * @author Dave Syer
+	 *
+	 */
+	public static class JobExecutionRowMapper implements RowMapper {
+		
+		public static final String FIND_JOB_EXECUTIONS = "SELECT ID, START_TIME, END_TIME, STATUS from %PREFIX%JOB_EXECUTION"
+			+ " where JOB_ID = ?";
+
+		public static final String GET_JOB_EXECUTION = "SELECT ID, START_TIME, END_TIME, STATUS from %PREFIX%JOB_EXECUTION"
+			+ " where ID = ?";
+
+		private JobInstance job;
+
+		public JobExecutionRowMapper(JobInstance job) {
+			super();
+			this.job = job;
+		}
+
+		public Object mapRow(ResultSet rs, int rowNum) throws SQLException {
+			JobExecution jobExecution = new JobExecution(job);
+			jobExecution.setId(new Long(rs.getLong(1)));
+			jobExecution.setStartTime(rs.getTimestamp(2));
+			jobExecution.setEndTime(rs.getTimestamp(3));
+			jobExecution.setStatus(BatchStatus.getStatus(rs.getString(4)));
+			return jobExecution;
+		}
+
+	}
+
 }

execution/src/main/java/org/springframework/batch/execution/repository/dao/SqlStepDao.java

@@ -22,10 +22,12 @@ import java.util.List;
 import java.util.Properties;
 
 import org.springframework.batch.core.domain.BatchStatus;
+import org.springframework.batch.core.domain.JobExecution;
 import org.springframework.batch.core.domain.JobInstance;
 import org.springframework.batch.core.domain.StepExecution;
 import org.springframework.batch.core.domain.StepInstance;
 import org.springframework.batch.core.repository.NoSuchBatchDomainObjectException;
+import org.springframework.batch.execution.repository.dao.SqlJobDao.JobExecutionRowMapper;
 import org.springframework.batch.restart.GenericRestartData;
 import org.springframework.batch.restart.RestartData;
 import org.springframework.batch.support.PropertiesConverter;
@@ -37,7 +39,7 @@ import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
 /**
- * Sql implementation of StepDao. Uses Sequences (via Spring's
+ * Sql implementation of {@link StepDao}. Uses Sequences (via Spring's
  * 
  * @link DataFieldMaxValueIncrementer abstraction) to create all Step and
  *       StepExecution primary keys before inserting a new row. All objects are
@@ -82,6 +84,8 @@ public class SqlStepDao implements StepDao, InitializingBean {
 
 	private JdbcOperations jdbcTemplate;
 
+	private JobDao jobDao;
+
 	private DataFieldMaxValueIncrementer stepIncrementer;
 
 	private DataFieldMaxValueIncrementer stepExecutionIncrementer;
@@ -100,6 +104,16 @@ public class SqlStepDao implements StepDao, InitializingBean {
 		this.tablePrefix = tablePrefix;
 	}
 
+	/**
+	 * Injection setter for job dao. Used to save {@link JobExecution}
+	 * instances.
+	 * 
+	 * @param jobDao a {@link JobDao}
+	 */
+	public void setJobDao(JobDao jobDao) {
+		this.jobDao = jobDao;
+	}
+
 	/**
 	 * Find one step for given job and stepName. A RowMapper is used to map each
 	 * row returned to a step object. If none are found, the list will be empty
@@ -141,7 +155,6 @@ public class SqlStepDao implements StepDao, InitializingBean {
 			return null;
 		} else if (steps.size() == 1) {
 			StepInstance step = (StepInstance) steps.get(0);
-			step.setName(stepName);
 			return step;
 		} else {
 			// This error will likely never be thrown, because there should
@@ -155,7 +168,7 @@ public class SqlStepDao implements StepDao, InitializingBean {
 	}
 
 	/**
-	 * @see StepDao#findSteps(Long)
+	 * @see StepDao#findSteps(JobInstance)
 	 * 
 	 * Sql implementation which uses a RowMapper to populate a list of all rows
 	 * in the step table with the same JOB_ID.
@@ -163,18 +176,18 @@ public class SqlStepDao implements StepDao, InitializingBean {
 	 * @throws IllegalArgumentException
 	 *             if jobId is null.
 	 */
-	public List findSteps(Long jobId) {
+	public List findSteps(final JobInstance job) {
 
-		Assert.notNull(jobId, "JobId cannot be null.");
+		Assert.notNull(job, "Job cannot be null.");
 
-		Object[] parameters = new Object[] { jobId };
+		Object[] parameters = new Object[] { job.getId() };
 
 		RowMapper rowMapper = new RowMapper() {
 
 			public Object mapRow(ResultSet rs, int rowNum) throws SQLException {
 
-				StepInstance step = new StepInstance(new Long(rs.getLong(1)));
-				step.setName(rs.getString(2));
+				StepInstance step = new StepInstance(job, rs.getString(2),
+						new Long(rs.getLong(1)));
 				String status = rs.getString(3);
 				step.setStatus(BatchStatus.getStatus(status));
 				step.setRestartData(new GenericRestartData(PropertiesConverter
@@ -204,9 +217,7 @@ public class SqlStepDao implements StepDao, InitializingBean {
 		Object[] parameters = new Object[] { stepId, job.getId(), stepName };
 		jdbcTemplate.update(getQuery(CREATE_STEP), parameters);
 
-		StepInstance step = new StepInstance(stepId);
-		step.setJob(job);
-		step.setName(stepName);
+		StepInstance step = new StepInstance(job, stepName, stepId);
 		return step;
 	}
 
@@ -245,6 +256,8 @@ public class SqlStepDao implements StepDao, InitializingBean {
 
 		validateStepExecution(stepExecution);
 
+		cascadeJobExecution(stepExecution.getJobExecution());
+
 		stepExecution.setId(new Long(stepExecutionIncrementer.nextLongValue()));
 		Object[] parameters = new Object[] {
 				stepExecution.getId(),
@@ -263,6 +276,14 @@ public class SqlStepDao implements StepDao, InitializingBean {
 
 	}
 
+	private void cascadeJobExecution(JobExecution jobExecution) {
+		if (jobExecution.getId() != null) {
+			// assume already saved...
+			return;
+		}
+		jobDao.save(jobExecution);
+	}
+
 	/**
 	 * @see StepDao#update(StepExecution)
 	 */
@@ -289,8 +310,7 @@ public class SqlStepDao implements StepDao, InitializingBean {
 				stepExecution.getTaskCount(),
 				PropertiesConverter.propertiesToString(stepExecution
 						.getStatistics()), stepExecution.getExitCode(),
-				stepExecution.getExitDescription(),
-				stepExecution.getId() };
+				stepExecution.getExitDescription(), stepExecution.getId() };
 		jdbcTemplate.update(getQuery(UPDATE_STEP_EXECUTION), parameters);
 
 	}
@@ -313,19 +333,21 @@ public class SqlStepDao implements StepDao, InitializingBean {
 	 * @throws NoSuchBatchDomainObjectException
 	 *             if more than one step execution is returned.
 	 */
-	public List findStepExecutions(StepInstance step) {
+	public List findStepExecutions(final StepInstance step) {
 
 		Assert.notNull(step, "Step cannot be null.");
 		Assert.notNull(step.getId(), "Step id cannot be null.");
 
-		final Long stepId = step.getId();
-
 		RowMapper rowMapper = new RowMapper() {
 			public Object mapRow(ResultSet rs, int rowNum) throws SQLException {
 
-				StepExecution stepExecution = new StepExecution(stepId,
-						new Long(rs.getLong(2)));
-				stepExecution.setId(new Long(rs.getLong(1)));
+				JobExecution jobExecution = (JobExecution) jdbcTemplate
+						.queryForObject(
+								getQuery(JobExecutionRowMapper.GET_JOB_EXECUTION),
+								new Object[] { new Long(rs.getLong(2)) },
+								new JobExecutionRowMapper(step.getJob()));
+				StepExecution stepExecution = new StepExecution(step,
+						jobExecution, new Long(rs.getLong(1)));
 				stepExecution.setStartTime(rs.getTimestamp(3));
 				stepExecution.setEndTime(rs.getTimestamp(4));
 				stepExecution.setStatus(BatchStatus.getStatus(rs.getString(5)));
@@ -340,7 +362,7 @@ public class SqlStepDao implements StepDao, InitializingBean {
 		};
 
 		return jdbcTemplate.query(getQuery(FIND_STEP_EXECUTIONS),
-				new Object[] { stepId }, rowMapper);
+				new Object[] { step.getId() }, rowMapper);
 
 	}
 

execution/src/main/java/org/springframework/batch/execution/repository/dao/StepDao.java

@@ -28,13 +28,13 @@ import org.springframework.batch.core.domain.StepInstance;
  * TODO: Add java doc.
  * 
  * @author Lucas Ward
- *
+ * 
  */
 public interface StepDao {
 
 	/**
-	 * Find a step with the given JobId and Step Name. Return null if none
-	 * are found.
+	 * Find a step with the given JobId and Step Name. Return null if none are
+	 * found.
 	 * 
 	 * @param jobId
 	 * @param stepName
@@ -43,15 +43,17 @@ public interface StepDao {
 	public StepInstance findStep(JobInstance job, String stepName);
 
 	/**
-	 * Find all steps with the given Job ID.
+	 * Find all steps with the given Job.
 	 * 
-	 * @param jobId
-	 * @return
+	 * @param job
+	 *            the job to use as a search key
+	 * @return list of {@link StepInstance}
 	 */
-	public List findSteps(Long jobId);
+	public List findSteps(JobInstance job);
 
 	/**
 	 * Create a step for the given Step Name and Job Id.
+	 * 
 	 * @param job
 	 * @param stepName
 	 * 
@@ -91,15 +93,17 @@ public interface StepDao {
 	 * Return the count of StepExecutions with the given StepId.
 	 * 
 	 * @param stepId
-	 * @return
+	 * @return the number of step executions for this step TODO: change
+	 *         signature to search by {@link StepInstance}
 	 */
 	public int getStepExecutionCount(Long stepId);
-	
+
 	/**
-	 * Return all StepExecutions for the given step. 
+	 * Return all StepExecutions for the given step.
 	 * 
-	 * @param id
-	 * @return list of stepExecutions
+	 * @param step
+	 *            the step to use as a search key
+	 * @return list of stepExecutions 
 	 */
 	public List findStepExecutions(StepInstance step);
 }

execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java

@@ -28,11 +28,9 @@ import org.springframework.batch.core.executor.ExitCodeExceptionClassifier;
 import org.springframework.batch.core.executor.StepExecutor;
 import org.springframework.batch.core.executor.StepInterruptedException;
 import org.springframework.batch.core.repository.JobRepository;
-import org.springframework.batch.core.runtime.JobExecutionContext;
-import org.springframework.batch.core.runtime.StepExecutionContext;
 import org.springframework.batch.core.tasklet.Tasklet;
-import org.springframework.batch.execution.scope.StepScope;
 import org.springframework.batch.execution.scope.SimpleStepContext;
+import org.springframework.batch.execution.scope.StepScope;
 import org.springframework.batch.execution.scope.StepSynchronizationManager;
 import org.springframework.batch.io.exception.BatchCriticalException;
 import org.springframework.batch.repeat.ExitStatus;
@@ -75,7 +73,7 @@ public class SimpleStepExecutor implements StepExecutor {
 
 	/**
 	 * Key placed in step scope context to identify the
-	 * {@link StepExecutionContext}.
+	 * {@link StepExecution}.
 	 */
 	public static final String STEP_KEY = "STEP";
 
@@ -149,16 +147,15 @@ public class SimpleStepExecutor implements StepExecutor {
 	 * @throws StepInterruptedException if the step or a chunk is interrupted
 	 * @throws RuntimeException if there is an exception during a chunk
 	 * execution
-	 * @see StepExecutor#process(StepConfiguration, StepExecutionContext)
+	 * @see StepExecutor#process(StepConfiguration, StepExecution)
 	 */
-	public ExitStatus process(final StepConfiguration configuration, final StepExecutionContext stepExecutionContext)
+	public ExitStatus process(final StepConfiguration configuration, final StepExecution stepExecution)
 			throws BatchCriticalException, StepInterruptedException {
 
-		final StepInstance step = stepExecutionContext.getStep();
+		final StepInstance step = stepExecution.getStep();
 		boolean isRestart = step.getStepExecutionCount() > 0 ? true : false;
 		Assert.notNull(step);
 
-		final StepExecution stepExecution = stepExecutionContext.getStepExecution();
 		final Tasklet module = configuration.getTasklet();
 		step.setStepExecution(stepExecution);
 
@@ -168,7 +165,7 @@ public class SimpleStepExecutor implements StepExecutor {
 
 		try {
 			stepExecution.setStartTime(new Timestamp(System.currentTimeMillis()));
-			updateStatus(stepExecutionContext, BatchStatus.STARTED);
+			updateStatus(stepExecution, BatchStatus.STARTED);
 
 			final boolean saveRestartData = ((AbstractStepConfiguration) configuration).isSaveRestartData();
 
@@ -180,18 +177,18 @@ public class SimpleStepExecutor implements StepExecutor {
 
 				public ExitStatus doInIteration(final RepeatContext context) throws Exception {
 
-					stepExecutionContext.getJobExecutionContext().registerStepContext(context);
+					stepExecution.getJobExecution().registerStepContext(context);
 					context.registerDestructionCallback("STEP_EXECUTION_CONTEXT_CALLBACK", new Runnable() {
 						public void run() {
-							stepExecutionContext.getJobExecutionContext().unregisterStepContext(context);
+							stepExecution.getJobExecution().unregisterStepContext(context);
 						}
 					});
-					stepScopeContext.setJobIdentifier(stepExecutionContext.getJobExecutionContext().getJobIdentifier());
-					context.setAttribute(StepScope.ID_KEY, stepExecutionContext.getJobExecutionContext()
+					stepScopeContext.setJobIdentifier(stepExecution.getJobExecution().getJobIdentifier());
+					context.setAttribute(StepScope.ID_KEY, stepExecution.getJobExecution()
 							.getJobIdentifier());
 					// Mark the context as a step context as a hint to scope
 					// implementations.
-					context.setAttribute(STEP_KEY, stepExecutionContext);
+					context.setAttribute(STEP_KEY, stepExecution);
 					// Add the step execution as an attribute so monitoring
 					// clients can see it.
 					context.setAttribute(STEP_EXECUTION_KEY, stepExecution);
@@ -208,7 +205,7 @@ public class SimpleStepExecutor implements StepExecutor {
 									ExitStatus result;
 
 									try {
-										result = processChunk(configuration, stepExecutionContext);
+										result = processChunk(configuration, stepExecution);
 									}
 									catch (Throwable t) {
 										/*
@@ -250,7 +247,7 @@ public class SimpleStepExecutor implements StepExecutor {
 
 			});
 
-			updateStatus(stepExecutionContext, BatchStatus.COMPLETED);
+			updateStatus(stepExecution, BatchStatus.COMPLETED);
 			return status;
 		}
 		catch (RuntimeException e) {
@@ -259,11 +256,11 @@ public class SimpleStepExecutor implements StepExecutor {
 			status = exceptionClassifier.classifyForExitCode(e);
 			stepExecution.setException(e);
 			if (e.getCause() instanceof StepInterruptedException) {
-				updateStatus(stepExecutionContext, BatchStatus.STOPPED);
+				updateStatus(stepExecution, BatchStatus.STOPPED);
 				throw (StepInterruptedException) e.getCause();
 			}
 			else {
-				updateStatus(stepExecutionContext, BatchStatus.FAILED);
+				updateStatus(stepExecution, BatchStatus.FAILED);
 				throw e;
 			}
 
@@ -294,14 +291,13 @@ public class SimpleStepExecutor implements StepExecutor {
 	 * @param stepExecution the current stepExecution
 	 * @param status the status to set
 	 */
-	private void updateStatus(StepExecutionContext stepExecutionContext, BatchStatus status) {
-		StepInstance step = stepExecutionContext.getStep();
-		StepExecution stepExecution = stepExecutionContext.getStepExecution();
+	private void updateStatus(StepExecution stepExecution, BatchStatus status) {
+		StepInstance step = stepExecution.getStep();
 		stepExecution.setStatus(status);
 		step.setStatus(status);
 		jobRepository.update(step);
 		jobRepository.saveOrUpdate(stepExecution);
-		for (Iterator iter = stepExecutionContext.getJobExecutionContext().getStepContexts().iterator(); iter.hasNext();) {
+		for (Iterator iter = stepExecution.getJobExecution().getStepContexts().iterator(); iter.hasNext();) {
 			RepeatContext context = (RepeatContext) iter.next();
 			context.setAttribute("JOB_STATUS", status);
 		}
@@ -314,24 +310,24 @@ public class SimpleStepExecutor implements StepExecutor {
 	 * transaction.
 	 *
 	 * @param configuration the current step configuration
-	 * @param stepExecutionContext the current step, containing the
+	 * @param stepExecution the current step, containing the
 	 * {@link Tasklet} with the business logic.
 	 * @return true if there is more data to process.
 	 */
 	protected final ExitStatus processChunk(final StepConfiguration configuration,
-			final StepExecutionContext stepExecutionContext) {
+			final StepExecution stepExecution) {
 		return chunkOperations.iterate(new RepeatCallback() {
 			public ExitStatus doInIteration(final RepeatContext context) throws Exception {
-				stepExecutionContext.getJobExecutionContext().registerChunkContext(context);
+				stepExecution.getJobExecution().registerChunkContext(context);
 				context.registerDestructionCallback("CHUNK_EXECUTION_CONTEXT_CALLBACK", new Runnable() {
 					public void run() {
-						stepExecutionContext.getJobExecutionContext().unregisterStepContext(context);
+						stepExecution.getJobExecution().unregisterStepContext(context);
 					}
 				});
 				// check for interruption before each item as well
 				interruptionPolicy.checkInterrupted(context);
-				ExitStatus exitStatus = doTaskletProcessing(configuration.getTasklet(), stepExecutionContext.getStep());
-				stepExecutionContext.getStepExecution().incrementTaskCount();
+				ExitStatus exitStatus = doTaskletProcessing(configuration.getTasklet(), stepExecution.getStep());
+				stepExecution.incrementTaskCount();
 				// check for interruption after each item as well
 				interruptionPolicy.checkInterrupted(context);
 				return exitStatus;