Added ability for SystemCommandTasklet to be stopped

Prior to this commit, the SystemCommandTasklet is able to be stopped
only by an external process calling directly into it or updating the
instance of the StepExecution it holds onto.  This update allows the
tasklet to poll the job repository for stop requests in a similar way
chunk based processing does.

BATCH-2331

spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/SystemCommandTasklet.java

@@ -16,12 +16,19 @@
 
 package org.springframework.batch.core.step.tasklet;
 
+import java.io.File;
+import java.util.concurrent.Callable;
+import java.util.concurrent.FutureTask;
+
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+
 import org.springframework.batch.core.ExitStatus;
+import org.springframework.batch.core.JobExecution;
 import org.springframework.batch.core.JobInterruptedException;
 import org.springframework.batch.core.StepContribution;
 import org.springframework.batch.core.StepExecution;
+import org.springframework.batch.core.explore.JobExplorer;
 import org.springframework.batch.core.listener.StepExecutionListenerSupport;
 import org.springframework.batch.core.scope.context.ChunkContext;
 import org.springframework.batch.repeat.RepeatStatus;
@@ -30,10 +37,6 @@ import org.springframework.core.task.SimpleAsyncTaskExecutor;
 import org.springframework.core.task.TaskExecutor;
 import org.springframework.util.Assert;
 
-import java.io.File;
-import java.util.concurrent.Callable;
-import java.util.concurrent.FutureTask;
-
 /**
  * {@link Tasklet} that executes a system command.
  *
@@ -80,6 +83,10 @@ public class SystemCommandTasklet extends StepExecutionListenerSupport implement
 
 	private volatile boolean stopped = false;
 
+	private JobExplorer jobExplorer;
+
+	private boolean stoppable = false;
+
 	/**
 	 * Execute system command and map its exit code to {@link ExitStatus} using
 	 * {@link SystemProcessExitCodeMapper}.
@@ -103,6 +110,16 @@ public class SystemCommandTasklet extends StepExecutionListenerSupport implement
 
 		while (true) {
 			Thread.sleep(checkInterval);//moved to the end of the logic
+
+			if(stoppable) {
+				JobExecution jobExecution =
+						jobExplorer.getJobExecution(chunkContext.getStepContext().getStepExecution().getJobExecutionId());
+
+				if(jobExecution.isStopping()) {
+					stopped = true;
+				}
+			}
+
 			if (systemCommandTask.isDone()) {
 				contribution.setExitStatus(systemProcessExitCodeMapper.getExitStatus(systemCommandTask.get()));
 				return RepeatStatus.FINISHED;
@@ -159,6 +176,11 @@ public class SystemCommandTasklet extends StepExecutionListenerSupport implement
 		Assert.notNull(systemProcessExitCodeMapper, "SystemProcessExitCodeMapper must be set");
 		Assert.isTrue(timeout > 0, "timeout value must be greater than zero");
 		Assert.notNull(taskExecutor, "taskExecutor is required");
+		stoppable = jobExplorer != null;
+	}
+
+	public void setJobExplorer(JobExplorer jobExplorer) {
+		this.jobExplorer = jobExplorer;
 	}
 
 	/**

spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/SystemCommandTaskletIntegrationTests.java

@@ -18,6 +18,7 @@ package org.springframework.batch.core.step.tasklet;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
+import static org.mockito.Mockito.when;
 
 import java.io.File;
 
@@ -25,6 +26,10 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.junit.Before;
 import org.junit.Test;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+
+import org.springframework.batch.core.BatchStatus;
 import org.springframework.batch.core.ExitStatus;
 import org.springframework.batch.core.JobExecution;
 import org.springframework.batch.core.JobInstance;
@@ -32,6 +37,9 @@ import org.springframework.batch.core.JobInterruptedException;
 import org.springframework.batch.core.JobParameters;
 import org.springframework.batch.core.StepContribution;
 import org.springframework.batch.core.StepExecution;
+import org.springframework.batch.core.explore.JobExplorer;
+import org.springframework.batch.core.scope.context.ChunkContext;
+import org.springframework.batch.core.scope.context.StepContext;
 import org.springframework.batch.repeat.RepeatStatus;
 import org.springframework.core.task.SimpleAsyncTaskExecutor;
 import org.springframework.util.Assert;
@@ -43,13 +51,26 @@ public class SystemCommandTaskletIntegrationTests {
 
 	private static final Log log = LogFactory.getLog(SystemCommandTaskletIntegrationTests.class);
 
-	private SystemCommandTasklet tasklet = new SystemCommandTasklet();
+	private SystemCommandTasklet tasklet;
 
 	private StepExecution stepExecution = new StepExecution("systemCommandStep", new JobExecution(new JobInstance(1L,
-			"systemCommandJob"), new JobParameters()));
+			"systemCommandJob"), 1L, new JobParameters(), "configurationName"));
+
+	@Mock
+	private JobExplorer jobExplorer;
 
 	@Before
 	public void setUp() throws Exception {
+		MockitoAnnotations.initMocks(this);
+
+		initializeTasklet();
+		tasklet.afterPropertiesSet();
+
+		tasklet.beforeStep(stepExecution);
+	}
+
+	private void initializeTasklet() {
+		tasklet = new SystemCommandTasklet();
 		tasklet.setEnvironmentParams(null); // inherit from parent process
 		tasklet.setWorkingDirectory(null); // inherit from parent process
 		tasklet.setSystemProcessExitCodeMapper(new TestExitCodeMapper());
@@ -58,9 +79,6 @@ public class SystemCommandTaskletIntegrationTests {
 		tasklet.setCommand("invalid command, change value for successful execution");
 		tasklet.setInterruptOnCancel(true);
 		tasklet.setTaskExecutor(new SimpleAsyncTaskExecutor());
-		tasklet.afterPropertiesSet();
-
-		tasklet.beforeStep(stepExecution);
 	}
 
 	/*
@@ -241,17 +259,27 @@ public class SystemCommandTaskletIntegrationTests {
 	 */
 	@Test
 	public void testStopped() throws Exception {
+		initializeTasklet();
+		tasklet.setJobExplorer(jobExplorer);
+		tasklet.afterPropertiesSet();
+		tasklet.beforeStep(stepExecution);
+
+		JobExecution stoppedJobExecution = new JobExecution(stepExecution.getJobExecution());
+		stoppedJobExecution.setStatus(BatchStatus.STOPPING);
+
+		when(jobExplorer.getJobExecution(1L)).thenReturn(stepExecution.getJobExecution(), stepExecution.getJobExecution(), stoppedJobExecution);
+
 		String command = System.getProperty("os.name").toLowerCase().indexOf("win") >= 0 ?
 				"ping 1.1.1.1 -n 1 -w 5000" :
-					"sleep 5";
+					"sleep 15";
 		tasklet.setCommand(command);
 		tasklet.setTerminationCheckInterval(10);
 		tasklet.afterPropertiesSet();
 
 		StepContribution contribution = stepExecution.createStepContribution();
-		//send stop
-		tasklet.stop();
-		tasklet.execute(contribution, null);
+		StepContext stepContext = new StepContext(stepExecution);
+		ChunkContext chunkContext = new ChunkContext(stepContext);
+		tasklet.execute(contribution, chunkContext);
 
 		assertEquals(contribution.getExitStatus().getExitCode(),ExitStatus.STOPPED.getExitCode());
 	}