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

OPEN - issue BATCH-303: Replace repeat contexts in core domain with boolean flag that can be checked by executors

Browse Source 
http://jira.springframework.org/browse/BATCH-303
This commit is contained in:
dsyer
2008-01-22 13:45:02 +00:00
parent 3d5c2b44e3
commit 552549f897
12 changed files with 149 additions and 323 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-batch-execution/src/main/java/org/springframework/batch/execution/job/DefaultJobExecutor.java
View File

@@ -38,7 +38,6 @@ import org.springframework.batch.execution.step.simple.SimpleExitCodeExceptionCl
import org.springframework.batch.execution.step.simple.SimpleStepExecutorFactory;
import org.springframework.batch.io.exception.BatchCriticalException;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.batch.repeat.RepeatContext;
/**
* Default implementation of (@link JobExecutor} interface. Sequentially
@@ -132,11 +131,6 @@ public class DefaultJobExecutor implements JobExecutor {
job.setStatus(status);
jobRepository.update(job);
jobRepository.saveOrUpdate(jobExecution);
for (Iterator iter = jobExecution.getStepContexts().iterator(); iter
.hasNext();) {
RepeatContext context = (RepeatContext) iter.next();
context.setAttribute("JOB_STATUS", status);
}
}
/*

52
spring-batch-execution/src/main/java/org/springframework/batch/execution/launch/DefaultJobLauncher.java
View File

@@ -24,27 +24,27 @@ import org.springframework.batch.core.domain.JobExecution;
import org.springframework.batch.core.domain.JobIdentifier;
import org.springframework.batch.core.domain.JobLocator;
import org.springframework.batch.core.domain.NoSuchJobException;
import org.springframework.batch.core.domain.StepExecution;
import org.springframework.batch.core.executor.JobExecutor;
import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.batch.repeat.RepeatContext;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
/**
* A test implementation of the JobLauncher interface. It exists
* solely to work through interface design issues for the JobLauncher,
* JobRepository, JobLocator, and JobExecutor interfaces. It is designed
* for simplicity, and despite unit testing may not be completely threadsafe,
* and therefore should not be used.
* A test implementation of the JobLauncher interface. It exists solely to work
* through interface design issues for the JobLauncher, JobRepository,
* JobLocator, and JobExecutor interfaces. It is designed for simplicity, and
* despite unit testing may not be completely threadsafe, and therefore should
* not be used.
*
* Rather than using a JobExecutorFacade, a JobExecutor is worked with directly.
* Not every method of the JobLauncher interface is used. Instead, new versions
* that take JobIdentifier as an argument were added. A JobExecution is considered
* to be running if it's JobIdentifier (the one it was ran with) exists in the
* HashMap execution registry. When a JobExecutor is finished processing it removes
* it's identifier from the map.
* Not every method of the JobLauncher interface is used. Instead, new versions
* that take JobIdentifier as an argument were added. A JobExecution is
* considered to be running if it's JobIdentifier (the one it was ran with)
* exists in the HashMap execution registry. When a JobExecutor is finished
* processing it removes it's identifier from the map.
*
* @author Lucas Ward
*
@@ -77,15 +77,13 @@ public class DefaultJobLauncher implements JobLauncher {
*
* @see org.springframework.batch.execution.launch.JobLauncher#run(org.springframework.batch.core.domain.JobIdentifier)
*/
public JobExecution run(JobIdentifier jobIdentifier)
throws NoSuchJobException, JobExecutionAlreadyRunningException {
public JobExecution run(JobIdentifier jobIdentifier) throws NoSuchJobException, JobExecutionAlreadyRunningException {
JobExecution jobExecution;
synchronized (monitor) {
if (jobExecutionRegistry.containsKey(jobIdentifier)) {
throw new JobExecutionAlreadyRunningException("Job: "
+ jobIdentifier + "is already running.");
throw new JobExecutionAlreadyRunningException("Job: " + jobIdentifier + "is already running.");
}
Job job = jobLocator.getJob(jobIdentifier.getName());
@@ -105,7 +103,7 @@ public class DefaultJobLauncher implements JobLauncher {
public void run() {
ExitStatus status = jobExecutor.run(job, jobExecution);
jobExecution.setExitStatus(status);
synchronized(monitor){
synchronized (monitor) {
jobExecutionRegistry.remove(jobIdentifier);
}
}
@@ -131,26 +129,22 @@ public class DefaultJobLauncher implements JobLauncher {
JobExecution jobExecution = (JobExecution)jobExecutionRegistry.get(jobIdentifier);
for (Iterator iter = jobExecution.getStepContexts().iterator(); iter
for (Iterator iter = jobExecution.getStepExecutions().iterator(); iter
.hasNext();) {
RepeatContext context = (RepeatContext) iter.next();
StepExecution context = (StepExecution) iter.next();
context.setTerminateOnly();
}
for (Iterator iter = jobExecution.getChunkContexts().iterator(); iter
.hasNext();) {
RepeatContext context = (RepeatContext) iter.next();
context.setTerminateOnly();
}
}
}
public boolean isRunning(JobIdentifier jobIdentifier){
synchronized(monitor){
if(jobExecutionRegistry.containsKey(jobIdentifier)){
public boolean isRunning(JobIdentifier jobIdentifier) {
synchronized (monitor) {
if (jobExecutionRegistry.containsKey(jobIdentifier)) {
return true;
}
else{
else {
return false;
}
}

23
spring-batch-execution/src/main/java/org/springframework/batch/execution/launch/SimpleJobExecutorFacade.java
View File

@@ -29,6 +29,7 @@ import org.springframework.batch.core.domain.JobExecution;
import org.springframework.batch.core.domain.JobIdentifier;
import org.springframework.batch.core.domain.JobLocator;
import org.springframework.batch.core.domain.NoSuchJobException;
import org.springframework.batch.core.domain.StepExecution;
import org.springframework.batch.core.executor.JobExecutor;
import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException;
import org.springframework.batch.core.repository.JobRepository;
@@ -272,14 +273,9 @@ class SimpleJobExecutorFacade implements JobExecutorFacade,
"The job is not executing in this executor: [" + execution
+ "]");
}
for (Iterator iter = execution.getStepContexts().iterator(); iter
for (Iterator iter = execution.getStepExecutions().iterator(); iter
.hasNext();) {
RepeatContext context = (RepeatContext) iter.next();
context.setTerminateOnly();
}
for (Iterator iter = execution.getChunkContexts().iterator(); iter
.hasNext();) {
RepeatContext context = (RepeatContext) iter.next();
StepExecution context = (StepExecution) iter.next();
context.setTerminateOnly();
}
this.onStop(execution);
@@ -301,20 +297,11 @@ class SimpleJobExecutorFacade implements JobExecutorFacade,
String runtime = "job" + i;
props.setProperty(runtime, "" + element.getJobInstance().getIdentifier());
int j = 0;
for (Iterator iterator = element.getStepContexts().iterator(); iterator
for (Iterator iterator = element.getStepExecutions().iterator(); iterator
.hasNext();) {
RepeatContext context = (RepeatContext) iterator.next();
StepExecution context = (StepExecution) iterator.next();
j++;
props.setProperty(runtime + ".step" + j, "" + context);
}
j = 0;
for (Iterator iterator = element.getChunkContexts().iterator(); iterator
.hasNext();) {
RepeatContext context = (RepeatContext) iterator.next();
j++;
props.setProperty(runtime + ".chunk" + j, "" + context);
}
}
return props;

13
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java
View File

@@ -17,7 +17,6 @@
package org.springframework.batch.execution.step.simple;
import java.util.Date;
import java.util.Iterator;
import java.util.Properties;
import org.springframework.batch.core.domain.BatchStatus;
@@ -175,7 +174,6 @@ public class SimpleStepExecutor implements StepExecutor {
public ExitStatus doInIteration(final RepeatContext context) throws Exception {
final StepContribution contribution = stepExecution.createStepContribution();
contribution.registerStepContext(context);
// Before starting a new transaction, check for
// interruption.
@@ -295,10 +293,6 @@ public class SimpleStepExecutor implements StepExecutor {
step.setStatus(status);
jobRepository.update(step);
jobRepository.saveOrUpdate(stepExecution);
for (Iterator iter = stepExecution.getJobExecution().getStepContexts().iterator(); iter.hasNext();) {
RepeatContext context = (RepeatContext) iter.next();
context.setAttribute("JOB_STATUS", status);
}
}
/**
@@ -307,15 +301,16 @@ public class SimpleStepExecutor implements StepExecutor {
* outside this method, so subclasses that override do not need to create a
* transaction.
*
* @param step the current step
* @param stepExecution the current step, containing the {@link Tasklet}
* @param step the current step containing the {@link Tasklet}
* with the business logic.
* @return true if there is more data to process.
*/
protected final ExitStatus processChunk(final Step step, final StepContribution contribution) {
ExitStatus result = chunkOperations.iterate(new RepeatCallback() {
public ExitStatus doInIteration(final RepeatContext context) throws Exception {
contribution.registerChunkContext(context);
if (contribution.isTerminateOnly()) {
context.setTerminateOnly();
}
// check for interruption before each item as well
interruptionPolicy.checkInterrupted(context);
ExitStatus exitStatus = doTaskletProcessing(step.getTasklet(), contribution);

12
spring-batch-execution/src/test/java/org/springframework/batch/execution/launch/DefaultJobLauncherTests.java
View File

@@ -13,14 +13,14 @@ import org.springframework.batch.core.domain.Job;
import org.springframework.batch.core.domain.JobExecution;
import org.springframework.batch.core.domain.JobIdentifier;
import org.springframework.batch.core.domain.JobLocator;
import org.springframework.batch.core.domain.StepExecution;
import org.springframework.batch.core.domain.StepInstance;
import org.springframework.batch.core.executor.JobExecutor;
import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.runtime.SimpleJobIdentifier;
import org.springframework.batch.io.exception.BatchCriticalException;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.batch.repeat.RepeatContext;
import org.springframework.batch.repeat.context.RepeatContextSupport;
/**
* @author Lucas Ward
@@ -124,8 +124,7 @@ public class DefaultJobLauncherTests extends TestCase {
public void testStop() throws Exception{
jobLauncher.setJobExecutor(blockingExecutor);
RepeatContext stepContext = new RepeatContextSupport(null);
jobExecution.registerStepContext(stepContext);
jobExecution.createStepExecution(new StepInstance(null, "step"));
jobLocator.getJob("job");
locatorControl.setDefaultReturnValue(job);
@@ -141,10 +140,9 @@ public class DefaultJobLauncherTests extends TestCase {
jobLauncher.stop(jobIdentifier);
Collection contexts = jobExecution.getStepContexts();
Collection contexts = jobExecution.getStepExecutions();
for(Iterator it = contexts.iterator();it.hasNext();){
RepeatContext context = (RepeatContext)it.next();
assertEquals(stepContext, context);
StepExecution context = (StepExecution)it.next();
assertTrue(context.isTerminateOnly());
}
}

110
spring-batch-execution/src/test/java/org/springframework/batch/execution/launch/SimpleJobExecutorFacadeTests.java
View File

@@ -30,13 +30,14 @@ import org.springframework.batch.core.domain.JobExecution;
import org.springframework.batch.core.domain.JobInstance;
import org.springframework.batch.core.domain.JobLocator;
import org.springframework.batch.core.domain.NoSuchJobException;
import org.springframework.batch.core.domain.StepExecution;
import org.springframework.batch.core.domain.StepInstance;
import org.springframework.batch.core.executor.JobExecutor;
import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.runtime.SimpleJobIdentifier;
import org.springframework.batch.io.exception.BatchCriticalException;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.batch.repeat.context.RepeatContextSupport;
import org.springframework.util.ReflectionUtils;
/**
@@ -53,8 +54,7 @@ public class SimpleJobExecutorFacadeTests extends TestCase {
private JobRepository jobRepository;
private MockControl jobRepositoryControl = MockControl
.createControl(JobRepository.class);
private MockControl jobRepositoryControl = MockControl.createControl(JobRepository.class);
private Job jobConfiguration = new Job();
@@ -62,8 +62,7 @@ public class SimpleJobExecutorFacadeTests extends TestCase {
private SimpleJobIdentifier jobIdentifier = new SimpleJobIdentifier("TestJob");
private JobExecution jobExecution = new JobExecution(new JobInstance(
jobIdentifier, null));
private JobExecution jobExecution = new JobExecution(new JobInstance(jobIdentifier, null));
private List list = new ArrayList();
@@ -96,12 +95,10 @@ public class SimpleJobExecutorFacadeTests extends TestCase {
}
private JobInstance setUpFacadeForNormalStart()
throws Exception {
private JobInstance setUpFacadeForNormalStart() throws Exception {
jobIdentifier = new SimpleJobIdentifier("bar");
jobExecutor = new JobExecutor() {
public ExitStatus run(Job configuration,
JobExecution execution) throws BatchCriticalException {
public ExitStatus run(Job configuration, JobExecution execution) throws BatchCriticalException {
jobExecution = execution;
return ExitStatus.FINISHED;
}
@@ -112,45 +109,41 @@ public class SimpleJobExecutorFacadeTests extends TestCase {
jobRepository.findOrCreateJob(jobConfiguration, jobIdentifier);
jobRepositoryControl.setReturnValue(jobExecution);
jobRepositoryControl.replay();
jobExecutorFacade
.setJobLocator(new JobLocator() {
public Job getJob(String name)
throws NoSuchJobException {
return jobConfiguration;
}
});
jobExecutorFacade.setJobLocator(new JobLocator() {
public Job getJob(String name) throws NoSuchJobException {
return jobConfiguration;
}
});
return job;
}
public void testIsRunning() throws Exception {
jobExecutorFacade.setJobExecutor(new JobExecutor() {
public ExitStatus run(Job configuration,
JobExecution execution) throws BatchCriticalException {
public ExitStatus run(Job 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
.setJobLocator(new JobLocator() {
public Job getJob(String name)
throws NoSuchJobException {
return jobConfiguration;
}
});
jobExecutorFacade.setJobLocator(new JobLocator() {
public Job getJob(String name) throws NoSuchJobException {
return jobConfiguration;
}
});
running = true;
new Thread(new Runnable() {
public void run() {
try {
jobExecutorFacade.start(jobExecution);
} catch (NoSuchJobException e) {
}
catch (NoSuchJobException e) {
throw new IllegalStateException("Shouldn't happen");
}
}
@@ -173,30 +166,31 @@ public class SimpleJobExecutorFacadeTests extends TestCase {
try {
jobExecutorFacade.afterPropertiesSet();
fail("Expected IllegalStateException");
} catch (IllegalArgumentException ex) {
}
catch (IllegalArgumentException ex) {
// expected
}
}
public void testStopWithNoJob() 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)));
try {
jobExecutorFacade.stop(execution);
fail("Expected NoSuchJobExecutionException");
} catch (NoSuchJobExecutionException e) {
}
catch (NoSuchJobExecutionException e) {
// expected
assertTrue("Wrong message in exception: "+e.getMessage(), e.getMessage().indexOf("TestJob") >= 0);
assertTrue("Wrong message in exception: " + e.getMessage(), 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");
JobInstance jobInstance = new JobInstance(runtimeInformation, new Long(0));
JobExecution execution = new JobExecution(jobInstance);
StepExecution stepExecution = execution.createStepExecution(new StepInstance(jobInstance, "step"));
List listeners = new ArrayList();
listeners.add(new JobExecutionListenerSupport() {
@@ -208,15 +202,9 @@ public class SimpleJobExecutorFacadeTests extends TestCase {
registerExecution(runtimeInformation, execution);
RepeatContextSupport stepContext = new RepeatContextSupport(null);
RepeatContextSupport chunkContext = new RepeatContextSupport(
stepContext);
execution.registerStepContext(stepContext);
execution.registerChunkContext(chunkContext);
jobExecutorFacade.stop(execution);
assertTrue(stepContext.isCompleteOnly());
assertTrue(chunkContext.isCompleteOnly());
assertTrue(stepExecution.isTerminateOnly());
assertEquals(1, list.size());
}
@@ -225,29 +213,27 @@ 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");
JobInstance jobInstance = new JobInstance(runtimeInformation, new Long(0));
JobExecution execution = new JobExecution(jobInstance);
registerExecution(runtimeInformation, execution);
execution.registerStepContext(new RepeatContextSupport(null));
execution.createStepExecution(new StepInstance(jobInstance, "step"));
Properties statistics = jobExecutorFacade.getStatistics();
assertNotNull(statistics);
assertTrue(statistics.containsKey("job1.step1"));
}
public void testJobAlreadyExecutingLocally() 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);
try {
jobExecutorFacade.createExecutionFrom(runtimeInformation);
fail("Expected JobExecutionAlreadyRunningException");
} catch (JobExecutionAlreadyRunningException e) {
}
catch (JobExecutionAlreadyRunningException e) {
// expected
assertTrue("Message does not contain TestJob: "+e.getMessage(), e.getMessage().indexOf("TestJob")>=0);
assertTrue("Message does not contain TestJob: " + e.getMessage(), e.getMessage().indexOf("TestJob") >= 0);
}
}
@@ -305,11 +291,9 @@ public class SimpleJobExecutorFacadeTests extends TestCase {
assertEquals("two", list.get(1));
}
private void registerExecution(SimpleJobIdentifier runtimeInformation,
JobExecution execution) throws NoSuchFieldException,
IllegalAccessException {
Field field = SimpleJobExecutorFacade.class
.getDeclaredField("jobExecutionRegistry");
private void registerExecution(SimpleJobIdentifier runtimeInformation, JobExecution execution)
throws NoSuchFieldException, IllegalAccessException {
Field field = SimpleJobExecutorFacade.class.getDeclaredField("jobExecutionRegistry");
ReflectionUtils.makeAccessible(field);
Map map = (Map) field.get(jobExecutorFacade);
map.put(runtimeInformation, execution);

2
spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/DefaultStepExecutorTests.java
View File

@@ -159,8 +159,6 @@ public class DefaultStepExecutorTests extends TestCase {
stepConfiguration.setTasklet(new Tasklet() {
public ExitStatus execute() throws Exception {
assertEquals(step, stepExecution.getStep());
assertEquals(1, jobExecution.getChunkContexts().size());
assertEquals(1, jobExecution.getStepContexts().size());
assertNotNull(StepSynchronizationManager.getContext()
.getStepExecution());
processed.add("foo");