From 0824f1aeb32064c343f46bdaf1c3e2ced64a8e57 Mon Sep 17 00:00:00 2001
From: robokaso <robokaso@2e8e966d-6415-0410-94b9-dc4e65435d03>
Date: Mon, 11 Feb 2008 16:15:24 +0000
Subject: [PATCH] IN PROGRESS - issue BATCH-340: Refactor JobRepository for
 greater clarity and consistency.
 http://jira.springframework.org/browse/BATCH-340

StepDao split into StepInstanceDao and StepExecutionDao. AbstractJdbcBatchMetadataDao created to remove the duplications.
---
 .../repository/SimpleJobRepository.java       |   34 +-
 .../dao/AbstractJdbcBatchMetadataDao.java     |   56 +
 .../execution/repository/dao/JdbcJobDao.java  |   55 +-
 ...StepDao.java => JdbcStepExecutionDao.java} | 1233 +++++++----------
 .../repository/dao/JdbcStepInstanceDao.java   |  183 +++
 .../execution/repository/dao/StepDao.java     |  120 +-
 .../repository/dao/StepExecutionDao.java      |   86 ++
 .../repository/dao/StepInstanceDao.java       |   46 +
 .../execution/job/simple/SimpleJobTests.java  |   12 +-
 .../execution/launch/SimpleJobTests.java      |    2 +-
 .../repository/SimpleJobRepositoryTests.java  |    2 +-
 .../repository/dao/AbstractStepDaoTests.java  |   76 +-
 .../dao/JdbcStepDaoPrefixTests.java           |   58 +-
 .../repository/dao/JdbcStepDaoTests.java      |    8 +-
 .../step/simple/SimpleStepExecutorTests.java  |    2 +-
 .../simple/StepExecutorInterruptionTests.java |    2 +-
 .../execution/repository/dao/sql-dao-test.xml |   11 +-
 .../resources/simple-container-definition.xml |    1 +
 .../resources/simple-container-definition.xml |   19 +-
 19 files changed, 1020 insertions(+), 986 deletions(-)
 create mode 100644 spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/AbstractJdbcBatchMetadataDao.java
 rename spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/{JdbcStepDao.java => JdbcStepExecutionDao.java} (52%)
 create mode 100644 spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/JdbcStepInstanceDao.java
 create mode 100644 spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/StepExecutionDao.java
 create mode 100644 spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/StepInstanceDao.java

diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/SimpleJobRepository.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/SimpleJobRepository.java
index dd628c7f3..2386b62b4 100644
--- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/SimpleJobRepository.java
+++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/SimpleJobRepository.java
@@ -32,7 +32,8 @@ import org.springframework.batch.core.repository.BatchRestartException;
 import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException;
 import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.execution.repository.dao.JobDao;
-import org.springframework.batch.execution.repository.dao.StepDao;
+import org.springframework.batch.execution.repository.dao.StepExecutionDao;
+import org.springframework.batch.execution.repository.dao.StepInstanceDao;
 import org.springframework.batch.item.ExecutionAttributes;
 import org.springframework.transaction.annotation.Isolation;
 import org.springframework.util.Assert;
@@ -56,7 +57,9 @@ public class SimpleJobRepository implements JobRepository {
 
 	private JobDao jobDao;
 
-	private StepDao stepDao;
+	private StepInstanceDao stepInstanceDao;
+
+	private StepExecutionDao stepExecutionDao;
 
 	/**
 	 * Provide default constructor with low visibility in case user wants to use
@@ -65,10 +68,11 @@ public class SimpleJobRepository implements JobRepository {
 	SimpleJobRepository() {
 	}
 
-	public SimpleJobRepository(JobDao jobDao, StepDao stepDao) {
+	public SimpleJobRepository(JobDao jobDao, StepInstanceDao stepInstanceDao, StepExecutionDao stepExecutionDao) {
 		super();
 		this.jobDao = jobDao;
-		this.stepDao = stepDao;
+		this.stepInstanceDao = stepInstanceDao;
+		this.stepExecutionDao = stepExecutionDao;
 	}
 
 	/**
@@ -85,8 +89,8 @@ public class SimpleJobRepository implements JobRepository {
 	 * There are two ways in which the method determines if a job should be
 	 * created or an existing one should be returned. The first is
 	 * restartability. The {@link JobSupport} restartable property will be
-	 * checked first. If it is false, a new job will be created, regardless
-	 * of whether or not one exists. If it is true, the {@link JobDao} will be
+	 * checked first. If it is false, a new job will be created, regardless of
+	 * whether or not one exists. If it is true, the {@link JobDao} will be
 	 * checked to determine if the job already exists, if it does, it's steps
 	 * will be populated (there must be at least 1) and a new
 	 * {@link JobExecution} will be returned. If no job is found, a new one will
@@ -256,13 +260,13 @@ public class SimpleJobRepository implements JobRepository {
 
 		if (stepExecution.getId() == null) {
 			// new execution, obtain id and insert
-			stepDao.saveStepExecution(stepExecution);
-			stepDao.saveExecutionAttributes(stepExecution.getId(), stepExecution.getExecutionAttributes());
+			stepExecutionDao.saveStepExecution(stepExecution);
+			stepExecutionDao.saveExecutionAttributes(stepExecution.getId(), stepExecution.getExecutionAttributes());
 		}
 		else {
 			// existing execution, update
-			stepDao.updateStepExecution(stepExecution);
-			stepDao.updateExecutionAttributes(stepExecution.getId(), stepExecution.getExecutionAttributes());
+			stepExecutionDao.updateStepExecution(stepExecution);
+			stepExecutionDao.updateExecutionAttributes(stepExecution.getId(), stepExecution.getExecutionAttributes());
 		}
 	}
 
@@ -278,7 +282,7 @@ public class SimpleJobRepository implements JobRepository {
 		Assert.notNull(step.getId(), "Step cannot be updated if it's ID is null.  It must be obtained"
 				+ "from SimpleJobRepository.findOrCreateJob to be considered valid.");
 
-		stepDao.updateStepInstance(step);
+		stepInstanceDao.updateStepInstance(step);
 
 	}
 
@@ -304,7 +308,7 @@ public class SimpleJobRepository implements JobRepository {
 		Iterator i = steps.iterator();
 		while (i.hasNext()) {
 			Step step = (Step) i.next();
-			StepInstance stepInstance = stepDao.createStepInstance(job, step.getName());
+			StepInstance stepInstance = stepInstanceDao.createStepInstance(job, step.getName());
 			stepInstances.add(stepInstance);
 		}
 
@@ -320,15 +324,15 @@ public class SimpleJobRepository implements JobRepository {
 		while (i.hasNext()) {
 
 			Step stepConfiguration = (Step) i.next();
-			StepInstance stepInstance = stepDao.findStepInstance(jobInstance, stepConfiguration.getName());
+			StepInstance stepInstance = stepInstanceDao.findStepInstance(jobInstance, stepConfiguration.getName());
 			if (stepInstance != null) {
 
 				if (stepInstance.getLastExecution() != null) {
-					ExecutionAttributes executionAttributes = stepDao.findExecutionAttributes(stepInstance
+					ExecutionAttributes executionAttributes = stepExecutionDao.findExecutionAttributes(stepInstance
 							.getLastExecution().getId());
 					stepInstance.getLastExecution().setExecutionAttributes(executionAttributes);
 				}
-				stepInstance.setStepExecutionCount(stepDao.getStepExecutionCount(stepInstance));
+				stepInstance.setStepExecutionCount(stepExecutionDao.getStepExecutionCount(stepInstance));
 				stepInstances.add(stepInstance);
 			}
 		}
diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/AbstractJdbcBatchMetadataDao.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/AbstractJdbcBatchMetadataDao.java
new file mode 100644
index 000000000..0a2653281
--- /dev/null
+++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/AbstractJdbcBatchMetadataDao.java
@@ -0,0 +1,56 @@
+package org.springframework.batch.execution.repository.dao;
+
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.jdbc.core.JdbcOperations;
+import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
+
+/**
+ * Encapsulates common functionality needed by JDBC batch metadata DAOs -
+ * provides jdbcTemplate for subclasses and handles table prefixes.
+ * 
+ * @author Robert Kasanicky
+ */
+public abstract class AbstractJdbcBatchMetadataDao implements InitializingBean {
+
+	/**
+	 * Default value for the table prefix property.
+	 */
+	public static final String DEFAULT_TABLE_PREFIX = "BATCH_";
+
+	private String tablePrefix = DEFAULT_TABLE_PREFIX;
+
+	private JdbcOperations jdbcTemplate;
+
+	protected String getQuery(String base) {
+		return StringUtils.replace(base, "%PREFIX%", tablePrefix);
+	}
+
+	/**
+	 * Public setter for the table prefix property. This will be prefixed to all
+	 * the table names before queries are executed. Defaults to
+	 * {@value #DEFAULT_TABLE_PREFIX}.
+	 * 
+	 * @param tablePrefix the tablePrefix to set
+	 */
+	public void setTablePrefix(String tablePrefix) {
+		this.tablePrefix = tablePrefix;
+	}
+
+	public void setJdbcTemplate(JdbcOperations jdbcTemplate) {
+		this.jdbcTemplate = jdbcTemplate;
+	}
+
+	protected String getTablePrefix() {
+		return tablePrefix;
+	}
+
+	protected JdbcOperations getJdbcTemplate() {
+		return jdbcTemplate;
+	}
+
+	public void afterPropertiesSet() throws Exception {
+		Assert.notNull(jdbcTemplate);
+	}
+
+}
diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/JdbcJobDao.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/JdbcJobDao.java
index 7b0e3debd..c18e955f0 100644
--- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/JdbcJobDao.java
+++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/JdbcJobDao.java
@@ -35,11 +35,9 @@ import org.springframework.batch.core.repository.NoSuchBatchDomainObjectExceptio
 import org.springframework.batch.repeat.ExitStatus;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.dao.IncorrectResultSizeDataAccessException;
-import org.springframework.jdbc.core.JdbcOperations;
 import org.springframework.jdbc.core.RowMapper;
 import org.springframework.jdbc.support.incrementer.DataFieldMaxValueIncrementer;
 import org.springframework.util.Assert;
-import org.springframework.util.StringUtils;
 
 /**
  * Jdbc implementation of {@link JobDao}. Uses sequences (via Spring's
@@ -53,7 +51,7 @@ import org.springframework.util.StringUtils;
  * @author Lucas Ward
  * @author Dave Syer
  */
-public class JdbcJobDao implements JobDao, InitializingBean {
+public class JdbcJobDao extends AbstractJdbcBatchMetadataDao implements JobDao, InitializingBean {
 
 	private static final String CHECK_JOB_EXECUTION_EXISTS = "SELECT COUNT(*) FROM %PREFIX%JOB_EXECUTION WHERE JOB_EXECUTION_ID = ?";
 
@@ -63,11 +61,6 @@ public class JdbcJobDao implements JobDao, InitializingBean {
 	
 	private static final String CREATE_JOB_PARAMETERS = "INSERT into %PREFIX%JOB_PARAMS(JOB_INSTANCE_ID, KEY_NAME, TYPE_CD, " +
 			"STRING_VAL, DATE_VAL, LONG_VAL) values (?, ?, ?, ?, ?, ?)";
-	
-	/**	
-	 * Default value for the table prefix property.
-	 */
-	public static final String DEFAULT_TABLE_PREFIX = "BATCH_";
 
 	private static final int EXIT_MESSAGE_LENGTH = 250;
 
@@ -87,14 +80,10 @@ public class JdbcJobDao implements JobDao, InitializingBean {
 	private static final String UPDATE_JOB_EXECUTION = "UPDATE %PREFIX%JOB_EXECUTION set START_TIME = ?, END_TIME = ?, "
 			+ " STATUS = ?, CONTINUABLE = ?, EXIT_CODE = ?, EXIT_MESSAGE = ? where JOB_EXECUTION_ID = ?";
 
-	private JdbcOperations jdbcTemplate;
-
 	private DataFieldMaxValueIncrementer jobExecutionIncrementer;
 
 	private DataFieldMaxValueIncrementer jobIncrementer;
 
-	private String tablePrefix = DEFAULT_TABLE_PREFIX;
-
 	/*
 	 * (non-Javadoc)
 	 * 
@@ -103,8 +92,6 @@ public class JdbcJobDao implements JobDao, InitializingBean {
 	 * Ensure jdbcTemplate and incrementers have been provided.
 	 */
 	public void afterPropertiesSet() throws Exception {
-
-		Assert.notNull(jdbcTemplate, "JdbcTemplate cannot be null");
 		Assert.notNull(jobIncrementer, "JobIncrementor cannot be null");
 		Assert.notNull(jobExecutionIncrementer,
 				"JobExecutionIncrementer cannot be null");
@@ -127,7 +114,7 @@ public class JdbcJobDao implements JobDao, InitializingBean {
 
 		Long jobId = new Long(jobIncrementer.nextLongValue());
 		Object[] parameters = new Object[] { jobId, jobName, createJobKey(jobParameters) };
-		jdbcTemplate.update(getCreateJobQuery(), parameters, new int[] {
+		getJdbcTemplate().update(getCreateJobQuery(), parameters, new int[] {
 			 Types.INTEGER, Types.VARCHAR, Types.VARCHAR});
 
 		insertJobParameters(jobId, jobParameters);
@@ -153,7 +140,7 @@ public class JdbcJobDao implements JobDao, InitializingBean {
 		Assert.notNull(job, "Job cannot be null.");
 		Assert.notNull(job.getId(), "Job Id cannot be null.");
 
-		return jdbcTemplate.query(
+		return getJdbcTemplate().query(
 				getQuery(JobExecutionRowMapper.FIND_JOB_EXECUTIONS),
 				new Object[] { job.getId() }, new JobExecutionRowMapper(job));
 	}
@@ -162,7 +149,7 @@ public class JdbcJobDao implements JobDao, InitializingBean {
 		
 		Assert.notNull(jobExecutionId, "Job Execution id must not be null.");
 		
-		List executions = jdbcTemplate.query(
+		List executions = getJdbcTemplate().query(
 				getQuery(JobExecutionRowMapper.GET_JOB_EXECUTION),
 				new Object[] { jobExecutionId }, new JobExecutionRowMapper(null));
 		
@@ -211,7 +198,7 @@ public class JdbcJobDao implements JobDao, InitializingBean {
 			}
 		};
 
-		return jdbcTemplate.query(getFindJobsQuery(), parameters, rowMapper);
+		return getJdbcTemplate().query(getFindJobsQuery(), parameters, rowMapper);
 	}
 
 	private String getCheckJobExecutionExistsQuery() {
@@ -241,7 +228,7 @@ public class JdbcJobDao implements JobDao, InitializingBean {
 
 		Object[] parameters = new Object[] { jobId };
 
-		return jdbcTemplate
+		return getJdbcTemplate()
 				.queryForInt(getJobExecutionCountQuery(), parameters);
 	}
 
@@ -249,10 +236,6 @@ public class JdbcJobDao implements JobDao, InitializingBean {
 		return getQuery(GET_JOB_EXECUTION_COUNT);
 	}
 
-	private String getQuery(String base) {
-		return StringUtils.replace(base, "%PREFIX%", tablePrefix);
-	}
-
 	private String getSaveJobExecutionQuery() {
 		return getQuery(SAVE_JOB_EXECUTION);
 	}
@@ -317,7 +300,7 @@ public class JdbcJobDao implements JobDao, InitializingBean {
 			args = new Object[]{jobId, key, type, "", value, new Long(0)};
 		}
 		
-		jdbcTemplate.update(getCreateJobParamsQuery(), args, argTypes);
+		getJdbcTemplate().update(getCreateJobParamsQuery(), args, argTypes);
 	}
 
 	/**
@@ -342,15 +325,11 @@ public class JdbcJobDao implements JobDao, InitializingBean {
 				jobExecution.getExitStatus().isContinuable() ? "Y" : "N",
 				jobExecution.getExitStatus().getExitCode(),
 				jobExecution.getExitStatus().getExitDescription() };
-		jdbcTemplate.update(getSaveJobExecutionQuery(), parameters, new int[] {
+		getJdbcTemplate().update(getSaveJobExecutionQuery(), parameters, new int[] {
 				Types.INTEGER, Types.INTEGER, Types.TIMESTAMP, Types.TIMESTAMP,
 				Types.VARCHAR, Types.CHAR, Types.VARCHAR, Types.VARCHAR });
 	}
 
-	public void setJdbcTemplate(JdbcOperations jdbcTemplate) {
-		this.jdbcTemplate = jdbcTemplate;
-	}
-
 	/**
 	 * Setter for {@link DataFieldMaxValueIncrementer} to be used when
 	 * generating primary keys for {@link JobExecution} instances.
@@ -374,18 +353,6 @@ public class JdbcJobDao implements JobDao, InitializingBean {
 		this.jobIncrementer = jobIncrementer;
 	}
 
-	/**
-	 * Public setter for the table prefix property. This will be prefixed to all
-	 * the table names before queries are executed. Defaults to
-	 * {@value #DEFAULT_TABLE_PREFIX}.
-	 * 
-	 * @param tablePrefix
-	 *            the tablePrefix to set
-	 */
-	public void setTablePrefix(String tablePrefix) {
-		this.tablePrefix = tablePrefix;
-	}
-
 	/**
 	 * Update given JobExecution using a SQL UPDATE statement. The JobExecution
 	 * is first checked to ensure all fields are not null, and that it has an
@@ -422,14 +389,14 @@ public class JdbcJobDao implements JobDao, InitializingBean {
 		// Check if given JobExecution's Id already exists, if none is found it
 		// is invalid and
 		// an exception should be thrown.
-		if (jdbcTemplate.queryForInt(getCheckJobExecutionExistsQuery(),
+		if (getJdbcTemplate().queryForInt(getCheckJobExecutionExistsQuery(),
 				new Object[] { jobExecution.getId() }) != 1) {
 			throw new NoSuchBatchDomainObjectException(
 					"Invalid JobExecution, ID " + jobExecution.getId()
 							+ " not found.");
 		}
 
-		jdbcTemplate
+		getJdbcTemplate()
 				.update(getUpdateJobExecutionQuery(), parameters,
 						new int[] { Types.TIMESTAMP, Types.TIMESTAMP,
 								Types.VARCHAR, Types.CHAR, Types.VARCHAR,
@@ -448,7 +415,7 @@ public class JdbcJobDao implements JobDao, InitializingBean {
 		
 		Long lastExecutionId = jobInstance.getLastExecution() == null ? null : jobInstance.getLastExecution().getId();
 		Object[] parameters = new Object[] { lastExecutionId, jobInstance.getId() };
-		jdbcTemplate.update(getUpdateJobQuery(), parameters, new int[] {
+		getJdbcTemplate().update(getUpdateJobQuery(), parameters, new int[] {
 			 Types.INTEGER, Types.INTEGER});
 	}
 
diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/JdbcStepDao.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/JdbcStepExecutionDao.java
similarity index 52%
rename from spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/JdbcStepDao.java
rename to spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/JdbcStepExecutionDao.java
index ea66de258..9fdf84999 100644
--- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/JdbcStepDao.java
+++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/JdbcStepExecutionDao.java
@@ -1,730 +1,503 @@
-/*
- * Copyright 2006-2007 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.execution.repository.dao;
-
-import java.io.Serializable;
-import java.sql.PreparedStatement;
-import java.sql.ResultSet;
-import java.sql.SQLException;
-import java.sql.Types;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map.Entry;
-
-import org.apache.commons.lang.SerializationUtils;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-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.execution.repository.dao.JdbcJobDao.JobExecutionRowMapper;
-import org.springframework.batch.io.exception.BatchCriticalException;
-import org.springframework.batch.item.ExecutionAttributes;
-import org.springframework.batch.repeat.ExitStatus;
-import org.springframework.batch.support.PropertiesConverter;
-import org.springframework.beans.factory.InitializingBean;
-import org.springframework.dao.DataAccessException;
-import org.springframework.dao.IncorrectResultSizeDataAccessException;
-import org.springframework.dao.OptimisticLockingFailureException;
-import org.springframework.jdbc.core.JdbcOperations;
-import org.springframework.jdbc.core.PreparedStatementCallback;
-import org.springframework.jdbc.core.RowCallbackHandler;
-import org.springframework.jdbc.core.RowMapper;
-import org.springframework.jdbc.core.support.AbstractLobCreatingPreparedStatementCallback;
-import org.springframework.jdbc.support.incrementer.DataFieldMaxValueIncrementer;
-import org.springframework.jdbc.support.lob.DefaultLobHandler;
-import org.springframework.jdbc.support.lob.LobCreator;
-import org.springframework.jdbc.support.lob.LobHandler;
-import org.springframework.util.Assert;
-import org.springframework.util.StringUtils;
-
-/**
- * Jdbc implementation of {@link StepDao}.<br/>
- * 
- * Allows customization of the tables names used by Spring Batch for step meta
- * data via a prefix property.<br/>
- * 
- * Uses sequences or tables (via Spring's {@link DataFieldMaxValueIncrementer}
- * abstraction) to create all primary keys before inserting a new row. All
- * objects are checked to ensure all fields to be stored are not null. If any
- * are found to be null, an IllegalArgumentException will be thrown. This could
- * be left to JdbcTemplate, however, the exception will be fairly vague, and
- * fails to highlight which field caused the exception.<br/>
- * 
- * @author Lucas Ward
- * @author Dave Syer
- * 
- * @see StepDao
- */
-public class JdbcStepDao implements StepDao, InitializingBean {
-
-	private static final String CREATE_STEP = "INSERT into %PREFIX%STEP_INSTANCE(STEP_INSTANCE_ID, JOB_INSTANCE_ID, STEP_NAME) values (?, ?, ?)";
-
-	private static final int EXIT_MESSAGE_LENGTH = 250;
-
-	private static final String FIND_STEP = "SELECT STEP_INSTANCE_ID, LAST_STEP_EXECUTION_ID from %PREFIX%STEP_INSTANCE where JOB_INSTANCE_ID = ? "
-			+ "and STEP_NAME = ?";
-
-	private static final String FIND_STEP_EXECUTIONS = "SELECT STEP_EXECUTION_ID, JOB_EXECUTION_ID, START_TIME, END_TIME, STATUS, COMMIT_COUNT,"
-			+ " TASK_COUNT, TASK_STATISTICS, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE from %PREFIX%STEP_EXECUTION where STEP_INSTANCE_ID = ?";
-
-	private static final String GET_STEP_EXECUTION = "SELECT STEP_EXECUTION_ID, JOB_EXECUTION_ID, START_TIME, END_TIME, STATUS, COMMIT_COUNT,"
-		+ " TASK_COUNT, TASK_STATISTICS, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE from %PREFIX%STEP_EXECUTION where STEP_EXECUTION_ID = ?";
-	
-	// Step SQL statements
-	private static final String FIND_STEPS = "SELECT STEP_INSTANCE_ID, LAST_STEP_EXECUTION_ID, STEP_NAME from %PREFIX%STEP_INSTANCE where JOB_INSTANCE_ID = ?";
-
-	private static final String GET_STEP_EXECUTION_COUNT = "SELECT count(STEP_EXECUTION_ID) from %PREFIX%STEP_EXECUTION where "
-			+ "STEP_INSTANCE_ID = ?";
-
-	protected static final Log logger = LogFactory.getLog(JdbcStepDao.class);
-
-	// StepExecution statements
-	private static final String SAVE_STEP_EXECUTION = "INSERT into %PREFIX%STEP_EXECUTION(STEP_EXECUTION_ID, VERSION, STEP_INSTANCE_ID, JOB_EXECUTION_ID, START_TIME, "
-			+ "END_TIME, STATUS, COMMIT_COUNT, TASK_COUNT, TASK_STATISTICS, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE) "
-			+ "values(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
-
-	private static final String UPDATE_STEP = "UPDATE %PREFIX%STEP_INSTANCE set LAST_STEP_EXECUTION_ID = ? where STEP_INSTANCE_ID = ?";
-
-	private static final String UPDATE_STEP_EXECUTION = "UPDATE %PREFIX%STEP_EXECUTION set START_TIME = ?, END_TIME = ?, "
-			+ "STATUS = ?, COMMIT_COUNT = ?, TASK_COUNT = ?, TASK_STATISTICS = ?, CONTINUABLE = ? , EXIT_CODE = ?, "
-			+ "EXIT_MESSAGE = ?, VERSION = ? where STEP_EXECUTION_ID = ? and VERSION = ?";
-
-	private static final String UPDATE_STEP_EXECUTION_ATTRS = "UPDATE %PREFIX%STEP_EXECUTION_ATTRS set " +
-			"TYPE_CD = ?, STRING_VAL = ?, DOUBLE_VAL = ?, LONG_VAL = ?, OBJECT_VAL = ? where STEP_EXECUTION_ID = ? and KEY_NAME = ?";
-	
-	private static final String INSERT_STEP_EXECUTION_ATTRS = "INSERT into %PREFIX%STEP_EXECUTION_ATTRS(STEP_EXECUTION_ID, TYPE_CD," +
-			" KEY_NAME, STRING_VAL, DOUBLE_VAL, LONG_VAL, OBJECT_VAL) values(?,?,?,?,?,?,?)";
-	
-	private static final String FIND_STEP_EXECUTION_ATTRS = "SELECT TYPE_CD, KEY_NAME, STRING_VAL, DOUBLE_VAL, LONG_VAL, OBJECT_VAL " +
-			"from %PREFIX%STEP_EXECUTION_ATTRS where STEP_EXECUTION_ID = ?";
-	
-	private JdbcOperations jdbcTemplate;
-
-	private JobDao jobDao;
-
-	private DataFieldMaxValueIncrementer stepExecutionIncrementer;
-
-	private DataFieldMaxValueIncrementer stepIncrementer;
-	
-	private LobHandler lobHandler = new DefaultLobHandler();
-
-	private String tablePrefix = JdbcJobDao.DEFAULT_TABLE_PREFIX;
-
-	public void afterPropertiesSet() throws Exception {
-		Assert.notNull(jdbcTemplate, "JdbcTemplate cannot be null.");
-		Assert.notNull(stepIncrementer, "StepIncrementer cannot be null.");
-		Assert.notNull(stepExecutionIncrementer, "StepExecutionIncrementer canot be null.");
-	}
-
-	private void cascadeJobExecution(JobExecution jobExecution) {
-		if (jobExecution.getId() != null) {
-			// assume already saved...
-			return;
-		}
-		jobDao.saveJobExecution(jobExecution);
-	}
-
-	/**
-	 * Create a step with the given job's id, and the provided step name. A
-	 * unique id is created for the step using an incrementer. (@link
-	 * DataFieldMaxValueIncrementer)
-	 * 
-	 * @see StepDao#createStepInstance(JobInstance, String)
-	 * @throws IllegalArgumentException if job or stepName is null.
-	 */
-	public StepInstance createStepInstance(JobInstance job, String stepName) {
-
-		Assert.notNull(job, "Job cannot be null.");
-		Assert.notNull(stepName, "StepName cannot be null.");
-
-		Long stepId = new Long(stepIncrementer.nextLongValue());
-		Object[] parameters = new Object[] { stepId, job.getId(), stepName };
-		jdbcTemplate.update(getCreateStepQuery(), parameters);
-
-		StepInstance step = new StepInstance(job, stepName, stepId);
-		return step;
-	}
-
-	/**
-	 * 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
-	 * and null will be returned. If one step is found, it will be returned. If
-	 * anymore than one step is found, an exception is thrown.
-	 * 
-	 * @see StepDao#findStepInstance(Long, String)
-	 * @throws IllegalArgumentException if job, stepName, or job.id is null.
-	 * @throws IncorrectResultSizeDataAccessException if more than one step is
-	 * found.
-	 */
-	public StepInstance findStepInstance(JobInstance jobInstance, String stepName) {
-
-		Assert.notNull(jobInstance, "Job cannot be null.");
-		Assert.notNull(jobInstance.getId(), "Job ID cannot be null");
-		Assert.notNull(stepName, "StepName cannot be null");
-
-		Object[] parameters = new Object[] { jobInstance.getId(), stepName };
-
-		RowMapper rowMapper = new StepInstanceRowMapper(jobInstance, stepName);
-
-		List steps = jdbcTemplate.query(getFindStepQuery(), parameters, rowMapper);
-
-		if (steps.size() == 0) {
-			// No step found
-			return null;
-		}
-		else if (steps.size() == 1) {
-			StepInstance step = (StepInstance) steps.get(0);
-			return step;
-		}
-		else {
-			// This error will likely never be thrown, because there should
-			// never be two steps with the same name and JOB_INSTANCE_ID due to database
-			// constraints.
-			throw new IncorrectResultSizeDataAccessException("Step Invalid, multiple steps found for StepName:"
-					+ stepName + " and JobId:" + jobInstance.getId(), 1, steps.size());
-		}
-
-	}
-
-	/**
-	 * Get StepExecution for the given step. Due to the nature of statistics,
-	 * they will not be returned with reconstituted object.
-	 * 
-	 * @see StepDao#getStepExecution(Long)
-	 * @throws IllegalArgumentException if id is null.
-	 */
-	public List findStepExecutions(final StepInstance step) {
-
-		Assert.notNull(step, "Step cannot be null.");
-		Assert.notNull(step.getId(), "Step id cannot be null.");
-
-		RowMapper rowMapper = new StepExecutionRowMapper(step);
-
-		return jdbcTemplate.query(getFindStepExecutionsQuery(), new Object[] { step.getId() }, rowMapper);
-	}
-	
-	public StepExecution getStepExecution(Long stepExecutionId, StepInstance stepInstance) {
-		
-		Assert.notNull(stepExecutionId, "Step Execution id must not be null");
-
-		RowMapper rowMapper = new StepExecutionRowMapper(stepInstance);
-
-		List executions = jdbcTemplate.query(getQuery(GET_STEP_EXECUTION), new Object[] { stepExecutionId }, rowMapper);
-		
-		StepExecution stepExecution;
-		if(executions.size() == 1){
-			stepExecution = (StepExecution)executions.get(0);
-		}
-		else if(executions.size() == 0){
-			stepExecution = null;
-		}
-		else{
-			throw new IncorrectResultSizeDataAccessException("Only one StepExecution may exist for given id: [" + 
-					stepExecutionId + "]", 1, executions.size());
-		}
-		
-		return stepExecution;
-	}
-	
-	/*
-	 * Insert execution attributes.  A lob creator must be used, since any attributes
-	 * that don't match a provided type must be serialized into a blob.
-	 */
-	public void saveExecutionAttributes(final Long executionId, final ExecutionAttributes executionAttributes){
-				
-		Assert.notNull(executionId, "ExecutionId must not be null.");
-		Assert.notNull(executionAttributes, "The ExecutionAttributes must not be null.");
-		
-		for(Iterator it = executionAttributes.entrySet().iterator();it.hasNext();){
-			Entry entry = (Entry)it.next();
-			final String key = entry.getKey().toString();
-			final Object value = entry.getValue();
-
-			if(value instanceof String){
-				insertExecutionAttribute(executionId, key, value, AttributeType.STRING);
-			}
-			else if(value instanceof Double){
-				insertExecutionAttribute(executionId, key, value, AttributeType.DOUBLE);
-			}
-			else if(value instanceof Long){
-				insertExecutionAttribute(executionId, key, value, AttributeType.LONG);
-			}
-			else
-			{
-				insertExecutionAttribute(executionId, key, value, AttributeType.OBJECT);
-			}
-		}
-	}
-	
-	private void insertExecutionAttribute(final Long executionId, final String key, final Object value, final AttributeType type){
-		PreparedStatementCallback callback = new AbstractLobCreatingPreparedStatementCallback(lobHandler){
-
-			protected void setValues(PreparedStatement ps, LobCreator lobCreator)
-					throws SQLException, DataAccessException {
-				
-				ps.setLong(1, executionId.longValue());
-				ps.setString(3, key);
-				if(type == AttributeType.STRING){
-					ps.setString(2,AttributeType.STRING.toString());
-					ps.setString(4,value.toString());
-					ps.setDouble(5, 0.0);
-					ps.setLong(6, 0);
-					lobCreator.setBlobAsBytes(ps, 7, null);
-				}
-				else if(type == AttributeType.DOUBLE){
-					ps.setString(2,AttributeType.DOUBLE.toString());
-					ps.setString(4,null);
-					ps.setDouble(5, ((Double)value).doubleValue());
-					ps.setLong(6, 0);
-					lobCreator.setBlobAsBytes(ps, 7, null);
-				}
-				else if(type == AttributeType.LONG){
-					ps.setString(2,AttributeType.LONG.toString());
-					ps.setString(4,null);
-					ps.setDouble(5, 0.0 );
-					ps.setLong(6, ((Long)value).longValue());
-					lobCreator.setBlobAsBytes(ps, 7, null);
-				}
-				else{
-					ps.setString(2,AttributeType.OBJECT.toString());
-					ps.setString(4,null);
-					ps.setDouble(5, 0.0 );
-					ps.setLong(6, 0);
-					lobCreator.setBlobAsBytes(ps, 7, SerializationUtils.serialize((Serializable)value));
-				}
-			}};
-			
-			jdbcTemplate.execute(getQuery(INSERT_STEP_EXECUTION_ATTRS), callback);
-	}
-	
-	/**
-	 * update execution attributes.  A lob creator must be used, since any attributes
-	 * that don't match a provided type must be serialized into a blob.
-	 * 
-	 * @see {@link LobCreator}
-	 */
-	public void updateExecutionAttributes(final Long executionId, ExecutionAttributes executionAttributes){
-		
-		Assert.notNull(executionId, "ExecutionId must not be null.");
-		Assert.notNull(executionAttributes, "The ExecutionAttributes must not be null.");
-		
-		for(Iterator it = executionAttributes.entrySet().iterator();it.hasNext();){
-			Entry entry = (Entry)it.next();
-			final String key = entry.getKey().toString();
-			final Object value = entry.getValue();
-			
-			if(value instanceof String){
-				updateExecutionAttribute(executionId, key, value, AttributeType.STRING);
-			}
-			else if(value instanceof Double){
-				updateExecutionAttribute(executionId, key, value, AttributeType.DOUBLE);
-			}
-			else if(value instanceof Long){
-				updateExecutionAttribute(executionId, key, value, AttributeType.LONG);
-			}
-			else
-			{
-				updateExecutionAttribute(executionId, key, value, AttributeType.OBJECT);
-			}
-		}
-	}
-	
-	private void updateExecutionAttribute(final Long executionId, final String key, final Object value, final AttributeType type){
-		
-		PreparedStatementCallback callback = new AbstractLobCreatingPreparedStatementCallback(lobHandler){
-
-			protected void setValues(PreparedStatement ps, LobCreator lobCreator)
-					throws SQLException, DataAccessException {
-				
-				ps.setLong(6, executionId.longValue());
-				ps.setString(7, key);
-				if(type == AttributeType.STRING){
-					ps.setString(1,AttributeType.STRING.toString());
-					ps.setString(2,value.toString());
-					ps.setDouble(3, 0.0);
-					ps.setLong(4, 0);
-					lobCreator.setBlobAsBytes(ps, 5, null);
-				}
-				else if(type == AttributeType.DOUBLE){
-					ps.setString(1,AttributeType.DOUBLE.toString());
-					ps.setString(2,null);
-					ps.setDouble(3, ((Double)value).doubleValue());
-					ps.setLong(4, 0);
-					lobCreator.setBlobAsBytes(ps, 5, null);
-				}
-				else if(type == AttributeType.LONG){
-					ps.setString(1,AttributeType.LONG.toString());
-					ps.setString(2,null);
-					ps.setDouble(3, 0.0 );
-					ps.setLong(4, ((Long)value).longValue());
-					lobCreator.setBlobAsBytes(ps, 5, null);
-				}
-				else{
-					ps.setString(1,AttributeType.OBJECT.toString());
-					ps.setString(2,null);
-					ps.setDouble(3, 0.0 );
-					ps.setLong(4, 0);
-					lobCreator.setBlobAsBytes(ps, 5, SerializationUtils.serialize((Serializable)value));
-				}
-			}};
-			
-			//LobCreating callbacks always return the affect row count for SQL DML statements, if less than 1 row
-			//is affected, then this row is new and should be inserted.
-			Integer affectedRows = (Integer)jdbcTemplate.execute(getQuery(UPDATE_STEP_EXECUTION_ATTRS), callback);
-			if(affectedRows.intValue() < 1){
-				insertExecutionAttribute(executionId, key, value, type);
-			}
-	}
-
-	/**
-	 * @see StepDao#findStepInstances(JobInstance)
-	 * 
-	 * Sql implementation which uses a RowMapper to populate a list of all rows
-	 * in the step table with the same JOB_INSTANCE_ID.
-	 * 
-	 * @throws IllegalArgumentException if jobId is null.
-	 */
-	public List findStepInstances(final JobInstance jobInstance) {
-
-		Assert.notNull(jobInstance, "Job cannot be null.");
-
-		Object[] parameters = new Object[] { jobInstance.getId() };
-
-		RowMapper rowMapper = new StepInstanceRowMapper(jobInstance, null);
-
-		return jdbcTemplate.query(getFindStepsQuery(), parameters, rowMapper);
-	}
-
-	private String getCreateStepQuery() {
-		return getQuery(CREATE_STEP);
-	}
-
-	private String getFindStepExecutionsQuery() {
-		return getQuery(FIND_STEP_EXECUTIONS);
-	}
-
-	private String getFindStepQuery() {
-		return getQuery(FIND_STEP);
-	}
-
-	private String getFindStepsQuery() {
-		return getQuery(FIND_STEPS);
-	}
-
-	private String getQuery(String base) {
-		return StringUtils.replace(base, "%PREFIX%", tablePrefix);
-	}
-
-	private String getSaveStepExecutionQuery() {
-		return getQuery(SAVE_STEP_EXECUTION);
-	}
-
-	public int getStepExecutionCount(StepInstance step) {
-
-		Object[] parameters = new Object[] { step.getId() };
-
-		return jdbcTemplate.queryForInt(getStepExecutionCountQuery(), parameters);
-	}
-
-	private String getStepExecutionCountQuery() {
-		return getQuery(GET_STEP_EXECUTION_COUNT);
-	}
-
-	private String getUpdateStepExecutionQuery() {
-		return getQuery(UPDATE_STEP_EXECUTION);
-	}
-
-	private String getUpdateStepQuery() {
-		return getQuery(UPDATE_STEP);
-	}
-
-	/**
-	 * Save a StepExecution. A unique id will be generated by the
-	 * stepExecutionIncrementor, and then set in the StepExecution. All values
-	 * will then be stored via an INSERT statement.
-	 * 
-	 * @see StepDao#saveStepExecution(StepExecution)
-	 */
-	public void saveStepExecution(StepExecution stepExecution) {
-
-		validateStepExecution(stepExecution);
-
-		cascadeJobExecution(stepExecution.getJobExecution());
-
-		stepExecution.setId(new Long(stepExecutionIncrementer.nextLongValue()));
-		stepExecution.incrementVersion(); // should be 0 now
-		Object[] parameters = new Object[] { stepExecution.getId(), stepExecution.getVersion(),
-				stepExecution.getStepId(), stepExecution.getJobExecutionId(), stepExecution.getStartTime(),
-				stepExecution.getEndTime(), stepExecution.getStatus().toString(), stepExecution.getCommitCount(),
-				stepExecution.getTaskCount(),
-				PropertiesConverter.propertiesToString(stepExecution.getExecutionAttributes().getProperties()),
-				stepExecution.getExitStatus().isContinuable() ? "Y" : "N", stepExecution.getExitStatus().getExitCode(),
-				stepExecution.getExitStatus().getExitDescription() };
-		jdbcTemplate.update(getSaveStepExecutionQuery(), parameters, new int[] { Types.INTEGER, Types.INTEGER,
-				Types.INTEGER, Types.INTEGER, Types.TIMESTAMP, Types.TIMESTAMP, Types.VARCHAR, Types.INTEGER,
-				Types.INTEGER, Types.VARCHAR, Types.CHAR, Types.VARCHAR, Types.VARCHAR });
-	}
-
-	public void setJdbcTemplate(JdbcOperations jdbcTemplate) {
-		this.jdbcTemplate = jdbcTemplate;
-	}
-
-	/**
-	 * Injection setter for job dao. Used to save {@link JobExecution}
-	 * instances.
-	 * 
-	 * @param jobDao a {@link JobDao}
-	 */
-	public void setJobDao(JobDao jobDao) {
-		this.jobDao = jobDao;
-	}
-
-	/**
-	 * Set the {@link DataFieldMaxValueIncrementer} that will be used to
-	 * increment the primary keys used for {@link StepExecution} instances.
-	 * 
-	 * @param stepExecutionIncrementer a {@link DataFieldMaxValueIncrementer}
-	 */
-	public void setStepExecutionIncrementer(DataFieldMaxValueIncrementer stepExecutionIncrementer) {
-		this.stepExecutionIncrementer = stepExecutionIncrementer;
-	}
-
-	/**
-	 * Set the {@link DataFieldMaxValueIncrementer} that will be used to
-	 * increment the primary keys used for {@link StepInstance} instances.
-	 * 
-	 * @param stepExecutionIncrementer a {@link DataFieldMaxValueIncrementer}
-	 */
-	public void setStepIncrementer(DataFieldMaxValueIncrementer stepIncrementer) {
-		this.stepIncrementer = stepIncrementer;
-	}
-
-	/**
-	 * Public setter for the table prefix property. This will be prefixed to all
-	 * the table names before queries are executed (unless individual queries
-	 * are overridden with the set*Query methods). Defaults to
-	 * {@value #DEFAULT_TABLE_PREFIX}.
-	 * 
-	 * @param tablePrefix the tablePrefix to set
-	 */
-	public void setTablePrefix(String tablePrefix) {
-		this.tablePrefix = tablePrefix;
-	}
-
-	/**
-	 * @see StepDao#updateStepExecution(StepExecution)
-	 */
-	public void updateStepExecution(StepExecution stepExecution) {
-
-		validateStepExecution(stepExecution);
-		Assert.notNull(stepExecution.getId(), "StepExecution Id cannot be null. StepExecution must saved"
-				+ " before it can be updated.");
-
-		// Do not check for existence of step execution considering
-		// it is saved at every commit point.
-
-		String exitDescription = stepExecution.getExitStatus().getExitDescription();
-		if (exitDescription != null && exitDescription.length() > EXIT_MESSAGE_LENGTH) {
-			exitDescription = exitDescription.substring(0, EXIT_MESSAGE_LENGTH);
-			logger.debug("Truncating long message before update of StepExecution: " + stepExecution);
-		}
-
-		// Attempt to prevent concurrent modification errors by blocking here if
-		// someone is already trying to do it.
-		synchronized (stepExecution) {
-
-			Integer version = new Integer(stepExecution.getVersion().intValue() + 1);
-			Object[] parameters = new Object[] { stepExecution.getStartTime(), stepExecution.getEndTime(),
-					stepExecution.getStatus().toString(), stepExecution.getCommitCount(), stepExecution.getTaskCount(),
-					PropertiesConverter.propertiesToString(stepExecution.getExecutionAttributes().getProperties()),
-					stepExecution.getExitStatus().isContinuable() ? "Y" : "N",
-					stepExecution.getExitStatus().getExitCode(), exitDescription, version, stepExecution.getId(),
-					stepExecution.getVersion() };
-			int count = jdbcTemplate.update(getUpdateStepExecutionQuery(), parameters, new int[] { Types.TIMESTAMP,
-					Types.TIMESTAMP, Types.VARCHAR, Types.INTEGER, Types.INTEGER, Types.VARCHAR, Types.CHAR,
-					Types.VARCHAR, Types.VARCHAR, Types.INTEGER, Types.INTEGER, Types.INTEGER });
-
-			// Avoid concurrent modifications...
-			if (count == 0) {
-				throw new OptimisticLockingFailureException("Attempt to update step execution id="
-						+ stepExecution.getId() + " with wrong version (" + stepExecution.getVersion() + ")");
-			}
-			
-			stepExecution.incrementVersion();
-			
-		}
-	}
-	
-	public ExecutionAttributes findExecutionAttributes(final Long executionId){
-		
-		Assert.notNull(executionId, "ExecutionId must not be null.");
-		
-		final ExecutionAttributes executionAttributes = new ExecutionAttributes();
-		
-		RowCallbackHandler callback = new RowCallbackHandler(){
-
-			public void processRow(ResultSet rs) throws SQLException {
-				
-				String typeCd = rs.getString("TYPE_CD");
-				AttributeType type = AttributeType.getType(typeCd);
-				String key = rs.getString("KEY_NAME");
-				if(type == AttributeType.STRING){
-					executionAttributes.putString(key, rs.getString("STRING_VAL"));
-				}
-				else if(type == AttributeType.LONG){
-					executionAttributes.putLong(key, rs.getLong("LONG_VAL"));
-				}
-				else if(type == AttributeType.DOUBLE){
-					executionAttributes.putDouble(key, rs.getDouble("DOUBLE_VAL"));
-				}
-				else if(type == AttributeType.OBJECT){
-					executionAttributes.putLong(key, rs.getLong("OBJECT_VAL"));
-				}
-				else{
-					throw new BatchCriticalException("Invalid type found: [" + typeCd + "] for execution id: [" +
-							executionId + "]");
-				}
-			}
-		};
-		
-		jdbcTemplate.query(getQuery(FIND_STEP_EXECUTION_ATTRS), new Object[]{executionId}, callback);
-		
-		return executionAttributes;
-	}
-
-	/**
-	 * @see StepDao#updateStepInstance(StepInstance)
-	 * @throws IllegalArgumentException if step, or it's status and id is null.
-	 */
-	public void updateStepInstance(final StepInstance step) {
-
-		Assert.notNull(step, "Step cannot be null.");
-		Assert.notNull(step.getId(), "Step Id cannot be null.");
-
-		Object[] parameters = new Object[] { step.getLastExecution().getId(), step.getId() };
-
-		jdbcTemplate.update(getUpdateStepQuery(), parameters);
-	}
-
-	/*
-	 * Validate StepExecution. At a minimum, JobId, StartTime, and Status cannot
-	 * be null. EndTime can be null for an unfinished job.
-	 * 
-	 * @param jobExecution @throws IllegalArgumentException
-	 */
-	private void validateStepExecution(StepExecution stepExecution) {
-		Assert.notNull(stepExecution);
-		Assert.notNull(stepExecution.getStepId(), "StepExecution Step-Id cannot be null.");
-		Assert.notNull(stepExecution.getStartTime(), "StepExecution start time cannot be null.");
-		Assert.notNull(stepExecution.getStatus(), "StepExecution status cannot be null.");
-	}
-	
-	public void setLobHandler(LobHandler lobHandler) {
-		this.lobHandler = lobHandler;
-	}
-	
-	public static class AttributeType {
-
-		private final String type;
-
-		private AttributeType(String type) {
-			this.type = type;
-		}
-
-		public String toString() {
-			return type;
-		}
-
-		public static final AttributeType STRING = new AttributeType("STRING");
-
-		public static final AttributeType LONG = new AttributeType("LONG");
-
-		public static final AttributeType OBJECT = new AttributeType("OBJECT");
-
-		public static final AttributeType DOUBLE = new AttributeType("DOUBLE");
-
-		private static final AttributeType[] VALUES = { STRING, OBJECT, LONG,
-				DOUBLE };
-
-		public static AttributeType getType(String typeAsString) {
-
-			for (int i = 0; i < VALUES.length; i++) {
-				if (VALUES[i].toString().equals(typeAsString)) {
-					return (AttributeType) VALUES[i];
-				}
-			}
-
-			return null;
-		}
-	}
-	
-	private class StepExecutionRowMapper implements RowMapper{
-
-		private final StepInstance stepInstance;
-		
-		public StepExecutionRowMapper(StepInstance stepInstance) {
-			this.stepInstance = stepInstance;
-		}
-		
-		public Object mapRow(ResultSet rs, int rowNum) throws SQLException {
-
-			JobExecution jobExecution = (JobExecution) jdbcTemplate.queryForObject(
-					getQuery(JobExecutionRowMapper.GET_JOB_EXECUTION), new Object[] { new Long(rs.getLong(2)) },
-					new JobExecutionRowMapper(stepInstance.getJobInstance()));
-			StepExecution stepExecution = new StepExecution(stepInstance, jobExecution, new Long(rs.getLong(1)));
-			stepExecution.setStartTime(rs.getTimestamp(3));
-			stepExecution.setEndTime(rs.getTimestamp(4));
-			stepExecution.setStatus(BatchStatus.getStatus(rs.getString(5)));
-			stepExecution.setCommitCount(rs.getInt(6));
-			stepExecution.setTaskCount(rs.getInt(7));
-			stepExecution.setExecutionAttributes(new ExecutionAttributes(PropertiesConverter
-					.stringToProperties(rs.getString(8))));
-			stepExecution.setExitStatus(new ExitStatus("Y".equals(rs.getString(9)), rs.getString(10), rs
-					.getString(11)));
-			return stepExecution;
-		}
-		
-	}
-	
-	private class StepInstanceRowMapper implements RowMapper{
-
-		private final JobInstance jobInstance;
-		private String stepName;
-		
-		public StepInstanceRowMapper(JobInstance jobInstance, String stepName) {
-			this.jobInstance = jobInstance;
-			this.stepName = stepName;
-		}
-		
-		public Object mapRow(ResultSet rs, int rowNum) throws SQLException {
-
-			if(stepName == null){
-				stepName = rs.getString(3);
-			}
-			StepInstance stepInstance = new StepInstance(jobInstance, stepName, new Long(rs.getLong(1)));
-			StepExecution lastExecution = getStepExecution(new Long(rs.getLong(2)), stepInstance);
-			stepInstance.setLastExecution(lastExecution);
-			return stepInstance;
-		}
-
-		
-	}
-
-}
+package org.springframework.batch.execution.repository.dao;
+
+import java.io.Serializable;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Types;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map.Entry;
+
+import org.apache.commons.lang.SerializationUtils;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.batch.core.domain.BatchStatus;
+import org.springframework.batch.core.domain.JobExecution;
+import org.springframework.batch.core.domain.StepExecution;
+import org.springframework.batch.core.domain.StepInstance;
+import org.springframework.batch.execution.repository.dao.JdbcJobDao.JobExecutionRowMapper;
+import org.springframework.batch.io.exception.BatchCriticalException;
+import org.springframework.batch.item.ExecutionAttributes;
+import org.springframework.batch.repeat.ExitStatus;
+import org.springframework.batch.support.PropertiesConverter;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.dao.DataAccessException;
+import org.springframework.dao.IncorrectResultSizeDataAccessException;
+import org.springframework.dao.OptimisticLockingFailureException;
+import org.springframework.jdbc.core.PreparedStatementCallback;
+import org.springframework.jdbc.core.RowCallbackHandler;
+import org.springframework.jdbc.core.RowMapper;
+import org.springframework.jdbc.core.support.AbstractLobCreatingPreparedStatementCallback;
+import org.springframework.jdbc.support.incrementer.DataFieldMaxValueIncrementer;
+import org.springframework.jdbc.support.lob.DefaultLobHandler;
+import org.springframework.jdbc.support.lob.LobCreator;
+import org.springframework.jdbc.support.lob.LobHandler;
+import org.springframework.util.Assert;
+
+/**
+ * Jdbc implementation of {@link StepExecutionDao}.<br/>
+ * 
+ * Allows customization of the tables names used by Spring Batch for step meta
+ * data via a prefix property.<br/>
+ * 
+ * Uses sequences or tables (via Spring's {@link DataFieldMaxValueIncrementer}
+ * abstraction) to create all primary keys before inserting a new row. All
+ * objects are checked to ensure all fields to be stored are not null. If any
+ * are found to be null, an IllegalArgumentException will be thrown. This could
+ * be left to JdbcTemplate, however, the exception will be fairly vague, and
+ * fails to highlight which field caused the exception.<br/>
+ * 
+ * @author Lucas Ward
+ * @author Dave Syer
+ * @author Robert Kasanicky
+ * 
+ * @see StepExecutionDao
+ */
+public class JdbcStepExecutionDao extends AbstractJdbcBatchMetadataDao 
+	implements StepExecutionDao, InitializingBean {
+
+	private static final Log logger = LogFactory.getLog(JdbcStepExecutionDao.class);
+
+	private static final String FIND_STEP_EXECUTION_ATTRS = "SELECT TYPE_CD, KEY_NAME, STRING_VAL, DOUBLE_VAL, LONG_VAL, OBJECT_VAL "
+			+ "from %PREFIX%STEP_EXECUTION_ATTRS where STEP_EXECUTION_ID = ?";
+
+	private static final String FIND_STEP_EXECUTIONS = "SELECT STEP_EXECUTION_ID, JOB_EXECUTION_ID, START_TIME, END_TIME, STATUS, COMMIT_COUNT,"
+			+ " TASK_COUNT, TASK_STATISTICS, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE from %PREFIX%STEP_EXECUTION where STEP_INSTANCE_ID = ?";
+
+	private static final String GET_STEP_EXECUTION = "SELECT STEP_EXECUTION_ID, JOB_EXECUTION_ID, START_TIME, END_TIME, STATUS, COMMIT_COUNT,"
+			+ " TASK_COUNT, TASK_STATISTICS, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE from %PREFIX%STEP_EXECUTION where STEP_EXECUTION_ID = ?";
+
+	private static final String GET_STEP_EXECUTION_COUNT = "SELECT count(STEP_EXECUTION_ID) from %PREFIX%STEP_EXECUTION where "
+			+ "STEP_INSTANCE_ID = ?";
+
+	private static final String INSERT_STEP_EXECUTION_ATTRS = "INSERT into %PREFIX%STEP_EXECUTION_ATTRS(STEP_EXECUTION_ID, TYPE_CD,"
+			+ " KEY_NAME, STRING_VAL, DOUBLE_VAL, LONG_VAL, OBJECT_VAL) values(?,?,?,?,?,?,?)";
+
+	private static final String SAVE_STEP_EXECUTION = "INSERT into %PREFIX%STEP_EXECUTION(STEP_EXECUTION_ID, VERSION, STEP_INSTANCE_ID, JOB_EXECUTION_ID, START_TIME, "
+			+ "END_TIME, STATUS, COMMIT_COUNT, TASK_COUNT, TASK_STATISTICS, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE) "
+			+ "values(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
+
+	private static final String UPDATE_STEP_EXECUTION_ATTRS = "UPDATE %PREFIX%STEP_EXECUTION_ATTRS set "
+			+ "TYPE_CD = ?, STRING_VAL = ?, DOUBLE_VAL = ?, LONG_VAL = ?, OBJECT_VAL = ? where STEP_EXECUTION_ID = ? and KEY_NAME = ?";
+
+	private static final String UPDATE_STEP_EXECUTION = "UPDATE %PREFIX%STEP_EXECUTION set START_TIME = ?, END_TIME = ?, "
+			+ "STATUS = ?, COMMIT_COUNT = ?, TASK_COUNT = ?, TASK_STATISTICS = ?, CONTINUABLE = ? , EXIT_CODE = ?, "
+			+ "EXIT_MESSAGE = ?, VERSION = ? where STEP_EXECUTION_ID = ? and VERSION = ?";
+
+	private static final int EXIT_MESSAGE_LENGTH = 250;
+
+	private LobHandler lobHandler = new DefaultLobHandler();
+
+	private DataFieldMaxValueIncrementer stepExecutionIncrementer;
+
+	private JobDao jobDao;
+
+	public ExecutionAttributes findExecutionAttributes(final Long executionId) {
+
+		Assert.notNull(executionId, "ExecutionId must not be null.");
+
+		final ExecutionAttributes executionAttributes = new ExecutionAttributes();
+
+		RowCallbackHandler callback = new RowCallbackHandler() {
+
+			public void processRow(ResultSet rs) throws SQLException {
+
+				String typeCd = rs.getString("TYPE_CD");
+				AttributeType type = AttributeType.getType(typeCd);
+				String key = rs.getString("KEY_NAME");
+				if (type == AttributeType.STRING) {
+					executionAttributes.putString(key, rs.getString("STRING_VAL"));
+				}
+				else if (type == AttributeType.LONG) {
+					executionAttributes.putLong(key, rs.getLong("LONG_VAL"));
+				}
+				else if (type == AttributeType.DOUBLE) {
+					executionAttributes.putDouble(key, rs.getDouble("DOUBLE_VAL"));
+				}
+				else if (type == AttributeType.OBJECT) {
+					executionAttributes.putLong(key, rs.getLong("OBJECT_VAL"));
+				}
+				else {
+					throw new BatchCriticalException("Invalid type found: [" + typeCd + "] for execution id: ["
+							+ executionId + "]");
+				}
+			}
+		};
+
+		getJdbcTemplate().query(getQuery(FIND_STEP_EXECUTION_ATTRS), new Object[] { executionId }, callback);
+
+		return executionAttributes;
+	}
+
+	/**
+	 * Get StepExecution for the given step. Due to the nature of statistics,
+	 * they will not be returned with reconstituted object.
+	 * 
+	 * @see StepDao#getStepExecution(Long)
+	 * @throws IllegalArgumentException if id is null.
+	 */
+	public List findStepExecutions(final StepInstance step) {
+
+		Assert.notNull(step, "Step cannot be null.");
+		Assert.notNull(step.getId(), "Step id cannot be null.");
+
+		RowMapper rowMapper = new StepExecutionRowMapper(step);
+
+		return getJdbcTemplate().query(getQuery(FIND_STEP_EXECUTIONS), new Object[] { step.getId() }, rowMapper);
+	}
+
+	public StepExecution getStepExecution(Long stepExecutionId, StepInstance stepInstance) {
+
+		Assert.notNull(stepExecutionId, "Step Execution id must not be null");
+
+		RowMapper rowMapper = new StepExecutionRowMapper(stepInstance);
+
+		List executions = getJdbcTemplate().query(getQuery(GET_STEP_EXECUTION), new Object[] { stepExecutionId }, rowMapper);
+
+		StepExecution stepExecution;
+		if (executions.size() == 1) {
+			stepExecution = (StepExecution) executions.get(0);
+		}
+		else if (executions.size() == 0) {
+			stepExecution = null;
+		}
+		else {
+			throw new IncorrectResultSizeDataAccessException("Only one StepExecution may exist for given id: ["
+					+ stepExecutionId + "]", 1, executions.size());
+		}
+
+		return stepExecution;
+	}
+
+	public int getStepExecutionCount(StepInstance step) {
+
+		Object[] parameters = new Object[] { step.getId() };
+
+		return getJdbcTemplate().queryForInt(getQuery(GET_STEP_EXECUTION_COUNT), parameters);
+	}
+
+	/**
+	 * Insert execution attributes. A lob creator must be used, since any
+	 * attributes that don't match a provided type must be serialized into a
+	 * blob.
+	 */
+	public void saveExecutionAttributes(final Long executionId, final ExecutionAttributes executionAttributes) {
+
+		Assert.notNull(executionId, "ExecutionId must not be null.");
+		Assert.notNull(executionAttributes, "The ExecutionAttributes must not be null.");
+
+		for (Iterator it = executionAttributes.entrySet().iterator(); it.hasNext();) {
+			Entry entry = (Entry) it.next();
+			final String key = entry.getKey().toString();
+			final Object value = entry.getValue();
+
+			if (value instanceof String) {
+				insertExecutionAttribute(executionId, key, value, AttributeType.STRING);
+			}
+			else if (value instanceof Double) {
+				insertExecutionAttribute(executionId, key, value, AttributeType.DOUBLE);
+			}
+			else if (value instanceof Long) {
+				insertExecutionAttribute(executionId, key, value, AttributeType.LONG);
+			}
+			else {
+				insertExecutionAttribute(executionId, key, value, AttributeType.OBJECT);
+			}
+		}
+	}
+
+	private void insertExecutionAttribute(final Long executionId, final String key, final Object value,
+			final AttributeType type) {
+		PreparedStatementCallback callback = new AbstractLobCreatingPreparedStatementCallback(lobHandler) {
+
+			protected void setValues(PreparedStatement ps, LobCreator lobCreator) throws SQLException,
+					DataAccessException {
+
+				ps.setLong(1, executionId.longValue());
+				ps.setString(3, key);
+				if (type == AttributeType.STRING) {
+					ps.setString(2, AttributeType.STRING.toString());
+					ps.setString(4, value.toString());
+					ps.setDouble(5, 0.0);
+					ps.setLong(6, 0);
+					lobCreator.setBlobAsBytes(ps, 7, null);
+				}
+				else if (type == AttributeType.DOUBLE) {
+					ps.setString(2, AttributeType.DOUBLE.toString());
+					ps.setString(4, null);
+					ps.setDouble(5, ((Double) value).doubleValue());
+					ps.setLong(6, 0);
+					lobCreator.setBlobAsBytes(ps, 7, null);
+				}
+				else if (type == AttributeType.LONG) {
+					ps.setString(2, AttributeType.LONG.toString());
+					ps.setString(4, null);
+					ps.setDouble(5, 0.0);
+					ps.setLong(6, ((Long) value).longValue());
+					lobCreator.setBlobAsBytes(ps, 7, null);
+				}
+				else {
+					ps.setString(2, AttributeType.OBJECT.toString());
+					ps.setString(4, null);
+					ps.setDouble(5, 0.0);
+					ps.setLong(6, 0);
+					lobCreator.setBlobAsBytes(ps, 7, SerializationUtils.serialize((Serializable) value));
+				}
+			}
+		};
+
+		getJdbcTemplate().execute(getQuery(INSERT_STEP_EXECUTION_ATTRS), callback);
+	}
+
+	/**
+	 * Save a StepExecution. A unique id will be generated by the
+	 * stepExecutionIncrementor, and then set in the StepExecution. All values
+	 * will then be stored via an INSERT statement.
+	 * 
+	 * @see StepDao#saveStepExecution(StepExecution)
+	 */
+	public void saveStepExecution(StepExecution stepExecution) {
+
+		validateStepExecution(stepExecution);
+
+		cascadeJobExecution(stepExecution.getJobExecution());
+
+		stepExecution.setId(new Long(stepExecutionIncrementer.nextLongValue()));
+		stepExecution.incrementVersion(); // should be 0 now
+		Object[] parameters = new Object[] { stepExecution.getId(), stepExecution.getVersion(),
+				stepExecution.getStepId(), stepExecution.getJobExecutionId(), stepExecution.getStartTime(),
+				stepExecution.getEndTime(), stepExecution.getStatus().toString(), stepExecution.getCommitCount(),
+				stepExecution.getTaskCount(),
+				PropertiesConverter.propertiesToString(stepExecution.getExecutionAttributes().getProperties()),
+				stepExecution.getExitStatus().isContinuable() ? "Y" : "N", stepExecution.getExitStatus().getExitCode(),
+				stepExecution.getExitStatus().getExitDescription() };
+		getJdbcTemplate().update(getQuery(SAVE_STEP_EXECUTION), parameters, new int[] { Types.INTEGER, Types.INTEGER,
+				Types.INTEGER, Types.INTEGER, Types.TIMESTAMP, Types.TIMESTAMP, Types.VARCHAR, Types.INTEGER,
+				Types.INTEGER, Types.VARCHAR, Types.CHAR, Types.VARCHAR, Types.VARCHAR });
+	}
+
+	private void cascadeJobExecution(JobExecution jobExecution) {
+		if (jobExecution.getId() != null) {
+			// assume already saved...
+			return;
+		}
+		jobDao.saveJobExecution(jobExecution);
+	}
+
+	/**
+	 * Validate StepExecution. At a minimum, JobId, StartTime, and Status cannot
+	 * be null. EndTime can be null for an unfinished job.
+	 * 
+	 * @param jobExecution
+	 * @throws IllegalArgumentException
+	 */
+	private void validateStepExecution(StepExecution stepExecution) {
+		Assert.notNull(stepExecution);
+		Assert.notNull(stepExecution.getStepId(), "StepExecution Step-Id cannot be null.");
+		Assert.notNull(stepExecution.getStartTime(), "StepExecution start time cannot be null.");
+		Assert.notNull(stepExecution.getStatus(), "StepExecution status cannot be null.");
+	}
+
+	/**
+	 * update execution attributes. A lob creator must be used, since any
+	 * attributes that don't match a provided type must be serialized into a
+	 * blob.
+	 * 
+	 * @see {@link LobCreator}
+	 */
+	public void updateExecutionAttributes(final Long executionId, ExecutionAttributes executionAttributes) {
+
+		Assert.notNull(executionId, "ExecutionId must not be null.");
+		Assert.notNull(executionAttributes, "The ExecutionAttributes must not be null.");
+
+		for (Iterator it = executionAttributes.entrySet().iterator(); it.hasNext();) {
+			Entry entry = (Entry) it.next();
+			final String key = entry.getKey().toString();
+			final Object value = entry.getValue();
+
+			if (value instanceof String) {
+				updateExecutionAttribute(executionId, key, value, AttributeType.STRING);
+			}
+			else if (value instanceof Double) {
+				updateExecutionAttribute(executionId, key, value, AttributeType.DOUBLE);
+			}
+			else if (value instanceof Long) {
+				updateExecutionAttribute(executionId, key, value, AttributeType.LONG);
+			}
+			else {
+				updateExecutionAttribute(executionId, key, value, AttributeType.OBJECT);
+			}
+		}
+	}
+
+	private void updateExecutionAttribute(final Long executionId, final String key, final Object value,
+			final AttributeType type) {
+
+		PreparedStatementCallback callback = new AbstractLobCreatingPreparedStatementCallback(lobHandler) {
+
+			protected void setValues(PreparedStatement ps, LobCreator lobCreator) throws SQLException,
+					DataAccessException {
+
+				ps.setLong(6, executionId.longValue());
+				ps.setString(7, key);
+				if (type == AttributeType.STRING) {
+					ps.setString(1, AttributeType.STRING.toString());
+					ps.setString(2, value.toString());
+					ps.setDouble(3, 0.0);
+					ps.setLong(4, 0);
+					lobCreator.setBlobAsBytes(ps, 5, null);
+				}
+				else if (type == AttributeType.DOUBLE) {
+					ps.setString(1, AttributeType.DOUBLE.toString());
+					ps.setString(2, null);
+					ps.setDouble(3, ((Double) value).doubleValue());
+					ps.setLong(4, 0);
+					lobCreator.setBlobAsBytes(ps, 5, null);
+				}
+				else if (type == AttributeType.LONG) {
+					ps.setString(1, AttributeType.LONG.toString());
+					ps.setString(2, null);
+					ps.setDouble(3, 0.0);
+					ps.setLong(4, ((Long) value).longValue());
+					lobCreator.setBlobAsBytes(ps, 5, null);
+				}
+				else {
+					ps.setString(1, AttributeType.OBJECT.toString());
+					ps.setString(2, null);
+					ps.setDouble(3, 0.0);
+					ps.setLong(4, 0);
+					lobCreator.setBlobAsBytes(ps, 5, SerializationUtils.serialize((Serializable) value));
+				}
+			}
+		};
+
+		// LobCreating callbacks always return the affect row count for SQL DML
+		// statements, if less than 1 row
+		// is affected, then this row is new and should be inserted.
+		Integer affectedRows = (Integer) getJdbcTemplate().execute(getQuery(UPDATE_STEP_EXECUTION_ATTRS), callback);
+		if (affectedRows.intValue() < 1) {
+			insertExecutionAttribute(executionId, key, value, type);
+		}
+	}
+
+	public void updateStepExecution(StepExecution stepExecution) {
+
+		validateStepExecution(stepExecution);
+		Assert.notNull(stepExecution.getId(), "StepExecution Id cannot be null. StepExecution must saved"
+				+ " before it can be updated.");
+
+		// Do not check for existence of step execution considering
+		// it is saved at every commit point.
+
+		String exitDescription = stepExecution.getExitStatus().getExitDescription();
+		if (exitDescription != null && exitDescription.length() > EXIT_MESSAGE_LENGTH) {
+			exitDescription = exitDescription.substring(0, EXIT_MESSAGE_LENGTH);
+			logger.debug("Truncating long message before update of StepExecution: " + stepExecution);
+		}
+
+		// Attempt to prevent concurrent modification errors by blocking here if
+		// someone is already trying to do it.
+		synchronized (stepExecution) {
+
+			Integer version = new Integer(stepExecution.getVersion().intValue() + 1);
+			Object[] parameters = new Object[] { stepExecution.getStartTime(), stepExecution.getEndTime(),
+					stepExecution.getStatus().toString(), stepExecution.getCommitCount(), stepExecution.getTaskCount(),
+					PropertiesConverter.propertiesToString(stepExecution.getExecutionAttributes().getProperties()),
+					stepExecution.getExitStatus().isContinuable() ? "Y" : "N",
+					stepExecution.getExitStatus().getExitCode(), exitDescription, version, stepExecution.getId(),
+					stepExecution.getVersion() };
+			int count = getJdbcTemplate().update(getQuery(UPDATE_STEP_EXECUTION), parameters, new int[] { Types.TIMESTAMP,
+					Types.TIMESTAMP, Types.VARCHAR, Types.INTEGER, Types.INTEGER, Types.VARCHAR, Types.CHAR,
+					Types.VARCHAR, Types.VARCHAR, Types.INTEGER, Types.INTEGER, Types.INTEGER });
+
+			// Avoid concurrent modifications...
+			if (count == 0) {
+				throw new OptimisticLockingFailureException("Attempt to update step execution id="
+						+ stepExecution.getId() + " with wrong version (" + stepExecution.getVersion() + ")");
+			}
+
+			stepExecution.incrementVersion();
+
+		}
+	}
+
+	private class StepExecutionRowMapper implements RowMapper {
+
+		private final StepInstance stepInstance;
+
+		public StepExecutionRowMapper(StepInstance stepInstance) {
+			this.stepInstance = stepInstance;
+		}
+
+		public Object mapRow(ResultSet rs, int rowNum) throws SQLException {
+
+			JobExecution jobExecution = (JobExecution) getJdbcTemplate().queryForObject(
+					getQuery(JobExecutionRowMapper.GET_JOB_EXECUTION), new Object[] { new Long(rs.getLong(2)) },
+					new JobExecutionRowMapper(stepInstance.getJobInstance()));
+			StepExecution stepExecution = new StepExecution(stepInstance, jobExecution, new Long(rs.getLong(1)));
+			stepExecution.setStartTime(rs.getTimestamp(3));
+			stepExecution.setEndTime(rs.getTimestamp(4));
+			stepExecution.setStatus(BatchStatus.getStatus(rs.getString(5)));
+			stepExecution.setCommitCount(rs.getInt(6));
+			stepExecution.setTaskCount(rs.getInt(7));
+			stepExecution.setExecutionAttributes(new ExecutionAttributes(PropertiesConverter.stringToProperties(rs
+					.getString(8))));
+			stepExecution
+					.setExitStatus(new ExitStatus("Y".equals(rs.getString(9)), rs.getString(10), rs.getString(11)));
+			return stepExecution;
+		}
+
+	}
+
+	public void setLobHandler(LobHandler lobHandler) {
+		this.lobHandler = lobHandler;
+	}
+
+	public void setStepExecutionIncrementer(DataFieldMaxValueIncrementer stepExecutionIncrementer) {
+		this.stepExecutionIncrementer = stepExecutionIncrementer;
+	}
+
+	public void setJobDao(JobDao jobDao) {
+		this.jobDao = jobDao;
+	}
+
+	public void afterPropertiesSet() throws Exception {
+		Assert.notNull(stepExecutionIncrementer, "StepExecutionIncrementer cannot be null.");
+		Assert.notNull(jobDao, "JobDao cannot be null");
+	}
+
+	public static class AttributeType {
+
+		private final String type;
+
+		private AttributeType(String type) {
+			this.type = type;
+		}
+
+		public String toString() {
+			return type;
+		}
+
+		public static final AttributeType STRING = new AttributeType("STRING");
+
+		public static final AttributeType LONG = new AttributeType("LONG");
+
+		public static final AttributeType OBJECT = new AttributeType("OBJECT");
+
+		public static final AttributeType DOUBLE = new AttributeType("DOUBLE");
+
+		private static final AttributeType[] VALUES = { STRING, OBJECT, LONG, DOUBLE };
+
+		public static AttributeType getType(String typeAsString) {
+
+			for (int i = 0; i < VALUES.length; i++) {
+				if (VALUES[i].toString().equals(typeAsString)) {
+					return (AttributeType) VALUES[i];
+				}
+			}
+
+			return null;
+		}
+	}
+}
diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/JdbcStepInstanceDao.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/JdbcStepInstanceDao.java
new file mode 100644
index 000000000..c4f5f5e9f
--- /dev/null
+++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/JdbcStepInstanceDao.java
@@ -0,0 +1,183 @@
+package org.springframework.batch.execution.repository.dao;
+
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.List;
+
+import org.springframework.batch.core.domain.JobInstance;
+import org.springframework.batch.core.domain.StepExecution;
+import org.springframework.batch.core.domain.StepInstance;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.dao.IncorrectResultSizeDataAccessException;
+import org.springframework.jdbc.core.RowMapper;
+import org.springframework.jdbc.support.incrementer.DataFieldMaxValueIncrementer;
+import org.springframework.util.Assert;
+
+/**
+ * Jdbc implementation of {@link StepInstanceDao}.<br/>
+ * 
+ * Allows customization of the tables names used by Spring Batch for step meta
+ * data via a prefix property.<br/>
+ * 
+ * Uses sequences or tables (via Spring's {@link DataFieldMaxValueIncrementer}
+ * abstraction) to create all primary keys before inserting a new row. All
+ * objects are checked to ensure all fields to be stored are not null. If any
+ * are found to be null, an IllegalArgumentException will be thrown. This could
+ * be left to JdbcTemplate, however, the exception will be fairly vague, and
+ * fails to highlight which field caused the exception.<br/>
+ * 
+ * @author Lucas Ward
+ * @author Dave Syer
+ * @author Robert Kasanicky
+ * 
+ * @see StepInstanceDao
+ */
+public class JdbcStepInstanceDao extends AbstractJdbcBatchMetadataDao implements StepInstanceDao, InitializingBean {
+
+	private static final String CREATE_STEP = "INSERT into %PREFIX%STEP_INSTANCE(STEP_INSTANCE_ID, JOB_INSTANCE_ID, STEP_NAME) values (?, ?, ?)";
+
+	private static final String FIND_STEP = "SELECT STEP_INSTANCE_ID, LAST_STEP_EXECUTION_ID from %PREFIX%STEP_INSTANCE where JOB_INSTANCE_ID = ? "
+			+ "and STEP_NAME = ?";
+
+	private static final String FIND_STEPS = "SELECT STEP_INSTANCE_ID, LAST_STEP_EXECUTION_ID, STEP_NAME from %PREFIX%STEP_INSTANCE where JOB_INSTANCE_ID = ?";
+
+	private static final String UPDATE_STEP = "UPDATE %PREFIX%STEP_INSTANCE set LAST_STEP_EXECUTION_ID = ? where STEP_INSTANCE_ID = ?";
+
+	private DataFieldMaxValueIncrementer stepIncrementer;
+
+	private StepExecutionDao stepExecutionDao;
+
+	/**
+	 * Create a step with the given job's id, and the provided step name. A
+	 * unique id is created for the step using an incrementer. (@link
+	 * DataFieldMaxValueIncrementer)
+	 * 
+	 * @see StepDao#createStepInstance(JobInstance, String)
+	 * @throws IllegalArgumentException if job or stepName is null.
+	 */
+	public StepInstance createStepInstance(JobInstance job, String stepName) {
+
+		Assert.notNull(job, "Job cannot be null.");
+		Assert.notNull(stepName, "StepName cannot be null.");
+
+		Long stepId = new Long(stepIncrementer.nextLongValue());
+		Object[] parameters = new Object[] { stepId, job.getId(), stepName };
+		getJdbcTemplate().update(getQuery(CREATE_STEP), parameters);
+
+		StepInstance step = new StepInstance(job, stepName, stepId);
+		return step;
+	}
+
+	/**
+	 * 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
+	 * and null will be returned. If one step is found, it will be returned. If
+	 * anymore than one step is found, an exception is thrown.
+	 * 
+	 * @see StepDao#findStepInstance(Long, String)
+	 * @throws IllegalArgumentException if job, stepName, or job.id is null.
+	 * @throws IncorrectResultSizeDataAccessException if more than one step is
+	 * found.
+	 */
+	public StepInstance findStepInstance(JobInstance jobInstance, String stepName) {
+
+		Assert.notNull(jobInstance, "Job cannot be null.");
+		Assert.notNull(jobInstance.getId(), "Job ID cannot be null");
+		Assert.notNull(stepName, "StepName cannot be null");
+
+		Object[] parameters = new Object[] { jobInstance.getId(), stepName };
+
+		RowMapper rowMapper = new StepInstanceRowMapper(jobInstance, stepName);
+
+		List steps = getJdbcTemplate().query(getQuery(FIND_STEP), parameters, rowMapper);
+
+		if (steps.size() == 0) {
+			// No step found
+			return null;
+		}
+		else if (steps.size() == 1) {
+			StepInstance step = (StepInstance) steps.get(0);
+			return step;
+		}
+		else {
+			// This error will likely never be thrown, because there should
+			// never be two steps with the same name and JOB_INSTANCE_ID due to
+			// database
+			// constraints.
+			throw new IncorrectResultSizeDataAccessException("Step Invalid, multiple steps found for StepName:"
+					+ stepName + " and JobId:" + jobInstance.getId(), 1, steps.size());
+		}
+
+	}
+
+	/**
+	 * @see StepDao#findStepInstances(JobInstance)
+	 * 
+	 * Sql implementation which uses a RowMapper to populate a list of all rows
+	 * in the step table with the same JOB_INSTANCE_ID.
+	 * 
+	 * @throws IllegalArgumentException if jobId is null.
+	 */
+	public List findStepInstances(final JobInstance jobInstance) {
+
+		Assert.notNull(jobInstance, "Job cannot be null.");
+
+		Object[] parameters = new Object[] { jobInstance.getId() };
+
+		RowMapper rowMapper = new StepInstanceRowMapper(jobInstance, null);
+
+		return getJdbcTemplate().query(getQuery(FIND_STEPS), parameters, rowMapper);
+	}
+
+	/**
+	 * @see StepDao#updateStepInstance(StepInstance)
+	 * @throws IllegalArgumentException if step, or it's status and id is null.
+	 */
+	public void updateStepInstance(final StepInstance step) {
+
+		Assert.notNull(step, "Step cannot be null.");
+		Assert.notNull(step.getId(), "Step Id cannot be null.");
+
+		Object[] parameters = new Object[] { step.getLastExecution().getId(), step.getId() };
+
+		getJdbcTemplate().update(getQuery(UPDATE_STEP), parameters);
+	}
+
+	public void setStepIncrementer(DataFieldMaxValueIncrementer stepIncrementer) {
+		this.stepIncrementer = stepIncrementer;
+	}
+
+	private class StepInstanceRowMapper implements RowMapper {
+
+		private final JobInstance jobInstance;
+
+		private String stepName;
+
+		public StepInstanceRowMapper(JobInstance jobInstance, String stepName) {
+			this.jobInstance = jobInstance;
+			this.stepName = stepName;
+		}
+
+		public Object mapRow(ResultSet rs, int rowNum) throws SQLException {
+
+			if (stepName == null) {
+				stepName = rs.getString(3);
+			}
+			StepInstance stepInstance = new StepInstance(jobInstance, stepName, new Long(rs.getLong(1)));
+			StepExecution lastExecution = stepExecutionDao.getStepExecution(new Long(rs.getLong(2)), stepInstance);
+			stepInstance.setLastExecution(lastExecution);
+			return stepInstance;
+		}
+
+	}
+
+	public void setStepExecutionDao(StepExecutionDao stepExecutionDao) {
+		this.stepExecutionDao = stepExecutionDao;
+	}
+
+	public void afterPropertiesSet() throws Exception {
+		Assert.notNull(stepIncrementer, "StepIncrementer cannot be null.");
+		Assert.notNull(stepExecutionDao, "StepExecutionDao cannot be null.");
+	}
+
+}
diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/StepDao.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/StepDao.java
index f766ab272..e23c6d8f8 100644
--- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/StepDao.java
+++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/StepDao.java
@@ -16,13 +16,6 @@
 
 package org.springframework.batch.execution.repository.dao;
 
-import java.util.List;
-
-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.item.ExecutionAttributes;
-import org.springframework.dao.IncorrectResultSizeDataAccessException;
 
 /**
  * Data access object for steps.
@@ -30,117 +23,6 @@ import org.springframework.dao.IncorrectResultSizeDataAccessException;
  * @author Lucas Ward
  * 
  */
-public interface StepDao {
+public interface StepDao extends StepInstanceDao, StepExecutionDao{
 
-	/**
-	 * Find a step with the given JobId and Step Name. Return null if none are
-	 * found.
-	 * 
-	 * @param jobInstance
-	 * @param stepName
-	 * @return StepInstance
-	 */
-	StepInstance findStepInstance(JobInstance jobInstance, String stepName);
-
-	/**
-	 * Find all StepInstances of the given JobInstance.
-	 * 
-	 * @param jobInstance the job to use as a search key
-	 * @return list of {@link StepInstance}
-	 */
-	List findStepInstances(JobInstance jobInstance);
-
-	/**
-	 * Create a StepInstance for the given name and JobInstance.
-	 * 
-	 * @param jobInstance
-	 * @param stepName
-	 * 
-	 * @return
-	 */
-	StepInstance createStepInstance(JobInstance jobInstance, String stepName);
-
-	/**
-	 * Update an existing StepInstance.
-	 * 
-	 * Preconditions: StepInstance must have an ID.
-	 * 
-	 * @param job
-	 */
-	void updateStepInstance(StepInstance stepInstance);
-
-	/**
-	 * Save the given StepExecution.
-	 * 
-	 * Preconditions: Id must be null.
-	 * 
-	 * Postconditions: Id will be set to a unique Long.
-	 * 
-	 * @param stepExecution
-	 */
-	void saveStepExecution(StepExecution stepExecution);
-
-	/**
-	 * Update the given StepExecution
-	 * 
-	 * Preconditions: Id must not be null.
-	 * 
-	 * @param stepExecution
-	 */
-	void updateStepExecution(StepExecution stepExecution);
-
-	/**
-	 * Return the count of StepExecutions for the given {@link StepInstance}.
-	 * 
-	 * @param stepInstance the {@link StepInstance} to check for executions
-	 * @return the number of step executions for this step
-	 */
-	int getStepExecutionCount(StepInstance stepInstance);
-
-	/**
-	 * Return all StepExecutions for the given step.
-	 * 
-	 * @param stepInstance the step to use as a search key
-	 * @return list of stepExecutions
-	 */
-	List findStepExecutions(StepInstance stepInstance);
-
-	/**
-	 * Return a StepExecution for the given id.
-	 * 
-	 * @param stepExecutionId
-	 * @return {@link StepExecution} for the provided id.
-	 * @throws {@link IncorrectResultSizeDataAccessException} if more than one
-	 * execution is found.
-	 */
-	StepExecution getStepExecution(Long stepExecutionId, StepInstance stepInstance);
-
-	/**
-	 * Find all {@link ExecutionAttributes} for the given execution id.
-	 * 
-	 * @param executionId - Long id of the {@link StepExecution} that the
-	 * attributes belongs to.
-	 * @return attributes for the provided id. If none are found, an empty
-	 * {@link ExecutionAttributes} will be returned.
-	 * @throws IllegalArgumentException if the id is null.
-	 */
-	ExecutionAttributes findExecutionAttributes(final Long executionId);
-
-	/**
-	 * Save the provided {@link ExecutionAttributes} for the given executionId.
-	 * 
-	 * @param executionId to be saved
-	 * @param executionAttributes to be saved.
-	 * @throws IllegalArgumentException if the executionId or attributes are
-	 * null.
-	 */
-	void saveExecutionAttributes(final Long executionId, final ExecutionAttributes executionAttributes);
-
-	/**
-	 * Update the provided ExecutionAttributes.
-	 * 
-	 * @param executionId
-	 * @param executionAttributes
-	 */
-	void updateExecutionAttributes(final Long executionId, ExecutionAttributes executionAttributes);
 }
diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/StepExecutionDao.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/StepExecutionDao.java
new file mode 100644
index 000000000..56629b5fa
--- /dev/null
+++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/StepExecutionDao.java
@@ -0,0 +1,86 @@
+package org.springframework.batch.execution.repository.dao;
+
+import java.util.List;
+
+import org.springframework.batch.core.domain.StepExecution;
+import org.springframework.batch.core.domain.StepInstance;
+import org.springframework.batch.item.ExecutionAttributes;
+
+public interface StepExecutionDao {
+
+	/**
+	 * Save the given StepExecution.
+	 * 
+	 * Preconditions: Id must be null.
+	 * 
+	 * Postconditions: Id will be set to a unique Long.
+	 * 
+	 * @param stepExecution
+	 */
+	void saveStepExecution(StepExecution stepExecution);
+
+	/**
+	 * Update the given StepExecution
+	 * 
+	 * Preconditions: Id must not be null.
+	 * 
+	 * @param stepExecution
+	 */
+	void updateStepExecution(StepExecution stepExecution);
+
+	/**
+	 * Return the count of StepExecutions for the given {@link StepInstance}.
+	 * 
+	 * @param stepInstance the {@link StepInstance} to check for executions
+	 * @return the number of step executions for this step
+	 */
+	int getStepExecutionCount(StepInstance stepInstance);
+
+	/**
+	 * Return all StepExecutions for the given step.
+	 * 
+	 * @param stepInstance the step to use as a search key
+	 * @return list of stepExecutions
+	 */
+	List findStepExecutions(StepInstance stepInstance);
+
+	/**
+	 * Return a StepExecution for the given id.
+	 * 
+	 * @param stepExecutionId
+	 * @return {@link StepExecution} for the provided id.
+	 * @throws {@link IncorrectResultSizeDataAccessException} if more than one
+	 * execution is found.
+	 */
+	StepExecution getStepExecution(Long stepExecutionId, StepInstance stepInstance);
+
+	/**
+	 * Find all {@link ExecutionAttributes} for the given execution id.
+	 * 
+	 * @param executionId - Long id of the {@link StepExecution} that the
+	 * attributes belongs to.
+	 * @return attributes for the provided id. If none are found, an empty
+	 * {@link ExecutionAttributes} will be returned.
+	 * @throws IllegalArgumentException if the id is null.
+	 */
+	ExecutionAttributes findExecutionAttributes(final Long executionId);
+
+	/**
+	 * Save the provided {@link ExecutionAttributes} for the given executionId.
+	 * 
+	 * @param executionId to be saved
+	 * @param executionAttributes to be saved.
+	 * @throws IllegalArgumentException if the executionId or attributes are
+	 * null.
+	 */
+	void saveExecutionAttributes(final Long executionId, final ExecutionAttributes executionAttributes);
+
+	/**
+	 * Update the provided ExecutionAttributes.
+	 * 
+	 * @param executionId
+	 * @param executionAttributes
+	 */
+	void updateExecutionAttributes(final Long executionId, ExecutionAttributes executionAttributes);
+
+}
diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/StepInstanceDao.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/StepInstanceDao.java
new file mode 100644
index 000000000..cde87b7bc
--- /dev/null
+++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/StepInstanceDao.java
@@ -0,0 +1,46 @@
+package org.springframework.batch.execution.repository.dao;
+
+import java.util.List;
+
+import org.springframework.batch.core.domain.JobInstance;
+import org.springframework.batch.core.domain.StepInstance;
+
+public interface StepInstanceDao {
+
+	/**
+	 * Find a step with the given JobId and Step Name. Return null if none are
+	 * found.
+	 * 
+	 * @param jobInstance
+	 * @param stepName
+	 * @return StepInstance
+	 */
+	StepInstance findStepInstance(JobInstance jobInstance, String stepName);
+
+	/**
+	 * Find all StepInstances of the given JobInstance.
+	 * 
+	 * @param jobInstance the job to use as a search key
+	 * @return list of {@link StepInstance}
+	 */
+	List findStepInstances(JobInstance jobInstance);
+
+	/**
+	 * Create a StepInstance for the given name and JobInstance.
+	 * 
+	 * @param jobInstance
+	 * @param stepName
+	 * 
+	 * @return
+	 */
+	StepInstance createStepInstance(JobInstance jobInstance, String stepName);
+
+	/**
+	 * Update an existing StepInstance.
+	 * 
+	 * Preconditions: StepInstance must have an ID.
+	 * 
+	 * @param job
+	 */
+	void updateStepInstance(StepInstance stepInstance);
+}
diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/job/simple/SimpleJobTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/job/simple/SimpleJobTests.java
index 9c8dcfadd..6e25902b5 100644
--- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/job/simple/SimpleJobTests.java
+++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/job/simple/SimpleJobTests.java
@@ -34,7 +34,8 @@ import org.springframework.batch.execution.repository.SimpleJobRepository;
 import org.springframework.batch.execution.repository.dao.JobDao;
 import org.springframework.batch.execution.repository.dao.MapJobDao;
 import org.springframework.batch.execution.repository.dao.MapStepDao;
-import org.springframework.batch.execution.repository.dao.StepDao;
+import org.springframework.batch.execution.repository.dao.StepExecutionDao;
+import org.springframework.batch.execution.repository.dao.StepInstanceDao;
 import org.springframework.batch.execution.step.simple.SimpleStep;
 import org.springframework.batch.io.exception.BatchCriticalException;
 import org.springframework.batch.item.reader.AbstractItemReader;
@@ -52,7 +53,9 @@ public class SimpleJobTests extends TestCase {
 
 	private JobDao jobDao;
 
-	private StepDao stepDao;
+	private StepInstanceDao stepInstanceDao;
+	
+	private StepExecutionDao stepExecutionDao;
 
 	private List list = new ArrayList();
 
@@ -82,8 +85,9 @@ public class SimpleJobTests extends TestCase {
 		MapJobDao.clear();
 		MapStepDao.clear();
 		jobDao = new MapJobDao();
-		stepDao = new MapStepDao();
-		jobRepository = new SimpleJobRepository(jobDao, stepDao);
+		stepInstanceDao = new MapStepDao();
+		stepExecutionDao = new MapStepDao();
+		jobRepository = new SimpleJobRepository(jobDao, stepInstanceDao, stepExecutionDao);
 		job = new SimpleJob();
 		job.setJobRepository(jobRepository);
 
diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/launch/SimpleJobTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/launch/SimpleJobTests.java
index 3c476925b..112f64cbe 100644
--- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/launch/SimpleJobTests.java
+++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/launch/SimpleJobTests.java
@@ -49,7 +49,7 @@ public class SimpleJobTests extends TestCase {
 
 	private List recovered = new ArrayList();
 
-	private SimpleJobRepository repository = new SimpleJobRepository(new MapJobDao(), new MapStepDao());
+	private SimpleJobRepository repository = new SimpleJobRepository(new MapJobDao(), new MapStepDao(), new MapStepDao());
 
 	private List processed = new ArrayList();
 
diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/SimpleJobRepositoryTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/SimpleJobRepositoryTests.java
index 4155f7cb7..5f7c2d631 100644
--- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/SimpleJobRepositoryTests.java
+++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/SimpleJobRepositoryTests.java
@@ -86,7 +86,7 @@ public class SimpleJobRepositoryTests extends TestCase {
 		jobDao = (JobDao) jobDaoControl.getMock();
 		stepDao = (StepDao) stepDaoControl.getMock();
 
-		jobRepository = new SimpleJobRepository(jobDao, stepDao);
+		jobRepository = new SimpleJobRepository(jobDao, stepDao, stepDao);
 
 		jobParameters = new JobParametersBuilder().toJobParameters();
 		
diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/AbstractStepDaoTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/AbstractStepDaoTests.java
index 2b8aec7c2..c1fcad017 100644
--- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/AbstractStepDaoTests.java
+++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/AbstractStepDaoTests.java
@@ -35,7 +35,7 @@ import org.springframework.test.AbstractTransactionalDataSourceSpringContextTest
 import org.springframework.util.ClassUtils;
 
 /**
- * Test for StepDao. Because it is very reasonable to assume that there is a
+ * Tests for step persistence (StepInstanceDao and StepExecutionDao). Because it is very reasonable to assume that there is a
  * foreign key constraint on the JobId of a step, the JobDao is used to create
  * jobs, to have an id for creating steps.
  * 
@@ -45,8 +45,10 @@ import org.springframework.util.ClassUtils;
 public abstract class AbstractStepDaoTests extends AbstractTransactionalDataSourceSpringContextTests {
 
 	protected JobDao jobDao;
-
-	protected StepDao stepDao;
+	
+	protected StepInstanceDao stepInstanceDao;
+	
+	protected StepExecutionDao stepExecutionDao;
 
 	protected JobInstance jobInstance;
 
@@ -66,8 +68,12 @@ public abstract class AbstractStepDaoTests extends AbstractTransactionalDataSour
 		this.jobDao = jobDao;
 	}
 
-	public void setStepDao(StepDao stepDao) {
-		this.stepDao = stepDao;
+	public void setStepInstanceDao(StepInstanceDao stepInstanceDao) {
+		this.stepInstanceDao = stepInstanceDao;
+	}
+
+	public void setStepExecutionDao(StepExecutionDao stepExecutionDao) {
+		this.stepExecutionDao = stepExecutionDao;
 	}
 
 	/*
@@ -85,16 +91,16 @@ public abstract class AbstractStepDaoTests extends AbstractTransactionalDataSour
 	protected void onSetUpInTransaction() throws Exception {
 		Job job = new JobSupport("TestJob");
 		jobInstance = jobDao.createJobInstance(job.getName(), jobParameters);
-		step1 = stepDao.createStepInstance(jobInstance, "TestStep1");
-		step2 = stepDao.createStepInstance(jobInstance, "TestStep2");
+		step1 = stepInstanceDao.createStepInstance(jobInstance, "TestStep1");
+		step2 = stepInstanceDao.createStepInstance(jobInstance, "TestStep2");
 		jobExecution = new JobExecution(step2.getJobInstance());
 
 		stepExecution = new StepExecution(step1, jobExecution, null);
 		stepExecution.setStatus(BatchStatus.STARTED);
 		stepExecution.setStartTime(new Date(System.currentTimeMillis()));
-		stepDao.saveStepExecution(stepExecution);
+		stepExecutionDao.saveStepExecution(stepExecution);
 		step1.setLastExecution(stepExecution);
-		stepDao.updateStepInstance(step1);
+		stepInstanceDao.updateStepInstance(step1);
 		
 		executionAttributes = new ExecutionAttributes();
 		executionAttributes.putString("1", "testString1");
@@ -118,19 +124,19 @@ public abstract class AbstractStepDaoTests extends AbstractTransactionalDataSour
 
 	public void testFindStepNull() {
 
-		StepInstance step = stepDao.findStepInstance(jobInstance, "UnSavedStep");
+		StepInstance step = stepInstanceDao.findStepInstance(jobInstance, "UnSavedStep");
 		assertNull(step);
 	}
 
 	public void testFindStep() {
 
-		StepInstance tempStep = stepDao.findStepInstance(jobInstance, "TestStep1");
+		StepInstance tempStep = stepInstanceDao.findStepInstance(jobInstance, "TestStep1");
 		assertEquals(tempStep, step1);
 	}
 
 	public void testFindSteps() {
 
-		List steps = stepDao.findStepInstances(jobInstance);
+		List steps = stepInstanceDao.findStepInstances(jobInstance);
 		assertEquals(steps.size(), 2);
 		assertTrue(steps.contains(step1));
 		assertTrue(steps.contains(step2));
@@ -139,29 +145,29 @@ public abstract class AbstractStepDaoTests extends AbstractTransactionalDataSour
 	public void testFindStepsNotSaved() {
 
 		// no steps are saved for given id, empty list should be returned
-		List steps = stepDao.findStepInstances(new JobInstance(new Long(38922), jobParameters));
+		List steps = stepInstanceDao.findStepInstances(new JobInstance(new Long(38922), jobParameters));
 		assertEquals(steps.size(), 0);
 	}
 
 	public void testCreateStep() {
 
-		StepInstance step3 = stepDao.createStepInstance(jobInstance, "TestStep3");
-		StepInstance tempStep = stepDao.findStepInstance(jobInstance, "TestStep3");
+		StepInstance step3 = stepInstanceDao.createStepInstance(jobInstance, "TestStep3");
+		StepInstance tempStep = stepInstanceDao.findStepInstance(jobInstance, "TestStep3");
 		assertEquals(step3, tempStep);
 	}
 
 	public void testUpdateStepWithoutExecutionAttributes() {
 
-		stepDao.updateStepInstance(step1);
-		StepInstance tempStep = stepDao.findStepInstance(jobInstance, step1.getName());
+		stepInstanceDao.updateStepInstance(step1);
+		StepInstance tempStep = stepInstanceDao.findStepInstance(jobInstance, step1.getName());
 		assertEquals(tempStep, step1);
 	}
 
 	public void testUpdateStepWithExecutionAttributes() {
 
-		stepDao.saveExecutionAttributes(step1.getId(), executionAttributes);
-		StepInstance tempStep = stepDao.findStepInstance(jobInstance, step1.getName());
-		ExecutionAttributes tempAttributes = stepDao.findExecutionAttributes(step1.getId());
+		stepExecutionDao.saveExecutionAttributes(step1.getId(), executionAttributes);
+		StepInstance tempStep = stepInstanceDao.findStepInstance(jobInstance, step1.getName());
+		ExecutionAttributes tempAttributes = stepExecutionDao.findExecutionAttributes(step1.getId());
 		assertEquals(tempStep, step1);
 		assertEquals(executionAttributes, tempAttributes);
 	}
@@ -174,8 +180,8 @@ public abstract class AbstractStepDaoTests extends AbstractTransactionalDataSour
 		execution.setExecutionAttributes(new ExecutionAttributes(PropertiesConverter.stringToProperties("key1=0,key2=5")));
 		execution.setExitStatus(new ExitStatus(false, ExitCodeExceptionClassifier.FATAL_EXCEPTION,
 				"java.lang.Exception"));
-		stepDao.saveStepExecution(execution);
-		List executions = stepDao.findStepExecutions(step2);
+		stepExecutionDao.saveStepExecution(execution);
+		List executions = stepExecutionDao.findStepExecutions(step2);
 		assertEquals(1, executions.size());
 		StepExecution tempExecution = (StepExecution) executions.get(0);
 		assertEquals(execution, tempExecution);
@@ -192,8 +198,8 @@ public abstract class AbstractStepDaoTests extends AbstractTransactionalDataSour
 		stepExecution.setExecutionAttributes(new ExecutionAttributes());
 		stepExecution.setExitStatus(new ExitStatus(false, ExitCodeExceptionClassifier.FATAL_EXCEPTION,
 				"java.lang.Exception"));
-		stepDao.updateStepExecution(stepExecution);
-		List executions = stepDao.findStepExecutions(step1);
+		stepExecutionDao.updateStepExecution(stepExecution);
+		List executions = stepExecutionDao.findStepExecutions(step1);
 		assertEquals(1, executions.size());
 		StepExecution tempExecution = (StepExecution) executions.get(0);
 		assertEquals(stepExecution, tempExecution);
@@ -203,7 +209,7 @@ public abstract class AbstractStepDaoTests extends AbstractTransactionalDataSour
 	public void testUpdateStepExecutionWithNullId() {
 		StepExecution stepExecution = new StepExecution(null, null, null);
 		try {
-			stepDao.updateStepExecution(stepExecution);
+			stepExecutionDao.updateStepExecution(stepExecution);
 			fail("Expected IllegalArgumentException");
 		}
 		catch (IllegalArgumentException ex) {
@@ -213,22 +219,22 @@ public abstract class AbstractStepDaoTests extends AbstractTransactionalDataSour
 
 	public void testGetStepExecutionCountForNoExecutions() {
 
-		int executionCount = stepDao.getStepExecutionCount(step2);
+		int executionCount = stepExecutionDao.getStepExecutionCount(step2);
 		assertEquals(executionCount, 0);
 	}
 
 	public void testIncrementStepExecutionCount() {
 
-		assertEquals(1, stepDao.getStepExecutionCount(step1));
+		assertEquals(1, stepExecutionDao.getStepExecutionCount(step1));
 		StepExecution execution = new StepExecution(step1, new JobExecution(step1.getJobInstance(), new Long(123)),
 				null);
-		stepDao.saveStepExecution(execution);
-		assertEquals(2, stepDao.getStepExecutionCount(step1));
+		stepExecutionDao.saveStepExecution(execution);
+		assertEquals(2, stepExecutionDao.getStepExecutionCount(step1));
 	}
 
 	public void testUpdateStepExecutionVersion() throws Exception {
 		int before = stepExecution.getVersion().intValue();
-		stepDao.updateStepExecution(stepExecution);
+		stepExecutionDao.updateStepExecution(stepExecution);
 		int after = stepExecution.getVersion().intValue();
 		assertEquals("StepExecution version not updated", before + 1, after);
 	}
@@ -237,7 +243,7 @@ public abstract class AbstractStepDaoTests extends AbstractTransactionalDataSour
 		stepExecution.incrementVersion(); // not really allowed outside dao
 		// code
 		try {
-			stepDao.updateStepExecution(stepExecution);
+			stepExecutionDao.updateStepExecution(stepExecution);
 			fail("Expected OptimisticLockingFailureException");
 		}
 		catch (OptimisticLockingFailureException e) {
@@ -251,12 +257,12 @@ public abstract class AbstractStepDaoTests extends AbstractTransactionalDataSour
 	
 	public void testSaveExecutionAttributes(){
 	
-		stepDao.saveExecutionAttributes(stepExecution.getId(), executionAttributes);
-		ExecutionAttributes attributes = stepDao.findExecutionAttributes(stepExecution.getId());
+		stepExecutionDao.saveExecutionAttributes(stepExecution.getId(), executionAttributes);
+		ExecutionAttributes attributes = stepExecutionDao.findExecutionAttributes(stepExecution.getId());
 		assertEquals(executionAttributes, attributes);
 		executionAttributes.putString("newString", "newString");
-		stepDao.updateExecutionAttributes(stepExecution.getId(), executionAttributes);
-		attributes = stepDao.findExecutionAttributes(stepExecution.getId());
+		stepExecutionDao.updateExecutionAttributes(stepExecution.getId(), executionAttributes);
+		attributes = stepExecutionDao.findExecutionAttributes(stepExecution.getId());
 		assertEquals(executionAttributes, attributes);
 	}
 		
diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/JdbcStepDaoPrefixTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/JdbcStepDaoPrefixTests.java
index 9fb053f2b..245181dde 100644
--- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/JdbcStepDaoPrefixTests.java
+++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/JdbcStepDaoPrefixTests.java
@@ -26,7 +26,9 @@ import org.springframework.jdbc.support.incrementer.DataFieldMaxValueIncrementer
  */
 public class JdbcStepDaoPrefixTests extends TestCase {
 	
-	private JdbcStepDao stepDao;
+	private JdbcStepInstanceDao stepInstanceDao;
+	
+	private JdbcStepExecutionDao stepExecutionDao;
 	
 	MockJdbcTemplate jdbcTemplate = new MockJdbcTemplate();
 	
@@ -42,14 +44,16 @@ public class JdbcStepDaoPrefixTests extends TestCase {
 	protected void setUp() throws Exception {
 		super.setUp();
 		
-		stepDao = new JdbcStepDao();
-		stepDao.setJobDao(new MapJobDao());
+		stepInstanceDao = new JdbcStepInstanceDao();
+		stepExecutionDao = new JdbcStepExecutionDao();
+		stepExecutionDao.setJobDao(new MapJobDao());
 		stepExecutionIncrementer = (DataFieldMaxValueIncrementer)stepExecutionIncrementerControl.getMock();
 		stepIncrementer = (DataFieldMaxValueIncrementer)stepIncrementerControl.getMock();
 		
-		stepDao.setJdbcTemplate(jdbcTemplate);
-		stepDao.setStepExecutionIncrementer(stepExecutionIncrementer);
-		stepDao.setStepIncrementer(stepIncrementer);
+		stepInstanceDao.setJdbcTemplate(jdbcTemplate);
+		stepExecutionDao.setJdbcTemplate(jdbcTemplate);
+		stepExecutionDao.setStepExecutionIncrementer(stepExecutionIncrementer);
+		stepInstanceDao.setStepIncrementer(stepIncrementer);
 		stepExecution.setId(new Long(1));
 		stepExecution.incrementVersion();
 		step.setLastExecution(stepExecution);
@@ -59,51 +63,51 @@ public class JdbcStepDaoPrefixTests extends TestCase {
 	}
 	
 	public void testModifiedUpdateStepExecution(){
-		stepDao.setTablePrefix("FOO_");
-		stepDao.updateStepExecution(stepExecution);
+		stepExecutionDao.setTablePrefix("FOO_");
+		stepExecutionDao.updateStepExecution(stepExecution);
 		assertTrue(jdbcTemplate.getSqlStatement().indexOf("FOO_STEP_EXECUTION") != -1);
 	}
 	
 	public void testModifiedSaveStepExecution(){
-		stepDao.setTablePrefix("FOO_");
+		stepExecutionDao.setTablePrefix("FOO_");
 		stepExecutionIncrementer.nextLongValue();
 		stepExecutionIncrementerControl.setReturnValue(1);
 		stepExecutionIncrementerControl.replay();
-		stepDao.saveStepExecution(stepExecution);
+		stepExecutionDao.saveStepExecution(stepExecution);
 		assertTrue(jdbcTemplate.getSqlStatement().indexOf("FOO_STEP_EXECUTION") != -1);
 	}
 	
 	public void testModifiedFindStepExecutions(){
-		stepDao.setTablePrefix("FOO_");
-		stepDao.findStepExecutions(step);
+		stepExecutionDao.setTablePrefix("FOO_");
+		stepExecutionDao.findStepExecutions(step);
 		assertTrue(jdbcTemplate.getSqlStatement().indexOf("FOO_STEP_EXECUTION") != -1);
 	}
 	
 	public void testModifiedUpdateStep(){
-		stepDao.setTablePrefix("FOO_");
-		stepDao.updateStepInstance(step);
+		stepInstanceDao.setTablePrefix("FOO_");
+		stepInstanceDao.updateStepInstance(step);
 		assertTrue(jdbcTemplate.getSqlStatement().indexOf("FOO_STEP") != -1);
 	}
 	
 	public void testModifiedCreateStep(){
-		stepDao.setTablePrefix("FOO_");
+		stepInstanceDao.setTablePrefix("FOO_");
 		stepIncrementer.nextLongValue();
 		stepIncrementerControl.setReturnValue(1);
 		stepIncrementerControl.replay();
-		stepDao.createStepInstance(job, "test");
+		stepInstanceDao.createStepInstance(job, "test");
 		assertTrue(jdbcTemplate.getSqlStatement().indexOf("FOO_STEP") != -1);
 	}
 	
 	public void testModifiedFindSteps(){
-		stepDao.setTablePrefix("FOO_");
-		stepDao.findStepInstances(new JobInstance(new Long(1), new JobParameters()));
+		stepInstanceDao.setTablePrefix("FOO_");
+		stepInstanceDao.findStepInstances(new JobInstance(new Long(1), new JobParameters()));
 		assertTrue(jdbcTemplate.getSqlStatement().indexOf("FOO_STEP") != -1);
 	}
 	
 	public void testModifiedFindStep(){
-		stepDao.setTablePrefix("FOO_");
+		stepInstanceDao.setTablePrefix("FOO_");
 		try{
-			stepDao.findStepInstance(job, "test");
+			stepInstanceDao.findStepInstance(job, "test");
 		}
 		catch(NullPointerException ex){
 			//It's going to throw a NullPointerException because the MockJdbcTemplate
@@ -115,7 +119,7 @@ public class JdbcStepDaoPrefixTests extends TestCase {
 	
 	public void testDefaultFindStep(){
 		try{
-			stepDao.findStepInstance(job, "test");
+			stepInstanceDao.findStepInstance(job, "test");
 		}
 		catch(NullPointerException ex){
 			//It's going to throw a NullPointerException because the MockJdbcTemplate
@@ -127,7 +131,7 @@ public class JdbcStepDaoPrefixTests extends TestCase {
 	}
 	
 	public void testDefaultFindSteps(){
-		stepDao.findStepInstances(new JobInstance(new Long(1), new JobParameters()));
+		stepInstanceDao.findStepInstances(new JobInstance(new Long(1), new JobParameters()));
 		assertTrue(jdbcTemplate.getSqlStatement().indexOf("BATCH_STEP") != -1);
 	}
 	
@@ -135,17 +139,17 @@ public class JdbcStepDaoPrefixTests extends TestCase {
 		stepIncrementer.nextLongValue();
 		stepIncrementerControl.setReturnValue(1);
 		stepIncrementerControl.replay();
-		stepDao.createStepInstance(job, "test");
+		stepInstanceDao.createStepInstance(job, "test");
 		assertTrue(jdbcTemplate.getSqlStatement().indexOf("BATCH_STEP") != -1);
 	}
 	
 	public void testDefaultUpdateStep(){
-		stepDao.updateStepInstance(step);
+		stepInstanceDao.updateStepInstance(step);
 		assertTrue(jdbcTemplate.getSqlStatement().indexOf("BATCH_STEP") != -1);
 	}
 	
 	public void testDefaultFindStepExecutions(){
-		stepDao.findStepExecutions(step);
+		stepExecutionDao.findStepExecutions(step);
 		assertTrue(jdbcTemplate.getSqlStatement().indexOf("BATCH_STEP_EXECUTION") != -1);
 	}
 	
@@ -153,12 +157,12 @@ public class JdbcStepDaoPrefixTests extends TestCase {
 		stepExecutionIncrementer.nextLongValue();
 		stepExecutionIncrementerControl.setReturnValue(1);
 		stepExecutionIncrementerControl.replay();
-		stepDao.saveStepExecution(stepExecution);
+		stepExecutionDao.saveStepExecution(stepExecution);
 		assertTrue(jdbcTemplate.getSqlStatement().indexOf("BATCH_STEP_EXECUTION") != -1);
 	}	
 
 	public void testDefaultUpdateStepExecution(){
-		stepDao.updateStepExecution(stepExecution);
+		stepExecutionDao.updateStepExecution(stepExecution);
 		assertTrue(jdbcTemplate.getSqlStatement().indexOf("BATCH_STEP_EXECUTION") != -1);
 	}
 	
diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/JdbcStepDaoTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/JdbcStepDaoTests.java
index d921a513c..65bfb1010 100644
--- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/JdbcStepDaoTests.java
+++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/JdbcStepDaoTests.java
@@ -11,11 +11,13 @@ public class JdbcStepDaoTests extends AbstractStepDaoTests {
 	private static final String LONG_STRING = JdbcJobDaoTests.LONG_STRING;
 
 	protected void onSetUpBeforeTransaction() throws Exception {
-		((JdbcStepDao) stepDao).setTablePrefix(JdbcJobDao.DEFAULT_TABLE_PREFIX);
+		((JdbcStepInstanceDao) stepInstanceDao).setTablePrefix(JdbcJobDao.DEFAULT_TABLE_PREFIX);
+		((JdbcStepExecutionDao) stepExecutionDao).setTablePrefix(JdbcJobDao.DEFAULT_TABLE_PREFIX);
 	}
 
 	public void testTablePrefix() throws Exception {
-		((JdbcStepDao) stepDao).setTablePrefix("FOO_");
+		((JdbcStepInstanceDao) stepInstanceDao).setTablePrefix("FOO_");
+		((JdbcStepExecutionDao) stepExecutionDao).setTablePrefix("FOO_");
 		try {
 			testCreateStep();
 			fail("Expected DataAccessException");
@@ -28,7 +30,7 @@ public class JdbcStepDaoTests extends AbstractStepDaoTests {
 
 		assertTrue(LONG_STRING.length()>250);
 		stepExecution.setExitStatus(ExitStatus.FINISHED.addExitDescription(LONG_STRING));
-		stepDao.updateStepExecution(stepExecution);
+		stepExecutionDao.updateStepExecution(stepExecution);
 
 		List executions = jdbcTemplate.queryForList(
 				"SELECT * FROM BATCH_STEP_EXECUTION where STEP_INSTANCE_ID=?",
diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorTests.java
index fc8022065..dd768f309 100644
--- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorTests.java
+++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorTests.java
@@ -202,7 +202,7 @@ public class SimpleStepExecutorTests extends TestCase {
 
 	public void testRepository() throws Exception {
 
-		SimpleJobRepository repository = new SimpleJobRepository(new MapJobDao(), new MapStepDao());
+		SimpleJobRepository repository = new SimpleJobRepository(new MapJobDao(), new MapStepDao(), new MapStepDao());
 		stepExecutor.setRepository(repository);
 
 		StepInstance step = new StepInstance(new Long(1));
diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/StepExecutorInterruptionTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/StepExecutorInterruptionTests.java
index 1de500900..a9880b881 100644
--- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/StepExecutorInterruptionTests.java
+++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/StepExecutorInterruptionTests.java
@@ -55,7 +55,7 @@ public class StepExecutorInterruptionTests extends TestCase {
 
 	public void setUp() throws Exception {
 
-		jobRepository = new SimpleJobRepository(jobDao, stepDao);
+		jobRepository = new SimpleJobRepository(jobDao, stepDao, stepDao);
 
 		JobSupport jobConfiguration = new JobSupport();
 		step = new RepeatOperationsStep();
diff --git a/spring-batch-execution/src/test/resources/org/springframework/batch/execution/repository/dao/sql-dao-test.xml b/spring-batch-execution/src/test/resources/org/springframework/batch/execution/repository/dao/sql-dao-test.xml
index 40e669f85..62906a0a2 100644
--- a/spring-batch-execution/src/test/resources/org/springframework/batch/execution/repository/dao/sql-dao-test.xml
+++ b/spring-batch-execution/src/test/resources/org/springframework/batch/execution/repository/dao/sql-dao-test.xml
@@ -11,11 +11,16 @@
 		<property name="jobExecutionIncrementer" ref="jobExecutionIncrementer" />
 	</bean>
 
-	<bean id="stepDao" class="org.springframework.batch.execution.repository.dao.JdbcStepDao" >
+	<bean id="stepInstanceDao" class="org.springframework.batch.execution.repository.dao.JdbcStepInstanceDao" >
 		<property name="jdbcTemplate" ref="jdbcTemplate" />
-		<property name="stepIncrementer" ref="stepIncrementer" />
+		<property name="stepIncrementer" ref="stepIncrementer" />
+		<property name="stepExecutionDao" ref="stepExecutionDao" />
+	</bean>
+	
+	<bean id="stepExecutionDao" class="org.springframework.batch.execution.repository.dao.JdbcStepExecutionDao" >
+		<property name="jdbcTemplate" ref="jdbcTemplate" />
 		<property name="stepExecutionIncrementer" ref="stepExecutionIncrementer" />
-		<property name="jobDao" ref="jobDao"/>
+		<property name="jobDao" ref="jobDao"/>
 	</bean>
 
 	<bean id="jdbcTemplate" class="org.springframework.jdbc.core.JdbcTemplate" >
diff --git a/spring-batch-execution/src/test/resources/simple-container-definition.xml b/spring-batch-execution/src/test/resources/simple-container-definition.xml
index aef82fe5c..fe2937f94 100644
--- a/spring-batch-execution/src/test/resources/simple-container-definition.xml
+++ b/spring-batch-execution/src/test/resources/simple-container-definition.xml
@@ -46,6 +46,7 @@
 	<bean id="simpleJobRepository"
 		class="org.springframework.batch.execution.repository.SimpleJobRepository">
 		<constructor-arg ref="jobDao" />
+		<constructor-arg ref="stepDao" />
 		<constructor-arg ref="stepDao" />
 	</bean>
 
diff --git a/spring-batch-samples/src/main/resources/simple-container-definition.xml b/spring-batch-samples/src/main/resources/simple-container-definition.xml
index cb8b1ce67..68610f5ed 100644
--- a/spring-batch-samples/src/main/resources/simple-container-definition.xml
+++ b/spring-batch-samples/src/main/resources/simple-container-definition.xml
@@ -46,7 +46,8 @@
 	<bean id="jobRepository"
 		class="org.springframework.batch.execution.repository.SimpleJobRepository">
 		<constructor-arg ref="jobDao" />
-		<constructor-arg ref="stepDao" />
+		<constructor-arg ref="stepInstanceDao" />
+		<constructor-arg ref="stepExecutionDao" />
 	</bean>
 
 	<bean id="jobDao" lazy-init="true"
@@ -57,13 +58,27 @@
 			ref="jobExecutionIncrementer" />
 	</bean>
 
-	<bean id="stepDao" lazy-init="true"
+	<!--bean id="stepDao" lazy-init="true"
 		class="org.springframework.batch.execution.repository.dao.JdbcStepDao">
 		<property name="jdbcTemplate" ref="jdbcTemplate" />
 		<property name="stepIncrementer" ref="stepIncrementer" />
 		<property name="stepExecutionIncrementer"
 			ref="stepExecutionIncrementer" />
 		<property name="jobDao" ref="jobDao" />
+	</bean-->
+	
+	<bean id="stepInstanceDao" lazy-init="true"
+		class="org.springframework.batch.execution.repository.dao.JdbcStepInstanceDao">
+		<property name="jdbcTemplate" ref="jdbcTemplate" />
+		<property name="stepIncrementer" ref="stepIncrementer" />
+		<property name="stepExecutionDao" ref="stepExecutionDao" />
+	</bean>
+	
+	<bean id="stepExecutionDao" lazy-init="true"
+		class="org.springframework.batch.execution.repository.dao.JdbcStepExecutionDao">
+		<property name="jdbcTemplate" ref="jdbcTemplate" />
+		<property name="stepExecutionIncrementer" ref="stepExecutionIncrementer" />
+		<property name="jobDao" ref="jobDao" />
 	</bean>
 
 	<bean id="mapJobDao" lazy-init="true"