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

IN PROGRESS - issue BATCH-159: JobExecutor should return a JobExecution (which itself contains the ExitStatus)

Browse Source 
http://opensource.atlassian.com/projects/spring/browse/BATCH-159

Move JobExecutionListener to core and add some Javadocs.
This commit is contained in:
dsyer
2007-10-23 18:45:48 +00:00
parent afe75ab8cc
commit 1981dedbf0
7 changed files with 130 additions and 184 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
execution/src/main/java/org/springframework/batch/execution/facade/JobExecutionListener.java
View File

@@ -1,43 +0,0 @@
/*
* Copyright 2006-2007 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.execution.facade;
import org.springframework.batch.core.domain.JobExecution;
/**
* Listener interface for the job execution lifecycle.
*
* @author Dave Syer
*
*/
public interface JobExecutionListener {
/**
* Callback for the start of a job, before any steps are processed.
*
* @param execution
* the current {@link JobExecution}
*/
void before(JobExecution execution);
/**
* Callback for the start of a job, after all steps are processed, or on an
* error.
*
* @param execution
*/
void after(JobExecution execution);
}

47
execution/src/main/java/org/springframework/batch/execution/facade/JobExecutionListenerSupport.java
View File

@@ -1,47 +0,0 @@
/*
* Copyright 2006-2007 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.execution.facade;
import org.springframework.batch.core.domain.JobExecution;
/**
* Simple no-op implementation of {@link JobExecutionListener} which does
* nothing.
*
* @author Dave Syer
*
*/
public class JobExecutionListenerSupport implements JobExecutionListener {
/**
* No-op for subclasses to extend.
*
* @see org.springframework.batch.execution.facade.JobExecutionListener#after(org.springframework.batch.core.domain.JobExecution)
*/
public void after(JobExecution execution) {
// no-op
}
/**
* No-op for subclasses to extend.
*
* @see org.springframework.batch.execution.facade.JobExecutionListener#before(org.springframework.batch.core.domain.JobExecution)
*/
public void before(JobExecution execution) {
// no-op
}
}

13
execution/src/main/java/org/springframework/batch/execution/facade/SimpleJobExecutorFacade.java
View File

@@ -30,6 +30,7 @@ import org.springframework.batch.core.configuration.NoSuchJobConfigurationExcept
import org.springframework.batch.core.domain.JobExecution;
import org.springframework.batch.core.domain.JobIdentifier;
import org.springframework.batch.core.domain.JobInstance;
import org.springframework.batch.core.executor.JobExecutionListener;
import org.springframework.batch.core.executor.JobExecutor;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.execution.job.DefaultJobExecutor;
@@ -142,17 +143,7 @@ public class SimpleJobExecutorFacade implements JobExecutorFacade,
jobIdentifier);
JobExecution jobExecution = new JobExecution(job);
try {
this.before(jobExecution);
jobExecutor.run(jobConfiguration, jobExecution);
} finally {
this.after(jobExecution);
}
jobExecutor.run(jobConfiguration, jobExecution, this);
return jobExecution.getExitStatus();
}

111
execution/src/main/java/org/springframework/batch/execution/job/DefaultJobExecutor.java
View File

@@ -20,6 +20,7 @@ import java.sql.Timestamp;
import java.util.Iterator;
import java.util.List;
import org.springframework.batch.common.ExceptionClassifier;
import org.springframework.batch.core.configuration.JobConfiguration;
import org.springframework.batch.core.configuration.StepConfiguration;
import org.springframework.batch.core.domain.BatchStatus;
@@ -27,8 +28,8 @@ import org.springframework.batch.core.domain.JobExecution;
import org.springframework.batch.core.domain.JobInstance;
import org.springframework.batch.core.domain.StepExecution;
import org.springframework.batch.core.domain.StepInstance;
import org.springframework.batch.core.executor.AbstractJobExecutor;
import org.springframework.batch.core.executor.ExitCodeExceptionClassifier;
import org.springframework.batch.core.executor.JobExecutor;
import org.springframework.batch.core.executor.StepExecutor;
import org.springframework.batch.core.executor.StepExecutorFactory;
import org.springframework.batch.core.executor.StepInterruptedException;
@@ -40,63 +41,72 @@ import org.springframework.batch.repeat.ExitStatus;
import org.springframework.batch.repeat.RepeatContext;
/**
* Default implementation of (@link JobExecutor} interface. Sequentially executes a
* job by iterating it's life of steps. Interruption of a job run is pluggable
* by passing in various interruption policies.
* Default implementation of (@link JobExecutor} interface. Sequentially
* executes a job by iterating it's life of steps.
*
* @author Lucas Ward
* @author Dave Syer
*/
public class DefaultJobExecutor implements JobExecutor {
public class DefaultJobExecutor extends AbstractJobExecutor {
private static final SimpleStepExecutorFactory DEFAULT_STEP_EXECUTOR_FACTORY = new SimpleStepExecutorFactory();
private JobRepository jobRepository;
private StepExecutorFactory stepExecutorFactory = DEFAULT_STEP_EXECUTOR_FACTORY;
private ExitCodeExceptionClassifier exceptionClassifier = new SimpleExitCodeExceptionClassifier();
public ExitStatus run(JobConfiguration configuration, JobExecution jobExecution)
/**
* Run the specified job by looping through the steps and delegating to the
* {@link StepExecutor}.
*
* @see org.springframework.batch.core.executor.JobExecutor#run(org.springframework.batch.core.configuration.JobConfiguration,
* org.springframework.batch.core.domain.JobExecution)
*/
public ExitStatus run(JobConfiguration configuration, JobExecution execution)
throws BatchCriticalException {
JobInstance job = jobExecution.getJob();
updateStatus(jobExecution, BatchStatus.STARTING);
JobInstance job = execution.getJob();
updateStatus(execution, BatchStatus.STARTING);
List steps = job.getSteps();
ExitStatus status = ExitStatus.FAILED;
try {
for (Iterator i = steps.iterator(), j = configuration.getStepConfigurations().iterator(); i.hasNext()
for (Iterator i = steps.iterator(), j = configuration
.getStepConfigurations().iterator(); i.hasNext()
&& j.hasNext();) {
StepInstance step = (StepInstance) i.next();
StepConfiguration stepConfiguration = (StepConfiguration) j.next();
StepConfiguration stepConfiguration = (StepConfiguration) j
.next();
if (shouldStart(step, stepConfiguration)) {
updateStatus(jobExecution, BatchStatus.STARTED);
StepExecutor stepExecutor = stepExecutorFactory.getExecutor(stepConfiguration);
StepExecution stepExecution = new StepExecution(step, jobExecution);
status = stepExecutor.process(stepConfiguration, stepExecution);
updateStatus(execution, BatchStatus.STARTED);
StepExecutor stepExecutor = stepExecutorFactory
.getExecutor(stepConfiguration);
StepExecution stepExecution = new StepExecution(step,
execution);
status = stepExecutor.process(stepConfiguration,
stepExecution);
}
}
updateStatus(jobExecution, BatchStatus.COMPLETED);
}
catch (StepInterruptedException e) {
updateStatus(jobExecution, BatchStatus.STOPPED);
updateStatus(execution, BatchStatus.COMPLETED);
} catch (StepInterruptedException e) {
updateStatus(execution, BatchStatus.STOPPED);
status = exceptionClassifier.classifyForExitCode(e);
rethrow(e);
}
catch (Throwable t) {
updateStatus(jobExecution, BatchStatus.FAILED);
} catch (Throwable t) {
updateStatus(execution, BatchStatus.FAILED);
status = exceptionClassifier.classifyForExitCode(t);
rethrow(t);
} finally {
execution.setEndTime(new Timestamp(System.currentTimeMillis()));
execution.setExitStatus(status);
jobRepository.saveOrUpdate(execution);
}
finally {
jobExecution.setEndTime(new Timestamp(System.currentTimeMillis()));
jobExecution.setExitStatus(status);
jobRepository.saveOrUpdate(jobExecution);
}
return status;
}
@@ -106,7 +116,8 @@ public class DefaultJobExecutor implements JobExecutor {
job.setStatus(status);
jobRepository.update(job);
jobRepository.saveOrUpdate(jobExecution);
for (Iterator iter = jobExecution.getStepContexts().iterator(); iter.hasNext();) {
for (Iterator iter = jobExecution.getStepContexts().iterator(); iter
.hasNext();) {
RepeatContext context = (RepeatContext) iter.next();
context.setAttribute("JOB_STATUS", status);
}
@@ -116,9 +127,11 @@ public class DefaultJobExecutor implements JobExecutor {
* Given a step and configuration, return true if the step should start,
* false if it should not, and throw an exception if the job should finish.
*/
private boolean shouldStart(StepInstance step, StepConfiguration stepConfiguration) {
private boolean shouldStart(StepInstance step,
StepConfiguration stepConfiguration) {
if (step.getStatus() == BatchStatus.COMPLETED && stepConfiguration.isAllowStartIfComplete() == false) {
if (step.getStatus() == BatchStatus.COMPLETED
&& stepConfiguration.isAllowStartIfComplete() == false) {
// step is complete, false should be returned, indicated that the
// step should
// not be started
@@ -128,11 +141,11 @@ public class DefaultJobExecutor implements JobExecutor {
if (step.getStepExecutionCount() < stepConfiguration.getStartLimit()) {
// step start count is less than start max, return true
return true;
}
else {
} else {
// start max has been exceeded, throw an exception.
throw new BatchCriticalException("Maximum start limit exceeded for step: " + step.getName() + "StartMax: "
+ stepConfiguration.getStartLimit());
throw new BatchCriticalException(
"Maximum start limit exceeded for step: " + step.getName()
+ "StartMax: " + stepConfiguration.getStartLimit());
}
}
@@ -142,21 +155,41 @@ public class DefaultJobExecutor implements JobExecutor {
private static void rethrow(Throwable t) throws RuntimeException {
if (t instanceof RuntimeException) {
throw (RuntimeException) t;
}
else {
} else {
throw new BatchCriticalException(t);
}
}
/**
* Public setter for the {@link JobRepository} that is needed to manage the
* state of the batch meta domain (jobs, steps, executions) during the life
* of a job.
*
* @param jobRepository
*/
public void setJobRepository(JobRepository jobRepository) {
this.jobRepository = jobRepository;
DEFAULT_STEP_EXECUTOR_FACTORY.setJobRepository(jobRepository);
}
public void setStepExecutorFactory(StepExecutorFactory stepExecutorResolver) {
this.stepExecutorFactory = stepExecutorResolver;
/**
* Setter for injecting a {@link StepExecutorFactory}. The factory is
* responsible for providing a {@link StepExecutor} to execute each step in
* turn. The values returned from the factory are not cached or re-used by
* this implementation.
*
* @param stepExecutorFactory
*/
public void setStepExecutorFactory(StepExecutorFactory stepExecutorFactory) {
this.stepExecutorFactory = stepExecutorFactory;
}
/**
* Public setter for injecting an {@link ExceptionClassifier} that can
* translate exceptions to {@link ExitStatus}.
*
* @param exceptionClassifier
*/
public void setExceptionClassifier(
ExitCodeExceptionClassifier exceptionClassifier) {
this.exceptionClassifier = exceptionClassifier;

2
execution/src/test/java/org/springframework/batch/execution/bootstrap/SimpleJobLauncherTests.java
View File

@@ -21,8 +21,8 @@ import junit.framework.TestCase;
import org.springframework.batch.core.configuration.JobConfiguration;
import org.springframework.batch.core.configuration.NoSuchJobConfigurationException;
import org.springframework.batch.core.domain.JobIdentifier;
import org.springframework.batch.core.executor.JobExecutionListener;
import org.springframework.batch.core.runtime.SimpleJobIdentifierFactory;
import org.springframework.batch.execution.facade.JobExecutionListener;
import org.springframework.batch.execution.facade.JobExecutorFacade;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.core.task.SimpleAsyncTaskExecutor;

2
execution/src/test/java/org/springframework/batch/execution/bootstrap/TaskExecutorJobLauncherTests.java
View File

@@ -28,9 +28,9 @@ import org.easymock.MockControl;
import org.springframework.batch.core.configuration.JobConfiguration;
import org.springframework.batch.core.configuration.NoSuchJobConfigurationException;
import org.springframework.batch.core.domain.JobIdentifier;
import org.springframework.batch.core.executor.JobExecutionListener;
import org.springframework.batch.core.runtime.SimpleJobIdentifier;
import org.springframework.batch.core.runtime.SimpleJobIdentifierFactory;
import org.springframework.batch.execution.facade.JobExecutionListener;
import org.springframework.batch.execution.facade.JobExecutorFacade;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.batch.repeat.interceptor.RepeatOperationsApplicationEvent;

96
execution/src/test/java/org/springframework/batch/execution/facade/SimpleJobExecutorFacadeTests.java
View File

@@ -30,6 +30,8 @@ import org.springframework.batch.core.configuration.JobConfigurationLocator;
import org.springframework.batch.core.configuration.NoSuchJobConfigurationException;
import org.springframework.batch.core.domain.JobExecution;
import org.springframework.batch.core.domain.JobInstance;
import org.springframework.batch.core.executor.AbstractJobExecutor;
import org.springframework.batch.core.executor.JobExecutionListenerSupport;
import org.springframework.batch.core.executor.JobExecutor;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.runtime.SimpleJobIdentifier;
@@ -52,15 +54,17 @@ public class SimpleJobExecutorFacadeTests extends TestCase {
private JobRepository jobRepository;
private MockControl jobRepositoryControl = MockControl.createControl(JobRepository.class);
private MockControl jobRepositoryControl = MockControl
.createControl(JobRepository.class);
private JobConfiguration jobConfiguration = new JobConfiguration();
private volatile boolean running = false;
private SimpleJobIdentifier jobIdentifier = new SimpleJobIdentifier();
private JobExecution jobExecution = new JobExecution(new JobInstance(jobIdentifier));
private JobExecution jobExecution = new JobExecution(new JobInstance(
jobIdentifier));
private List list = new ArrayList();
@@ -80,66 +84,70 @@ public class SimpleJobExecutorFacadeTests extends TestCase {
assertEquals(job, jobExecution.getJob());
assertEquals("bar", job.getName());
jobRepositoryControl.verify();
}
private JobInstance setUpFacadeForNormalStart() {
jobIdentifier = new SimpleJobIdentifier("bar");
jobRepository.findOrCreateJob(jobConfiguration, jobIdentifier);
jobExecutor = new JobExecutor() {
public ExitStatus run(JobConfiguration configuration, JobExecution jobExecutionContext) throws BatchCriticalException {
jobExecution = jobExecutionContext;
jobExecutor = new AbstractJobExecutor() {
public ExitStatus run(JobConfiguration configuration,
JobExecution execution)
throws BatchCriticalException {
jobExecution = execution;
return ExitStatus.FINISHED;
}
};
jobExecutorFacade.setJobExecutor(jobExecutor);
JobInstance job = new JobInstance(jobIdentifier);
jobExecution = new JobExecution(job);
jobRepositoryControl.setReturnValue(job);
jobRepositoryControl.replay();
jobExecutorFacade.setJobConfigurationLocator(new JobConfigurationLocator() {
public JobConfiguration getJobConfiguration(String name) throws NoSuchJobConfigurationException {
return jobConfiguration;
}
});
jobExecutorFacade
.setJobConfigurationLocator(new JobConfigurationLocator() {
public JobConfiguration getJobConfiguration(String name)
throws NoSuchJobConfigurationException {
return jobConfiguration;
}
});
return job;
}
public void testIsRunning() throws Exception {
jobExecutorFacade.setJobExecutor(new JobExecutor() {
public ExitStatus run(JobConfiguration configuration, JobExecution jobExecutionContext)
jobExecutorFacade.setJobExecutor(new AbstractJobExecutor() {
public ExitStatus run(JobConfiguration configuration,
JobExecution execution)
throws BatchCriticalException {
while (running) {
try {
Thread.sleep(100L);
} catch (InterruptedException e) {
throw new BatchCriticalException(
"Interrupted unexpectedly!");
}
catch (InterruptedException e) {
throw new BatchCriticalException("Interrupted unexpectedly!");
}
}
return ExitStatus.FINISHED;
}
});
jobExecutorFacade.setJobConfigurationLocator(new JobConfigurationLocator() {
public JobConfiguration getJobConfiguration(String name) throws NoSuchJobConfigurationException {
return jobConfiguration;
}
});
jobExecutorFacade
.setJobConfigurationLocator(new JobConfigurationLocator() {
public JobConfiguration getJobConfiguration(String name)
throws NoSuchJobConfigurationException {
return jobConfiguration;
}
});
final SimpleJobIdentifier jobIdentifier = new SimpleJobIdentifier("foo");
jobRepository.findOrCreateJob(jobConfiguration, jobIdentifier);
JobInstance job = new JobInstance(jobIdentifier);
jobRepositoryControl.setReturnValue(job);
jobRepositoryControl.replay();
running = true;
new Thread(new Runnable() {
public void run() {
try {
jobExecutorFacade.start(jobIdentifier);
}
catch (NoSuchJobConfigurationException e) {
} catch (NoSuchJobConfigurationException e) {
System.err.println("Shouldn't happen");
}
}
@@ -149,7 +157,7 @@ public class SimpleJobExecutorFacadeTests extends TestCase {
assertTrue(jobExecutorFacade.isRunning());
running = false;
int count = 0;
while(jobExecutorFacade.isRunning() && count ++<5) {
while (jobExecutorFacade.isRunning() && count++ < 5) {
Thread.sleep(100L);
}
assertFalse(jobExecutorFacade.isRunning());
@@ -163,30 +171,33 @@ public class SimpleJobExecutorFacadeTests extends TestCase {
try {
jobExecutorFacade.start(new SimpleJobIdentifier("TestJob"));
fail("Expected IllegalStateException");
}
catch (IllegalStateException ex) {
} catch (IllegalStateException ex) {
// expected
}
}
public void testStopWithNoJob() throws Exception {
SimpleJobIdentifier runtimeInformation = new SimpleJobIdentifier("TestJob");
SimpleJobIdentifier runtimeInformation = new SimpleJobIdentifier(
"TestJob");
try {
jobExecutorFacade.stop(runtimeInformation);
fail("Expected NoSuchJobExecutionException");
} catch (NoSuchJobExecutionException e) {
// expected
assertTrue(e.getMessage().indexOf("TestJob")>=0);
assertTrue(e.getMessage().indexOf("TestJob") >= 0);
}
}
public void testStop() throws Exception {
SimpleJobIdentifier runtimeInformation = new SimpleJobIdentifier("TestJob");
JobExecution execution = new JobExecution(new JobInstance(runtimeInformation, new Long(0)));
SimpleJobIdentifier runtimeInformation = new SimpleJobIdentifier(
"TestJob");
JobExecution execution = new JobExecution(new JobInstance(
runtimeInformation, new Long(0)));
registerExecution(runtimeInformation, execution);
RepeatContextSupport stepContext = new RepeatContextSupport(null);
RepeatContextSupport chunkContext = new RepeatContextSupport(stepContext);
RepeatContextSupport chunkContext = new RepeatContextSupport(
stepContext);
execution.registerStepContext(stepContext);
execution.registerChunkContext(chunkContext);
jobExecutorFacade.stop(runtimeInformation);
@@ -200,8 +211,10 @@ public class SimpleJobExecutorFacadeTests extends TestCase {
}
public void testStatisticsWithContext() throws Exception {
SimpleJobIdentifier runtimeInformation = new SimpleJobIdentifier("TestJob");
JobExecution execution = new JobExecution(new JobInstance(runtimeInformation, new Long(0)));
SimpleJobIdentifier runtimeInformation = new SimpleJobIdentifier(
"TestJob");
JobExecution execution = new JobExecution(new JobInstance(
runtimeInformation, new Long(0)));
registerExecution(runtimeInformation, execution);
execution.registerStepContext(new RepeatContextSupport(null));
Properties statistics = jobExecutorFacade.getStatistics();
@@ -226,7 +239,7 @@ public class SimpleJobExecutorFacadeTests extends TestCase {
assertEquals(2, list.size());
assertEquals("two", list.get(1));
}
public void testOrderedListenersCalledFirstOnBefore() throws Exception {
List listeners = new ArrayList();
listeners.add(new JobExecutionListenerSupport() {
@@ -248,12 +261,11 @@ public class SimpleJobExecutorFacadeTests extends TestCase {
private void registerExecution(SimpleJobIdentifier runtimeInformation,
JobExecution execution) throws NoSuchFieldException,
IllegalAccessException {
Field field = SimpleJobExecutorFacade.class.getDeclaredField("jobExecutionRegistry");
Field field = SimpleJobExecutorFacade.class
.getDeclaredField("jobExecutionRegistry");
ReflectionUtils.makeAccessible(field);
Map map = (Map) field.get(jobExecutorFacade);
map.put(runtimeInformation, execution);
}
}