diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/SimpleJobLauncher.java b/spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/SimpleJobLauncher.java
index 68e3d2df3..41dd76ac0 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/SimpleJobLauncher.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/SimpleJobLauncher.java
@@ -24,6 +24,7 @@ import org.springframework.batch.core.JobExecution;
 import org.springframework.batch.core.JobInstance;
 import org.springframework.batch.core.JobParameters;
 import org.springframework.batch.core.JobParametersInvalidException;
+import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.launch.JobLauncher;
 import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException;
 import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException;
@@ -53,6 +54,7 @@ import org.springframework.util.Assert;
  *
  * @author Lucas Ward
  * @Author Dave Syer
+ * @author Will Schipp
  *
  * @since 1.0
  *
@@ -97,12 +99,22 @@ public class SimpleJobLauncher implements JobLauncher, InitializingBean {
 			if (!job.isRestartable()) {
 				throw new JobRestartException("JobInstance already exists and is not restartable");
 			}
+			/*
+			 * validate here if it has stepExecutions that are UNKNOWN
+			 * retrieve the previous execution and check
+			 */
+			for (StepExecution execution : lastExecution.getStepExecutions()) {
+				if (execution.getStatus() == BatchStatus.UNKNOWN) {
+					//throw
+					throw new JobRestartException("Step [" + execution.getStepName() + "] is of status UNKNOWN"); 
+				}//end if
+			}//end for			
 		}
 
 		// Check the validity of the parameters before doing creating anything
 		// in the repository...
 		job.getJobParametersValidator().validate(jobParameters);
-
+		
 		/*
 		 * There is a very small probability that a non-restartable job can be
 		 * restarted, but only if another process or thread manages to launch
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/launch/SimpleJobLauncherTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/launch/SimpleJobLauncherTests.java
index f7e768bf2..0d3dd9d6f 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/launch/SimpleJobLauncherTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/launch/SimpleJobLauncherTests.java
@@ -16,26 +16,33 @@
 
 package org.springframework.batch.core.launch;
 
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 
 import org.junit.Before;
 import org.junit.Test;
 import org.springframework.batch.core.BatchStatus;
 import org.springframework.batch.core.ExitStatus;
+import org.springframework.batch.core.Job;
 import org.springframework.batch.core.JobExecution;
 import org.springframework.batch.core.JobInstance;
 import org.springframework.batch.core.JobParameters;
+import org.springframework.batch.core.JobParametersBuilder;
 import org.springframework.batch.core.JobParametersInvalidException;
+import org.springframework.batch.core.JobParametersValidator;
+import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.job.DefaultJobParametersValidator;
 import org.springframework.batch.core.job.JobSupport;
 import org.springframework.batch.core.launch.support.SimpleJobLauncher;
+import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException;
+import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException;
 import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.core.repository.JobRestartException;
 import org.springframework.core.task.TaskExecutor;
@@ -262,4 +269,36 @@ public class SimpleJobLauncherTests {
 	private boolean contains(String str, String searchStr) {
 		return str.indexOf(searchStr) != -1;
 	}
+	
+	/**
+	 * Test to support BATCH-1770 -> throw in parent thread JobRestartException when 
+	 * a stepExecution is UNKNOWN
+	 */
+	@Test(expected=JobRestartException.class)
+	public void testRunStepStatusUnknown() throws Exception {
+		//try and restart a job where the step execution is UNKNOWN 
+		//setup
+		String jobName = "test_job";
+		JobRepository jobRepository = mock(JobRepository.class);
+		JobParameters parameters = new JobParametersBuilder().addLong("runtime", System.currentTimeMillis()).toJobParameters();
+		JobExecution jobExecution = mock(JobExecution.class);
+		Job job = mock(Job.class);
+		JobParametersValidator validator = mock(JobParametersValidator.class);
+		StepExecution stepExecution = mock(StepExecution.class);
+		
+		when(job.getName()).thenReturn(jobName);
+		when(job.isRestartable()).thenReturn(true);		
+		when(job.getJobParametersValidator()).thenReturn(validator);
+		when(jobRepository.getLastJobExecution(jobName, parameters)).thenReturn(jobExecution);
+		when(stepExecution.getStatus()).thenReturn(BatchStatus.UNKNOWN);
+		when(jobExecution.getStepExecutions()).thenReturn(Arrays.asList(stepExecution));
+		
+		//setup launcher
+		jobLauncher = new SimpleJobLauncher();
+		jobLauncher.setJobRepository(jobRepository);
+		
+		//run
+		jobLauncher.run(job, parameters);
+
+	}	
 }