diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/SimpleJobOperator.java b/spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/SimpleJobOperator.java
index a006243497..f535646819 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/SimpleJobOperator.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/SimpleJobOperator.java
@@ -245,10 +245,13 @@ public JobExecution restart(JobExecution jobExecution) throws JobRestartExceptio
 		try {
 			return run(job, parameters);
 		}
-		catch (Exception e) {
+		catch (JobExecutionAlreadyRunningException e) {
 			throw new JobRestartException(
 					String.format(ILLEGAL_STATE_MSG, "job execution already running", jobName, parameters), e);
 		}
+		catch (JobInstanceAlreadyCompleteException | InvalidJobParametersException e) {
+			throw new JobRestartException(e.getMessage(), e);
+		}
 
 	}
 
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/launch/support/TaskExecutorJobOperatorTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/launch/support/TaskExecutorJobOperatorTests.java
index 38f960d301..1a7b5e1d13 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/launch/support/TaskExecutorJobOperatorTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/launch/support/TaskExecutorJobOperatorTests.java
@@ -15,6 +15,8 @@
  */
 package org.springframework.batch.core.launch.support;
 
+import java.time.LocalDateTime;
+
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -127,4 +129,24 @@ public RepeatStatus execute(StepContribution contribution, ChunkContext chunkCon
 		Assertions.assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus());
 	}
 
+	@Test
+	void testRestartFromUnknownStatusSurfacesRealReason() throws Exception {
+		JobParameters jobParameters = new JobParameters();
+		JobExecution jobExecution = jobOperator.start(job, jobParameters);
+
+		Assertions.assertNotNull(jobExecution);
+		Assertions.assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus());
+
+		// Simulate an execution that ended in UNKNOWN status.
+		jobExecution.setStatus(BatchStatus.UNKNOWN);
+		jobExecution.setEndTime(LocalDateTime.now());
+		jobRepository.update(jobExecution);
+
+		JobRestartException exception = Assertions.assertThrows(JobRestartException.class,
+				() -> jobOperator.restart(jobExecution));
+
+		Assertions.assertTrue(exception.getMessage().contains("Cannot restart job from UNKNOWN status"),
+				"Expected the UNKNOWN status reason to be surfaced but was: " + exception.getMessage());
+	}
+
 }