From 81e0c38d52abde655ceea8481122f6fc4cc0192e Mon Sep 17 00:00:00 2001 From: willschipp Date: Wed, 22 May 2013 07:34:57 -0400 Subject: [PATCH] BATCH-2009 - Stoppable tasklet * Added the StoppableTasklet interface (an extension of the Tasklet interface). * Updated JobOperator#stop(long executionId) to call StoppableTasklet#stop() on any currently running tasklets (local executions only). * Updated the SystemCommandTasklet to implement StoppableTasklet --- .../launch/support/SimpleJobOperator.java | 34 +++++++ .../core/step/tasklet/StoppableTasklet.java | 38 ++++++++ .../step/tasklet/SystemCommandTasklet.java | 27 +++++- .../batch/core/step/tasklet/TaskletStep.java | 9 ++ .../support/SimpleJobOperatorTests.java | 94 +++++++++++++++++++ .../SystemCommandTaskletIntegrationTests.java | 20 ++++ 6 files changed, 219 insertions(+), 3 deletions(-) create mode 100644 spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/StoppableTasklet.java diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/SimpleJobOperator.java b/spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/SimpleJobOperator.java index 820526fd0..f0ea157c9 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/SimpleJobOperator.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/SimpleJobOperator.java @@ -33,6 +33,7 @@ import org.springframework.batch.core.JobInstance; import org.springframework.batch.core.JobParameters; import org.springframework.batch.core.JobParametersIncrementer; import org.springframework.batch.core.JobParametersInvalidException; +import org.springframework.batch.core.Step; import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.UnexpectedJobExecutionException; import org.springframework.batch.core.configuration.JobRegistry; @@ -52,6 +53,11 @@ import org.springframework.batch.core.repository.JobExecutionAlreadyRunningExcep import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException; import org.springframework.batch.core.repository.JobRepository; import org.springframework.batch.core.repository.JobRestartException; +import org.springframework.batch.core.step.NoSuchStepException; +import org.springframework.batch.core.step.StepLocator; +import org.springframework.batch.core.step.tasklet.StoppableTasklet; +import org.springframework.batch.core.step.tasklet.Tasklet; +import org.springframework.batch.core.step.tasklet.TaskletStep; import org.springframework.batch.support.PropertiesConverter; import org.springframework.beans.factory.InitializingBean; import org.springframework.transaction.annotation.Transactional; @@ -71,6 +77,7 @@ import org.springframework.util.Assert; * * @author Dave Syer * @author Lucas Ward + * @author Will Schipp * @since 2.0 */ public class SimpleJobOperator implements JobOperator, InitializingBean { @@ -394,6 +401,33 @@ public class SimpleJobOperator implements JobOperator, InitializingBean { jobExecution.setStatus(BatchStatus.STOPPING); jobRepository.update(jobExecution); + try { + Job job = jobRegistry.getJob(jobExecution.getJobInstance().getJobName()); + if (job instanceof StepLocator) {//can only process as StepLocator is the only way to get the step object + //get the current stepExecution + for (StepExecution stepExecution : jobExecution.getStepExecutions()) { + if (stepExecution.getStatus().isRunning()) { + try { + //have the step execution that's running -> need to 'stop' it + Step step = ((StepLocator)job).getStep(stepExecution.getStepName()); + if (step instanceof TaskletStep) { + Tasklet tasklet = ((TaskletStep)step).getTasklet(); + if (tasklet instanceof StoppableTasklet) { + ((StoppableTasklet)tasklet).stop(); + } + } + } + catch (NoSuchStepException e) { + logger.warn("Step not found",e); + } + } + } + } + } + catch (NoSuchJobException e) { + logger.warn("Cannot find Job object",e); + } + return true; } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/StoppableTasklet.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/StoppableTasklet.java new file mode 100644 index 000000000..21f357a93 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/StoppableTasklet.java @@ -0,0 +1,38 @@ +/* + * Copyright 2013 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.batch.core.step.tasklet; + +import org.springframework.batch.core.launch.JobOperator; + +/** + * An extension to the {@link Tasklet} interface to allow users to + * add logic for stopping a tasklet. It is up to each implementation + * as to how the stop will behave. The only guarantee provided by the + * framework is that a call to {@link JobOperator#stop(long)} will + * attempt to call the stop method on any currently running + * StoppableTasklet. + * + * @author Will Schipp + * @since 3.0 + */ +public interface StoppableTasklet extends Tasklet { + + /** + * Used to signal that the job this {@link Tasklet} is executing + * within has been requested to stop. + */ + void stop(); +} 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 397ef7578..ba437cc47 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 @@ -54,8 +54,9 @@ import org.springframework.util.Assert; * still running when tasklet exits (abnormally). * * @author Robert Kasanicky + * @author Will Schipp */ -public class SystemCommandTasklet extends StepExecutionListenerSupport implements Tasklet, InitializingBean { +public class SystemCommandTasklet extends StepExecutionListenerSupport implements StoppableTasklet, InitializingBean { protected static final Log logger = LogFactory.getLog(SystemCommandTasklet.class); @@ -77,6 +78,8 @@ public class SystemCommandTasklet extends StepExecutionListenerSupport implement private boolean interruptOnCancel = false; + private boolean stopped = false; + /** * Execute system command and map its exit code to {@link ExitStatus} using * {@link SystemProcessExitCodeMapper}. @@ -99,7 +102,7 @@ public class SystemCommandTasklet extends StepExecutionListenerSupport implement taskExecutor.execute(systemCommandTask); while (true) { - Thread.sleep(checkInterval); + Thread.sleep(checkInterval);//moved to the end of the logic if (systemCommandTask.isDone()) { contribution.setExitStatus(systemProcessExitCodeMapper.getExitStatus(systemCommandTask.get())); return RepeatStatus.FINISHED; @@ -112,8 +115,12 @@ public class SystemCommandTasklet extends StepExecutionListenerSupport implement systemCommandTask.cancel(interruptOnCancel); throw new JobInterruptedException("Job interrupted while executing system command '" + command + "'"); } + else if (stopped) { + systemCommandTask.cancel(interruptOnCancel); + contribution.setExitStatus(ExitStatus.STOPPED); + return RepeatStatus.FINISHED; + } } - } /** @@ -208,4 +215,18 @@ public class SystemCommandTasklet extends StepExecutionListenerSupport implement this.interruptOnCancel = interruptOnCancel; } + /** + * Will interrupt the thread executing the system command only if + * {@link #setInterruptOnCancel(boolean)} has been set to true. Otherwise + * the underlying command will be allowed to finish before the tasklet + * ends. + * + * @since 3.0 + * @see StoppableTasklet#stop() + */ + @Override + public void stop() { + stopped = true; + } + } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java index 8b98f9fe4..7cca3e0af 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java @@ -71,6 +71,7 @@ import org.springframework.util.Assert; * @author Ben Hale * @author Robert Kasanicky * @author Michael Minella + * @author Will Schipp */ @SuppressWarnings("serial") public class TaskletStep extends AbstractStep { @@ -306,6 +307,14 @@ public class TaskletStep extends AbstractStep { stream.open(ctx); } + /** + * retrieve the tasklet - helper method for JobOperator + * @return the {@link Tasklet} instance being executed within this step + */ + public Tasklet getTasklet() { + return tasklet; + } + /** * A callback for the transactional work inside a chunk. Also detects * failures in the transaction commit and rollback, only panicking if the diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/launch/support/SimpleJobOperatorTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/launch/support/SimpleJobOperatorTests.java index 96813e5af..fa0e4cbdb 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/launch/support/SimpleJobOperatorTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/launch/support/SimpleJobOperatorTests.java @@ -19,10 +19,12 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +import static org.mockito.Matchers.anyString; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; import java.util.Arrays; +import java.util.Collection; import java.util.Collections; import java.util.HashSet; import java.util.List; @@ -35,12 +37,17 @@ import org.junit.Test; import org.springframework.batch.core.BatchStatus; import org.springframework.batch.core.Job; import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobExecutionException; import org.springframework.batch.core.JobInstance; import org.springframework.batch.core.JobParameters; import org.springframework.batch.core.JobParametersIncrementer; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.StepContribution; +import org.springframework.batch.core.configuration.JobRegistry; import org.springframework.batch.core.configuration.support.MapJobRegistry; import org.springframework.batch.core.converter.DefaultJobParametersConverter; import org.springframework.batch.core.explore.JobExplorer; +import org.springframework.batch.core.job.AbstractJob; import org.springframework.batch.core.job.JobSupport; import org.springframework.batch.core.launch.JobInstanceAlreadyExistsException; import org.springframework.batch.core.launch.JobLauncher; @@ -51,6 +58,10 @@ import org.springframework.batch.core.repository.JobExecutionAlreadyRunningExcep import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException; import org.springframework.batch.core.repository.JobRepository; import org.springframework.batch.core.repository.JobRestartException; +import org.springframework.batch.core.scope.context.ChunkContext; +import org.springframework.batch.core.step.tasklet.StoppableTasklet; +import org.springframework.batch.core.step.tasklet.TaskletStep; +import org.springframework.batch.repeat.RepeatStatus; import org.springframework.batch.support.PropertiesConverter; /** @@ -254,6 +265,7 @@ public class SimpleJobOperatorTests { } @Test + @SuppressWarnings("unchecked") public void testFindRunningExecutionsNoSuchJob() throws Exception { jobParameters = new JobParameters(); when(jobExplorer.findRunningJobExecutions("no-such-job")).thenReturn(Collections.EMPTY_SET); @@ -348,6 +360,67 @@ public class SimpleJobOperatorTests { assertEquals(BatchStatus.STOPPING, jobExecution.getStatus()); } + @Test + public void testStopTasklet() throws Exception { + JobInstance jobInstance = new JobInstance(123L, job.getName()); + JobExecution jobExecution = new JobExecution(jobInstance, 111L, jobParameters, null); + StoppableTasklet tasklet = mock(StoppableTasklet.class); + TaskletStep taskletStep = new TaskletStep(); + taskletStep.setTasklet(tasklet); + MockJob job = new MockJob(); + job.taskletStep = taskletStep; + + JobRegistry jobRegistry = mock(JobRegistry.class); + TaskletStep step = mock(TaskletStep.class); + + when(step.getTasklet()).thenReturn(tasklet); + when(step.getName()).thenReturn("test_job.step1"); + when(jobRegistry.getJob(anyString())).thenReturn(job); + when(jobExplorer.getJobExecution(111L)).thenReturn(jobExecution); + + jobOperator.setJobRegistry(jobRegistry); + jobExplorer.getJobExecution(111L); + jobRepository.update(jobExecution); + jobOperator.stop(111L); + assertEquals(BatchStatus.STOPPING, jobExecution.getStatus()); + } + + @Test + public void testStopTaskletException() throws Exception { + JobInstance jobInstance = new JobInstance(123L, job.getName()); + JobExecution jobExecution = new JobExecution(jobInstance, 111L, jobParameters, null); + StoppableTasklet tasklet = new StoppableTasklet() { + + @Override + public RepeatStatus execute(StepContribution contribution, + ChunkContext chunkContext) throws Exception { + return null; + } + + @Override + public void stop() { + throw new IllegalStateException(); + }}; + TaskletStep taskletStep = new TaskletStep(); + taskletStep.setTasklet(tasklet); + MockJob job = new MockJob(); + job.taskletStep = taskletStep; + + JobRegistry jobRegistry = mock(JobRegistry.class); + TaskletStep step = mock(TaskletStep.class); + + when(step.getTasklet()).thenReturn(tasklet); + when(step.getName()).thenReturn("test_job.step1"); + when(jobRegistry.getJob(anyString())).thenReturn(job); + when(jobExplorer.getJobExecution(111L)).thenReturn(jobExecution); + + jobOperator.setJobRegistry(jobRegistry); + jobExplorer.getJobExecution(111L); + jobRepository.update(jobExecution); + jobOperator.stop(111L); + assertEquals(BatchStatus.STOPPING, jobExecution.getStatus()); + } + @Test public void testAbort() throws Exception { JobInstance jobInstance = new JobInstance(123L, job.getName()); @@ -369,4 +442,25 @@ public class SimpleJobOperatorTests { jobRepository.update(jobExecution); jobOperator.abandon(123L); } + + class MockJob extends AbstractJob { + + private TaskletStep taskletStep; + + @Override + public Step getStep(String stepName) { + return taskletStep; + } + + @Override + public Collection getStepNames() { + return Arrays.asList("test_job.step1"); + } + + @Override + protected void doExecute(JobExecution execution) throws JobExecutionException { + + } + + } } 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 ff692f571..bc861564b 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 @@ -221,6 +221,26 @@ public class SystemCommandTaskletIntegrationTests { tasklet.setWorkingDirectory(directory.getCanonicalPath()); } + /* + * test stopping a tasklet + */ + @Test + public void testStopped() throws Exception { + String command = System.getProperty("os.name").toLowerCase().indexOf("win") >= 0 ? + "ping 1.1.1.1 -n 1 -w 5000" : + "sleep 5"; + tasklet.setCommand(command); + tasklet.setTerminationCheckInterval(10); + tasklet.afterPropertiesSet(); + + StepContribution contribution = stepExecution.createStepContribution(); + //send stop + tasklet.stop(); + tasklet.execute(contribution, null); + + assertEquals(contribution.getExitStatus().getExitCode(),ExitStatus.STOPPED.getExitCode()); + } + /** * Exit code mapper containing mapping logic expected by the tests. 0 means * finished successfully, other value means failure.