From fecbe075d462f0b0e1fadb151d9c11b0b5a2a348 Mon Sep 17 00:00:00 2001 From: Michael Minella Date: Wed, 31 Dec 2014 11:45:13 -0600 Subject: [PATCH] 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 --- .../step/tasklet/SystemCommandTasklet.java | 30 ++++++++++-- .../SystemCommandTaskletIntegrationTests.java | 46 +++++++++++++++---- 2 files changed, 63 insertions(+), 13 deletions(-) diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/SystemCommandTasklet.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/SystemCommandTasklet.java index 1fd71b975..6dc013fba 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/SystemCommandTasklet.java +++ b/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; } /** diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/SystemCommandTaskletIntegrationTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/SystemCommandTaskletIntegrationTests.java index 607d315b1..e73a1ab33 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/SystemCommandTaskletIntegrationTests.java +++ b/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()); }