diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/CommandLineJobOperator.java b/spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/CommandLineJobOperator.java
index e85fd745d3..57925165e2 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/CommandLineJobOperator.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/CommandLineJobOperator.java
@@ -22,6 +22,7 @@
 import org.springframework.batch.core.configuration.JobRegistry;
 import org.springframework.batch.core.converter.DefaultJobParametersConverter;
 import org.springframework.batch.core.converter.JobParametersConverter;
+import org.springframework.batch.core.BatchStatus;
 import org.springframework.batch.core.job.Job;
 import org.springframework.batch.core.job.JobExecution;
 import org.springframework.batch.core.job.parameters.JobParameters;
@@ -52,6 +53,7 @@
  *
  * @author Mahmoud Ben Hassine
  * @author Yejeong Ham
+ * @author Cheolhwan Ihn
  * @since 6.0
  */
 public class CommandLineJobOperator {
@@ -185,7 +187,12 @@ public int restart(long jobExecutionId) {
 				logger.error(() -> "No job execution found with ID: " + jobExecutionId);
 				return JVM_EXITCODE_GENERIC_ERROR;
 			}
-			// TODO should check and log error if the job execution did not fail
+			BatchStatus status = jobExecution.getStatus();
+			if (status != BatchStatus.FAILED && status != BatchStatus.STOPPED) {
+				logger.error(() -> "Cannot restart job execution " + jobExecutionId + ": current status is " + status
+						+ " (must be FAILED or STOPPED).");
+				return JVM_EXITCODE_GENERIC_ERROR;
+			}
 			JobExecution restartedExecution = this.jobOperator.restart(jobExecution);
 			return this.exitCodeMapper.intValue(restartedExecution.getExitStatus().getExitCode());
 		}
@@ -208,8 +215,12 @@ public int abandon(long jobExecutionId) {
 				logger.error(() -> "No job execution found with ID: " + jobExecutionId);
 				return JVM_EXITCODE_GENERIC_ERROR;
 			}
-			// TODO should throw JobExecutionNotStoppedException if the job execution is
-			// not stopped
+			BatchStatus status = jobExecution.getStatus();
+			if (status != BatchStatus.STOPPED) {
+				logger.error(() -> "Cannot abandon job execution " + jobExecutionId + ": current status is " + status
+						+ " (must be STOPPED).");
+				return JVM_EXITCODE_GENERIC_ERROR;
+			}
 			JobExecution abandonedExecution = this.jobOperator.abandon(jobExecution);
 			return this.exitCodeMapper.intValue(abandonedExecution.getExitStatus().getExitCode());
 		}
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/launch/support/CommandLineJobOperatorTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/launch/support/CommandLineJobOperatorTests.java
index 31a4610e3a..07b2491931 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/launch/support/CommandLineJobOperatorTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/launch/support/CommandLineJobOperatorTests.java
@@ -17,10 +17,12 @@
 
 import java.util.Properties;
 
+import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
 
+import org.springframework.batch.core.BatchStatus;
 import org.springframework.batch.core.configuration.JobRegistry;
 import org.springframework.batch.core.converter.JobParametersConverter;
 import org.springframework.batch.core.job.Job;
@@ -30,12 +32,14 @@
 import org.springframework.batch.core.repository.JobRepository;
 
 import static org.mockito.Mockito.mock;
+import static org.springframework.batch.core.launch.support.ExitCodeMapper.JVM_EXITCODE_GENERIC_ERROR;
 
 /**
  * Tests for {@link CommandLineJobOperator}.
  *
  * @author Mahmoud Ben Hassine
  * @author Yejeong Ham
+ * @author Cheolhwan Ihn
  */
 class CommandLineJobOperatorTests {
 
@@ -111,6 +115,7 @@ void restart() throws Exception {
 		// given
 		long jobExecutionId = 1;
 		JobExecution jobExecution = mock();
+		Mockito.when(jobExecution.getStatus()).thenReturn(BatchStatus.FAILED);
 
 		// when
 		Mockito.when(jobRepository.getJobExecution(jobExecutionId)).thenReturn(jobExecution);
@@ -121,19 +126,67 @@ void restart() throws Exception {
 	}
 
 	@Test
-	void abandon() throws Exception {
+	void restartJobExecutionStopped() throws Exception {
 		// given
 		long jobExecutionId = 1;
 		JobExecution jobExecution = mock();
+		Mockito.when(jobExecution.getStatus()).thenReturn(BatchStatus.STOPPED);
+		Mockito.when(jobRepository.getJobExecution(jobExecutionId)).thenReturn(jobExecution);
 
 		// when
+		this.commandLineJobOperator.restart(jobExecutionId);
+
+		// then
+		Mockito.verify(jobOperator).restart(jobExecution);
+	}
+
+	@Test
+	void restartJobExecutionNotFailed() throws Exception {
+		// given
+		long jobExecutionId = 1;
+		JobExecution jobExecution = mock();
+		Mockito.when(jobExecution.getStatus()).thenReturn(BatchStatus.COMPLETED);
 		Mockito.when(jobRepository.getJobExecution(jobExecutionId)).thenReturn(jobExecution);
+
+		// when
+		int exitCode = this.commandLineJobOperator.restart(jobExecutionId);
+
+		// then
+		Assertions.assertEquals(JVM_EXITCODE_GENERIC_ERROR, exitCode);
+		Mockito.verify(jobOperator, Mockito.never()).restart(jobExecution);
+	}
+
+	@Test
+	void abandon() throws Exception {
+		// given
+		long jobExecutionId = 1;
+		JobExecution jobExecution = mock();
+		Mockito.when(jobExecution.getStatus()).thenReturn(BatchStatus.STOPPED);
+		Mockito.when(jobRepository.getJobExecution(jobExecutionId)).thenReturn(jobExecution);
+
+		// when
 		this.commandLineJobOperator.abandon(jobExecutionId);
 
 		// then
 		Mockito.verify(jobOperator).abandon(jobExecution);
 	}
 
+	@Test
+	void abandonJobExecutionNotStopped() throws Exception {
+		// given
+		long jobExecutionId = 1;
+		JobExecution jobExecution = mock();
+		Mockito.when(jobExecution.getStatus()).thenReturn(BatchStatus.COMPLETED);
+		Mockito.when(jobRepository.getJobExecution(jobExecutionId)).thenReturn(jobExecution);
+
+		// when
+		int exitCode = this.commandLineJobOperator.abandon(jobExecutionId);
+
+		// then
+		Assertions.assertEquals(ExitCodeMapper.JVM_EXITCODE_GENERIC_ERROR, exitCode); // JVM_EXITCODE_GENERIC_ERROR
+		Mockito.verify(jobOperator, Mockito.never()).abandon(jobExecution);
+	}
+
 	@Test
 	void recover() {
 		// given
@@ -148,4 +201,4 @@ void recover() {
 		Mockito.verify(jobOperator).recover(jobExecution);
 	}
 
-}
\ No newline at end of file
+}