diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/JdbcStepExecutionDao.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/JdbcStepExecutionDao.java
index 9592bae03..e1ff8bf10 100644
--- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/JdbcStepExecutionDao.java
+++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/JdbcStepExecutionDao.java
@@ -121,8 +121,8 @@ public class JdbcStepExecutionDao extends AbstractJdbcBatchMetadataDao implement
 	}
 
 	/**
-	 * Insert execution attributes. A lob creator must be used, since any
-	 * attributes that don't match a provided type must be serialized into a
+	 * Insert execution attributes. A {@link LobHandler} must be provided, since
+	 * any attributes that don't match a provided type are serialized into a
 	 * blob.
 	 */
 	public void saveExecutionContext(final StepExecution stepExecution) {
@@ -367,7 +367,7 @@ public class JdbcStepExecutionDao extends AbstractJdbcBatchMetadataDao implement
 	private class StepExecutionRowMapper implements RowMapper {
 
 		private final JobExecution jobExecution;
-		
+
 		private final Step step;
 
 		public StepExecutionRowMapper(JobExecution jobExecution, Step step) {
@@ -383,10 +383,9 @@ public class JdbcStepExecutionDao extends AbstractJdbcBatchMetadataDao implement
 			stepExecution.setStatus(BatchStatus.getStatus(rs.getString(5)));
 			stepExecution.setCommitCount(rs.getInt(6));
 			stepExecution.setTaskCount(rs.getInt(7));
-			stepExecution.setExecutionContext(new ExecutionContext(PropertiesConverter.stringToProperties(rs
-					.getString(8))));
 			stepExecution
 					.setExitStatus(new ExitStatus("Y".equals(rs.getString(9)), rs.getString(10), rs.getString(11)));
+			stepExecution.setExecutionContext(findExecutionContext(stepExecution));
 			return stepExecution;
 		}
 
diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemOrientedStep.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemOrientedStep.java
index 701590ec9..04adc66b6 100644
--- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemOrientedStep.java
+++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemOrientedStep.java
@@ -101,8 +101,6 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean {
 
 	private int commitInterval = 0;
 
-	private boolean saveExecutionContext = false;
-
 	/**
 	 * The {@link RepeatOperations} to use for the outer loop of the batch
 	 * processing. Should be set up by the caller through a factory. Defaults to
@@ -115,11 +113,11 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean {
 	}
 
 	/**
-	 * the {@link repeatoperations} to use for the inner loop of the batch
+	 * The {@link RepeatOperations} to use for the inner loop of the batch
 	 * processing. should be set up by the caller through a factory. defaults to
-	 * a plain {@link repeattemplate}.
+	 * a plain {@link RepeatTemplate}.
 	 * 
-	 * @param chunkoperations a {@link repeatoperations} instance.
+	 * @param chunkoperations a {@link RepeatOperations} instance.
 	 */
 	public void setChunkOperations(RepeatOperations chunkoperations) {
 		this.chunkOperations = chunkoperations;
@@ -256,7 +254,7 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean {
 			// the conversation in StepScope
 			stepContext.setAttribute(StepScope.ID_KEY, stepExecution.getJobExecution().getId());
 
-			if (saveExecutionContext && isRestart && lastStepExecution != null) {
+			if (isSaveExecutionContext() && isRestart && lastStepExecution != null) {
 				stepExecution.setExecutionContext(lastStepExecution.getExecutionContext());
 			}
 			else {
@@ -303,6 +301,7 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean {
 
 						}
 
+						itemReader.mark();
 						itemWriter.flush();
 						streamManager.commit(transaction);
 
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 28e2882b5..3ef400175 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
@@ -33,7 +33,6 @@ import org.springframework.batch.core.domain.StepSupport;
 import org.springframework.batch.core.runtime.ExitStatusExceptionClassifier;
 import org.springframework.batch.item.ExecutionContext;
 import org.springframework.batch.repeat.ExitStatus;
-import org.springframework.batch.support.PropertiesConverter;
 import org.springframework.dao.OptimisticLockingFailureException;
 import org.springframework.test.AbstractTransactionalDataSourceSpringContextTests;
 import org.springframework.util.ClassUtils;
@@ -128,18 +127,32 @@ public abstract class AbstractStepDaoTests extends AbstractTransactionalDataSour
 	}
 	
 	public void testSaveStepExecution() {
-
 		StepExecution execution = new StepExecution(step2, jobExecution, null);
 		execution.setStatus(BatchStatus.STARTED);
 		execution.setStartTime(new Date(System.currentTimeMillis()));
-		execution.setExecutionContext(new ExecutionContext(PropertiesConverter.stringToProperties("key1=0,key2=5")));
 		execution.setExitStatus(new ExitStatus(false, ExitStatusExceptionClassifier.FATAL_EXCEPTION,
 				"java.lang.Exception"));
 		stepExecutionDao.saveStepExecution(execution);
 		StepExecution retrievedExecution = stepExecutionDao.getStepExecution(jobExecution, step2);
 		assertNotNull(retrievedExecution);
 		assertEquals(execution, retrievedExecution);
-		assertEquals(execution.getExecutionContext().getString("key1"), retrievedExecution.getExecutionContext().getString("key1"));
+		assertEquals(execution.getExitStatus(), retrievedExecution.getExitStatus());
+	}
+
+	public void testSaveStepExecutionAndExecutionContext() {
+		StepExecution execution = new StepExecution(step2, jobExecution, null);
+		execution.setStatus(BatchStatus.STARTED);
+		execution.setStartTime(new Date(System.currentTimeMillis()));
+		execution.setExecutionContext(executionContext);
+		execution.setExitStatus(new ExitStatus(false, ExitStatusExceptionClassifier.FATAL_EXCEPTION,
+				"java.lang.Exception"));
+		stepExecutionDao.saveStepExecution(execution);
+		stepExecutionDao.saveExecutionContext(execution);
+		StepExecution retrievedExecution = stepExecutionDao.getStepExecution(jobExecution, step2);
+		assertNotNull(retrievedExecution);
+		assertEquals(execution, retrievedExecution);
+		assertEquals(execution.getExecutionContext().getString("1"), retrievedExecution.getExecutionContext().getString("1"));
+		assertEquals(execution.getExecutionContext().getLong("3"), retrievedExecution.getExecutionContext().getLong("3"));
 		assertEquals(execution.getExitStatus(), retrievedExecution.getExitStatus());
 	}
 
diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ExecutionContext.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ExecutionContext.java
index e8e6c1c3e..bc0be9ba0 100644
--- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ExecutionContext.java
+++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ExecutionContext.java
@@ -99,7 +99,7 @@ public class ExecutionContext {
 
 		if (!type.isInstance(value)) {
 			throw new ClassCastException("Value for key=[" + key + "] is not of type: [" + type
-					+ "], it is [" + (value == null ? null : value) + "]");
+					+ "], it is [" + (value == null ? null : "("+value.getClass()+")"+value) + "]");
 		}
 
 		return value;
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 eab62bc84..81b34e2d4 100644
--- a/spring-batch-samples/src/main/resources/simple-container-definition.xml
+++ b/spring-batch-samples/src/main/resources/simple-container-definition.xml
@@ -45,8 +45,8 @@
 
 	<bean id="jobRepository"
 		class="org.springframework.batch.execution.repository.SimpleJobRepository">
-		<constructor-arg ref="jobInstanceDao" />
-		<constructor-arg ref="jobExecutionDao" />
+		<constructor-arg ref="jobInstanceDao" />
+		<constructor-arg ref="jobExecutionDao" />
 		<constructor-arg ref="stepExecutionDao" />
 	</bean>
 
@@ -54,18 +54,20 @@
 		class="org.springframework.batch.execution.repository.dao.JdbcJobInstanceDao">
 		<property name="jdbcTemplate" ref="jdbcTemplate" />
 		<property name="jobIncrementer" ref="jobIncrementer" />
-	</bean>
-	
-	<bean id="jobExecutionDao" lazy-init="true"
-		class="org.springframework.batch.execution.repository.dao.JdbcJobExecutionDao">
-		<property name="jdbcTemplate" ref="jdbcTemplate" />
-		<property name="jobExecutionIncrementer" ref="jobExecutionIncrementer" />
-	</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" />
+	</bean>
+
+	<bean id="jobExecutionDao" lazy-init="true"
+		class="org.springframework.batch.execution.repository.dao.JdbcJobExecutionDao">
+		<property name="jdbcTemplate" ref="jdbcTemplate" />
+		<property name="jobExecutionIncrementer"
+			ref="jobExecutionIncrementer" />
+	</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" />
 	</bean>
 
 	<bean id="mapJobDao" lazy-init="true"
@@ -80,7 +82,8 @@
 	</bean>
 
 	<bean id="simpleJob"
-		class="org.springframework.batch.execution.job.simple.SimpleJob" abstract="true">
+		class="org.springframework.batch.execution.job.simple.SimpleJob"
+		abstract="true">
 		<property name="jobRepository" ref="jobRepository" />
 		<property name="restartable" value="true" />
 	</bean>
@@ -93,15 +96,15 @@
 		<property name="saveExecutionContext" value="true" />
 	</bean>
 
-	<bean id="abstractStep" class="org.springframework.batch.execution.step.ItemOrientedStep" abstract="true">
+	<bean id="simpleStep"
+		class="org.springframework.batch.execution.step.ItemOrientedStep"
+		abstract="true">
+		
 		<property name="transactionManager" ref="transactionManager" />
 		<property name="jobRepository" ref="jobRepository" />
 		<property name="allowStartIfComplete" value="true" />
 		<property name="saveExecutionContext" value="true" />
-	</bean>
 
-	<bean id="simpleStep" parent="abstractStep"
-		abstract="true">
 		<property name="exceptionHandler">
 			<bean
 				class="org.springframework.batch.repeat.exception.handler.SimpleLimitExceptionHandler">
@@ -109,9 +112,11 @@
 				<property name="useParent" value="true" />
 			</bean>
 		</property>
+
 		<property name="commitInterval" value="1" />
+
 	</bean>
-	
+
 	<bean id="customEditorConfigurer"
 		class="org.springframework.beans.factory.config.CustomEditorConfigurer">
 		<property name="customEditors">