Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

IN PROGRESS - issue BATCH-241: Support for JobLauncher.stop() on job that hasn't started yet (delayed execution)

Browse Source 
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.
This commit is contained in:
dsyer
2007-11-30 12:11:47 +00:00
parent 42a8a78259
commit 7d8abccc0e
2 changed files with 160 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

90
execution/src/main/java/org/springframework/batch/execution/launch/SimpleJobLauncher.java
View File

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

120
execution/src/test/java/org/springframework/batch/execution/launch/TaskExecutorJobLauncherTests.java
View File

@@ -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 {
}