From 7d8abccc0ef6245c6738453317eab317318ffb8f Mon Sep 17 00:00:00 2001 From: dsyer Date: Fri, 30 Nov 2007 12:11:47 +0000 Subject: [PATCH] IN PROGRESS - issue BATCH-241: Support for JobLauncher.stop() on job that hasn't started yet (delayed execution) http://opensource.atlassian.com/projects/spring/browse/BATCH-241 Added some more internal bookkeeping in the JobLauncher so that it can recognise a job that it was asked to start, but hasn't been executed by the task executor yet. A side effect is that the JobExecution for such a job is never updated, so I'll keep this issue open until that is also fixed. --- .../execution/launch/SimpleJobLauncher.java | 90 +++++++++++-- .../launch/TaskExecutorJobLauncherTests.java | 120 ++++++++++++------ 2 files changed, 160 insertions(+), 50 deletions(-) diff --git a/execution/src/main/java/org/springframework/batch/execution/launch/SimpleJobLauncher.java b/execution/src/main/java/org/springframework/batch/execution/launch/SimpleJobLauncher.java index 1c738cc57..781092a35 100644 --- a/execution/src/main/java/org/springframework/batch/execution/launch/SimpleJobLauncher.java +++ b/execution/src/main/java/org/springframework/batch/execution/launch/SimpleJobLauncher.java @@ -237,16 +237,16 @@ public class SimpleJobLauncher implements JobLauncher, InitializingBean, throws NoSuchJobConfigurationException { JobIdentifier jobIdentifier = execution.getJob().getIdentifier(); - - synchronized (monitor) { - if (isInternalRunning(jobIdentifier)) { - return; - } + + if (getJobExecution(jobIdentifier)==null) { + logger.info("Job already stopped (not launching): "+jobIdentifier); + return; } - register(execution); try { + logger.info("Launching: "+jobIdentifier); jobExecutorFacade.start(execution); + logger.info("Completed successfully: "+jobIdentifier); } finally { unregister(jobIdentifier); } @@ -266,7 +266,7 @@ public class SimpleJobLauncher implements JobLauncher, InitializingBean, throws NoSuchJobConfigurationException, JobExecutionAlreadyRunningException { - if (get(jobIdentifier) != null) { + if (getJobExecution(jobIdentifier) != null) { throw new JobExecutionAlreadyRunningException( "A job is already executing with this identifier: [" + jobIdentifier + "]"); @@ -275,11 +275,22 @@ public class SimpleJobLauncher implements JobLauncher, InitializingBean, .createExecutionFrom(jobIdentifier); // TODO: throw JobExecutionAlreadyRunningException if it is in a running // state (someone else launched it) + final JobExecutionHolder holder = register(execution); taskExecutor.execute(new Runnable() { public void run() { try { + + synchronized (monitor) { + if (isInternalRunning(jobIdentifier)) { + logger.info("This job is already running, so not re-launched: "+jobIdentifier); + return; + } + } + + holder.start(); runInternal(execution); + } catch (NoSuchJobConfigurationException e) { applicationEventPublisher .publishEvent(new RepeatOperationsApplicationEvent( @@ -288,6 +299,8 @@ public class SimpleJobLauncher implements JobLauncher, InitializingBean, logger.error( "JobConfiguration could not be located inside Runnable for identifier: [" + jobIdentifier + "]", e); + } finally { + holder.stop(); } } }); @@ -347,10 +360,12 @@ public class SimpleJobLauncher implements JobLauncher, InitializingBean, */ protected void doStop(JobIdentifier jobIdentifier) throws NoSuchJobExecutionException { - JobExecution execution = get(jobIdentifier); + JobExecution execution = getJobExecution(jobIdentifier); + logger.info("Stopping job: "+jobIdentifier); if (execution != null) { jobExecutorFacade.stop(execution); } + unregister(jobIdentifier); } /** @@ -420,8 +435,9 @@ public class SimpleJobLauncher implements JobLauncher, InitializingBean, private boolean isInternalRunning(JobIdentifier jobIdentifier) { synchronized (registry) { + JobExecutionHolder jobExecutionHolder = getJobExecutionHolder(jobIdentifier); return isRunning(jobIdentifier) - && registry.containsKey(jobIdentifier); + && jobExecutionHolder!=null && jobExecutionHolder.isRunning(); } } @@ -443,11 +459,14 @@ public class SimpleJobLauncher implements JobLauncher, InitializingBean, * Convenient synchronized accessor for the registry. * * @param jobIdentifier + * @return TODO */ - private void register(JobExecution execution) { + private JobExecutionHolder register(JobExecution execution) { + JobExecutionHolder jobExecutionHolder = new JobExecutionHolder(execution); synchronized (registry) { - registry.put(execution.getJob().getIdentifier(), execution); + registry.put(execution.getJob().getIdentifier(), jobExecutionHolder); } + return jobExecutionHolder; } /** @@ -455,10 +474,24 @@ public class SimpleJobLauncher implements JobLauncher, InitializingBean, * * @param jobIdentifier */ - private JobExecution get(JobIdentifier jobIdentifier) { + private JobExecution getJobExecution(JobIdentifier jobIdentifier) { synchronized (registry) { if (registry.containsKey(jobIdentifier)) { - return (JobExecution) registry.get(jobIdentifier); + return ((JobExecutionHolder) registry.get(jobIdentifier)).getExecution(); + } + } + return null; + } + + /** + * Convenient synchronized accessor for the registry. + * + * @param jobIdentifier + */ + private JobExecutionHolder getJobExecutionHolder(JobIdentifier jobIdentifier) { + synchronized (registry) { + if (registry.containsKey(jobIdentifier)) { + return (JobExecutionHolder) registry.get(jobIdentifier); } } return null; @@ -514,4 +547,35 @@ public class SimpleJobLauncher implements JobLauncher, InitializingBean, this.applicationEventPublisher = applicationEventPublisher; } + private class JobExecutionHolder { + + private static final int NEW = 0; + private static final int STARTED = 1; + private static final int STOPPED = 2; + + private JobExecution execution; + private int status = NEW; + + public JobExecutionHolder(JobExecution execution) { + this.execution = execution; + } + + JobExecution getExecution() { + return execution; + } + + boolean isRunning() { + return status==STARTED; + } + + void start() { + status = STARTED; + } + + void stop() { + status = STOPPED; + } + + } + } diff --git a/execution/src/test/java/org/springframework/batch/execution/launch/TaskExecutorJobLauncherTests.java b/execution/src/test/java/org/springframework/batch/execution/launch/TaskExecutorJobLauncherTests.java index 338c8da58..03bc0e363 100644 --- a/execution/src/test/java/org/springframework/batch/execution/launch/TaskExecutorJobLauncherTests.java +++ b/execution/src/test/java/org/springframework/batch/execution/launch/TaskExecutorJobLauncherTests.java @@ -19,6 +19,8 @@ package org.springframework.batch.execution.launch; import java.util.ArrayList; import java.util.List; import java.util.Properties; +import java.util.Timer; +import java.util.TimerTask; import junit.framework.TestCase; @@ -30,11 +32,13 @@ import org.springframework.batch.core.domain.JobIdentifier; import org.springframework.batch.core.domain.JobInstance; import org.springframework.batch.core.runtime.SimpleJobIdentifier; import org.springframework.batch.core.runtime.SimpleJobIdentifierFactory; +import org.springframework.batch.repeat.ExitStatus; import org.springframework.batch.statistics.StatisticsProvider; import org.springframework.batch.support.PropertiesConverter; import org.springframework.context.ApplicationEvent; import org.springframework.context.ApplicationEventPublisher; import org.springframework.core.task.SimpleAsyncTaskExecutor; +import org.springframework.scheduling.timer.TimerTaskExecutor; public class TaskExecutorJobLauncherTests extends TestCase { @@ -54,7 +58,7 @@ public class TaskExecutorJobLauncherTests extends TestCase { launcher.setJobExecutorFacade(container); launcher.setJobConfigurationName(new JobConfiguration("foo").getName()); - launcher.run(); + JobExecution execution = launcher.run(); // give the thread some time to start up... Thread.sleep(100); assertTrue(launcher.isRunning()); @@ -62,6 +66,47 @@ public class TaskExecutorJobLauncherTests extends TestCase { // ...and to shut down: Thread.sleep(400); assertFalse(launcher.isRunning()); + assertEquals("COMPLETED_BY_TEST", execution.getExitStatus().getExitCode()); + } + + public void testStopContainerWhenJobNotRunning() throws Exception { + + final List list = new ArrayList(); + + // Important (otherwise start() does not return!) + TimerTaskExecutor taskExecutor = new TimerTaskExecutor(new Timer() { + public void schedule(final TimerTask task, long delay) { + TimerTask wrapper = new TimerTask() { + public void run() { + list.add(task); + task.run(); + } + }; + super.schedule(wrapper, 400); + } + }); + taskExecutor.afterPropertiesSet(); + launcher.setTaskExecutor(taskExecutor); + + InterruptibleContainer container = new InterruptibleContainer(); + launcher.setJobExecutorFacade(container); + launcher.setJobConfigurationName("foo"); + + JobExecution execution = launcher.run(); + // give the thread some time to start up... + Thread.sleep(100); + // The launcher thinks it has started the job... + assertTrue(launcher.isRunning()); + // ...but the task has not been started yet + assertEquals(0, list.size()); + launcher.stop(); + // ...and to shut down: + Thread.sleep(1000); + assertFalse(launcher.isRunning()); + // The timer task has been started... + assertEquals(1, list.size()); + // ...but the job is not executed + assertEquals(ExitStatus.UNKNOWN, execution.getExitStatus()); } public void testRunTwice() throws Exception { @@ -125,42 +170,6 @@ public class TaskExecutorJobLauncherTests extends TestCase { control.verify(); } - private class InterruptibleContainer implements JobExecutorFacade { - private volatile boolean running = true; - - public void start() { - while (running) { - try { - // 1 seconds should be long enough to allow the thread to be - // started and - // for interrupt to be called; - Thread.sleep(300); - } catch (InterruptedException ex) { - // thread interrupted, allow to exit normally - } - } - } - - public void start(JobExecution execution) - throws NoSuchJobConfigurationException { - start(); - } - - public JobExecution createExecutionFrom(JobIdentifier jobIdentifier) - throws NoSuchJobConfigurationException { - return new JobExecution(new JobInstance(jobIdentifier)); - } - - public void stop(JobExecution execution) { - running = false; - } - - public boolean isRunning() { - // not needed - return false; - } - } - public void testPublishApplicationEvent() throws Exception { final List list = new ArrayList(); launcher.setApplicationEventPublisher(new ApplicationEventPublisher() { @@ -188,6 +197,43 @@ public class TaskExecutorJobLauncherTests extends TestCase { control.verify(); } + private class InterruptibleContainer implements JobExecutorFacade { + private volatile boolean running = true; + + private void start() { + while (running) { + try { + // 1 seconds should be long enough to allow the thread to be + // started and + // for interrupt to be called; + Thread.sleep(300); + } catch (InterruptedException ex) { + // thread interrupted, allow to exit normally + } + } + } + + public void start(JobExecution execution) + throws NoSuchJobConfigurationException { + start(); + execution.setExitStatus(new ExitStatus(false, "COMPLETED_BY_TEST")); + } + + public JobExecution createExecutionFrom(JobIdentifier jobIdentifier) + throws NoSuchJobConfigurationException { + return new JobExecution(new JobInstance(jobIdentifier)); + } + + public void stop(JobExecution execution) { + running = false; + } + + public boolean isRunning() { + // not needed + return false; + } + } + private interface JobExecutorFacadeWithStatistics extends JobExecutorFacade, StatisticsProvider { }