diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/tasklet/SystemCommandTasklet.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/tasklet/SystemCommandTasklet.java index 816e19ff6..7a34576d9 100644 --- a/spring-batch-samples/src/main/java/org/springframework/batch/sample/tasklet/SystemCommandTasklet.java +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/tasklet/SystemCommandTasklet.java @@ -3,6 +3,10 @@ package org.springframework.batch.sample.tasklet; import java.io.File; import java.io.IOException; +import org.apache.commons.lang.time.StopWatch; +import org.springframework.batch.core.JobInterruptedException; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.listener.StepExecutionListenerSupport; import org.springframework.batch.core.step.tasklet.Tasklet; import org.springframework.batch.repeat.ExitStatus; import org.springframework.beans.factory.InitializingBean; @@ -15,9 +19,18 @@ import org.springframework.util.Assert; * be set, so that the batch job does not hang forever if the external process * hangs. * + * Tasklet periodically checks for termination status (i.e. + * {@link #setCommand(String)} finished its execution or + * {@link #setTimeout(long)} expired or job was interrupted). The check interval + * is given by {@link #setTerminationCheckInterval(long)}. + * + * When job interrupt is detected the thread executing the system command will + * be interrupted and tasklet's execution terminated by throwing + * {@link JobInterruptedException}. + * * @author Robert Kasanicky */ -public class SystemCommandTasklet implements Tasklet, InitializingBean { +public class SystemCommandTasklet extends StepExecutionListenerSupport implements Tasklet, InitializingBean { private String command; @@ -29,6 +42,10 @@ public class SystemCommandTasklet implements Tasklet, InitializingBean { private long timeout = 0; + private long checkInterval = 1000; + + private StepExecution execution = null; + /** * Execute system command and map its exit code to {@link ExitStatus} using * {@link SystemProcessExitCodeMapper}. @@ -36,15 +53,28 @@ public class SystemCommandTasklet implements Tasklet, InitializingBean { public ExitStatus execute() throws Exception { ExecutorThread executorThread = new ExecutorThread(); executorThread.start(); - executorThread.join(timeout); + + StopWatch stopWatch = new StopWatch(); + stopWatch.start(); + + while (stopWatch.getTime() < timeout && executorThread.isAlive() && !execution.isTerminateOnly()) { + Thread.sleep(checkInterval); + } + + stopWatch.stop(); if (executorThread.finishedSuccessfully) { return systemProcessExitCodeMapper.getExitStatus(executorThread.exitCode); } else { executorThread.interrupt(); - throw new SystemCommandException( - "Execution of system command failed (did not finish successfully within the timeout)"); + if (execution.isTerminateOnly()) { + throw new JobInterruptedException("Job interrupted while executing system command '" + command + "'"); + } + else { + throw new SystemCommandException( + "Execution of system command failed (did not finish successfully within the timeout)"); + } } } @@ -102,6 +132,24 @@ public class SystemCommandTasklet implements Tasklet, InitializingBean { this.timeout = timeout; } + /** + * The time interval how often the tasklet will check for termination + * status. + * + * @param checkInterval time interval in milliseconds (1 second by default). + */ + public void setTerminationCheckInterval(long checkInterval) { + this.checkInterval = checkInterval; + } + + /** + * Get a reference to {@link StepExecution} for interrupt checks during + * system command execution. + */ + public void beforeStep(StepExecution stepExecution) { + this.execution = stepExecution; + } + /** * Thread that executes the system command. */ diff --git a/spring-batch-samples/src/test/java/org/springframework/batch/sample/tasklet/SystemCommandTaskletIntegrationTests.java b/spring-batch-samples/src/test/java/org/springframework/batch/sample/tasklet/SystemCommandTaskletIntegrationTests.java index 445e73180..7b2eb5517 100644 --- a/spring-batch-samples/src/test/java/org/springframework/batch/sample/tasklet/SystemCommandTaskletIntegrationTests.java +++ b/spring-batch-samples/src/test/java/org/springframework/batch/sample/tasklet/SystemCommandTaskletIntegrationTests.java @@ -6,6 +6,11 @@ import junit.framework.TestCase; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobInstance; +import org.springframework.batch.core.JobInterruptedException; +import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.StepExecution; import org.springframework.batch.repeat.ExitStatus; import org.springframework.batch.sample.tasklet.SystemCommandException; import org.springframework.batch.sample.tasklet.SystemCommandTasklet; @@ -21,11 +26,19 @@ public class SystemCommandTaskletIntegrationTests extends TestCase { private SystemCommandTasklet tasklet = new SystemCommandTasklet(); + private StepExecution stepExecution = new StepExecution("systemCommandStep", new JobExecution(new JobInstance( + new Long(1), new JobParameters(), "systemCommandJob"))); + protected void setUp() throws Exception { tasklet.setEnvironmentParams(null); // inherit from parent process tasklet.setWorkingDirectory(null); // inherit from parent process tasklet.setSystemProcessExitCodeMapper(new TestExitCodeMapper()); tasklet.setTimeout(5000); // long enough timeout + tasklet.setTerminationCheckInterval(500); + tasklet.setCommand("invalid command, change value for successful execution"); + tasklet.afterPropertiesSet(); + + tasklet.beforeStep(stepExecution); } /** @@ -75,6 +88,27 @@ public class SystemCommandTaskletIntegrationTests extends TestCase { } } + /** + * Job interrupted scenario. + */ + public void testInterruption() throws Exception { + String command = "sleep 5"; + tasklet.setCommand(command); + tasklet.setTerminationCheckInterval(10); + tasklet.afterPropertiesSet(); + + stepExecution.setTerminateOnly(); + try { + tasklet.execute(); + fail(); + } + catch (JobInterruptedException e) { + System.out.println(e.getMessage()); + assertTrue(e.getMessage().indexOf("Job interrupted while executing system command") > -1); + assertTrue(e.getMessage().indexOf(command) > -1); + } + } + /** * Command property value is required to be set. */ @@ -158,7 +192,8 @@ public class SystemCommandTaskletIntegrationTests extends TestCase { public ExitStatus getExitStatus(int exitCode) { if (exitCode == 0) { return ExitStatus.FINISHED; - } else { + } + else { return ExitStatus.FAILED; } }