RESOLVED - BATCH-707: interrupt support in SystemCommandTasklet

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.
 	 */

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;
 			}
 		}