diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/SimpleJobRepositoryIntegrationTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/SimpleJobRepositoryIntegrationTests.java
index 06fe2a3ed..eb196ba52 100644
--- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/SimpleJobRepositoryIntegrationTests.java
+++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/SimpleJobRepositoryIntegrationTests.java
@@ -4,10 +4,16 @@ import java.util.Date;
 import java.util.HashMap;
 import java.util.Map;
 
+import org.springframework.batch.core.domain.BatchStatus;
 import org.springframework.batch.core.domain.JobExecution;
 import org.springframework.batch.core.domain.JobParameters;
+import org.springframework.batch.core.domain.Step;
+import org.springframework.batch.core.domain.StepExecution;
+import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException;
 import org.springframework.batch.core.repository.JobRestartException;
 import org.springframework.batch.execution.job.JobSupport;
+import org.springframework.batch.execution.step.StepSupport;
+import org.springframework.batch.item.ExecutionContext;
 import org.springframework.test.AbstractTransactionalDataSourceSpringContextTests;
 import org.springframework.util.ClassUtils;
 
@@ -24,6 +30,10 @@ public class SimpleJobRepositoryIntegrationTests extends AbstractTransactionalDa
 
 	private SimpleJobRepository jobRepository;
 
+	private JobSupport job = new JobSupport("testJob");
+
+	private JobParameters jobParameters = new JobParameters();
+
 	public void setJobRepository(SimpleJobRepository jobRepository) {
 		this.jobRepository = jobRepository;
 	}
@@ -34,7 +44,6 @@ public class SimpleJobRepositoryIntegrationTests extends AbstractTransactionalDa
 	 */
 	public void testCreateAndFind() throws Exception {
 
-		JobSupport job = new JobSupport("testJob");
 		job.setRestartable(true);
 
 		Map stringParams = new HashMap() {
@@ -78,15 +87,12 @@ public class SimpleJobRepositoryIntegrationTests extends AbstractTransactionalDa
 	 * executions belong to the same job instance and job.
 	 */
 	public void testCreateAndFindWithNoStartDate() throws Exception {
-
-		JobSupport job = new JobSupport("testJob");
 		job.setRestartable(true);
-		JobParameters jobParams = new JobParameters();
 
-		JobExecution firstExecution = jobRepository.createJobExecution(job, jobParams);
+		JobExecution firstExecution = jobRepository.createJobExecution(job, jobParameters);
 		firstExecution.setEndTime(new Date());
 		jobRepository.saveOrUpdate(firstExecution);
-		JobExecution secondExecution = jobRepository.createJobExecution(job, jobParams);
+		JobExecution secondExecution = jobRepository.createJobExecution(job, jobParameters);
 
 		assertEquals(firstExecution.getJobInstance(), secondExecution.getJobInstance());
 		assertEquals(job, secondExecution.getJobInstance().getJob());
@@ -97,13 +103,11 @@ public class SimpleJobRepositoryIntegrationTests extends AbstractTransactionalDa
 	 * existing non-restartable JobInstance causes error.
 	 */
 	public void testRunNonRestartableJobInstanceTwice() throws Exception {
-		JobSupport job = new JobSupport("nonRestartableJob");
 		job.setRestartable(false);
-		JobParameters jobParameters = new JobParameters();
-		
+
 		JobExecution firstExecution = jobRepository.createJobExecution(job, jobParameters);
 		jobRepository.saveOrUpdate(firstExecution);
-		
+
 		try {
 			jobRepository.createJobExecution(job, jobParameters);
 			fail();
@@ -112,4 +116,81 @@ public class SimpleJobRepositoryIntegrationTests extends AbstractTransactionalDa
 			// expected
 		}
 	}
+
+	/**
+	 * Save multiple StepExecutions for the same step and check the returned
+	 * count and last execution are correct.
+	 */
+	public void testGetStepExecutionCountAndLastStepExecution() throws Exception {
+		job.setRestartable(true);
+		StepSupport step = new StepSupport("restartedStep");
+
+		// first execution
+		JobExecution firstJobExec = jobRepository.createJobExecution(job, jobParameters);
+		StepExecution firstStepExec = new StepExecution(step, firstJobExec);
+		jobRepository.saveOrUpdate(firstJobExec);
+		jobRepository.saveOrUpdate(firstStepExec);
+
+		assertEquals(1, jobRepository.getStepExecutionCount(firstJobExec.getJobInstance(), step));
+		assertEquals(firstStepExec, jobRepository.getLastStepExecution(firstJobExec.getJobInstance(), step));
+
+		// first execution failed
+		firstJobExec.setStartTime(new Date(4));
+		firstStepExec.setStartTime(new Date(5));
+		firstStepExec.setStatus(BatchStatus.FAILED);
+		firstStepExec.setEndTime(new Date(6));
+		jobRepository.saveOrUpdate(firstStepExec);
+		firstJobExec.setStatus(BatchStatus.FAILED);
+		firstJobExec.setEndTime(new Date(7));
+		jobRepository.saveOrUpdate(firstJobExec);
+
+		// second execution
+		JobExecution secondJobExec = jobRepository.createJobExecution(job, jobParameters);
+		StepExecution secondStepExec = new StepExecution(step, secondJobExec);
+		jobRepository.saveOrUpdate(secondJobExec);
+		jobRepository.saveOrUpdate(secondStepExec);
+
+		assertEquals(2, jobRepository.getStepExecutionCount(secondJobExec.getJobInstance(), step));
+		assertEquals(secondStepExec, jobRepository.getLastStepExecution(secondJobExec.getJobInstance(), step));
+	}
+
+	/**
+	 * Save execution context and retrieve it.
+	 */
+	public void testSaveExecutionContext() throws Exception {
+		ExecutionContext ctx = new ExecutionContext() {
+			{
+				putLong("crashedPosition", 7);
+			}
+		};
+		JobExecution jobExec = jobRepository.createJobExecution(job, jobParameters);
+		Step step = new StepSupport("step1");
+		StepExecution stepExec = new StepExecution(step, jobExec);
+		stepExec.setExecutionContext(ctx);
+
+		jobRepository.saveOrUpdate(stepExec);
+		jobRepository.saveOrUpdateExecutionContext(stepExec);
+
+		StepExecution retrievedExec = jobRepository.getLastStepExecution(jobExec.getJobInstance(), step);
+		assertEquals(stepExec, retrievedExec);
+		assertEquals(ctx, retrievedExec.getExecutionContext());
+	}
+
+	/**
+	 * If JobExecution is already running, exception will be thrown in attempt
+	 * to create new execution.
+	 */
+	public void testOnlyOneJobExecutionAllowedRunning() throws Exception {
+		job.setRestartable(true);
+		jobRepository.createJobExecution(job, jobParameters);
+		
+		try {
+			jobRepository.createJobExecution(job, jobParameters);
+			fail();
+		}
+		catch (JobExecutionAlreadyRunningException e) {
+			// expected
+		}
+	}
+	
 }
\ No newline at end of file