From f49c9d9526988f2e2540d39c0af2976314601c51 Mon Sep 17 00:00:00 2001 From: dsyer Date: Sat, 15 Dec 2007 15:25:11 +0000 Subject: [PATCH] IN PROGRESS - issue BATCH-127: Allow job configuration to control re-entrant behavior http://opensource.atlassian.com/projects/spring/browse/BATCH-127 SimpleJobRepository.findOrCreateJob now blocks if the transaction isolation level is high enough (see Javadocs). --- .../batch/core/domain/JobExecution.java | 2 +- spring-batch-execution/.project | 45 +++---- spring-batch-execution/.springBeans | 1 + .../repository/SimpleJobRepository.java | 113 ++++++++++++------ spring-batch-samples/.springBeans | 20 +++- .../resources/alt-data-source-context.xml | 24 ++++ .../src/main/resources/alt.properties | 3 + .../main/resources/data-source-context.xml | 5 +- .../resources/simple-container-definition.xml | 14 +-- .../sample/FootballJobFunctionalTests.java | 2 +- .../src/test/resources/log4j.properties | 12 +- 11 files changed, 171 insertions(+), 70 deletions(-) create mode 100644 spring-batch-samples/src/main/resources/alt-data-source-context.xml create mode 100644 spring-batch-samples/src/main/resources/alt.properties diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/domain/JobExecution.java b/spring-batch-core/src/main/java/org/springframework/batch/core/domain/JobExecution.java index 98fc21c6d..7c9c1d42e 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/domain/JobExecution.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/domain/JobExecution.java @@ -232,7 +232,7 @@ public class JobExecution extends Entity { * @see org.springframework.batch.core.domain.Entity#toString() */ public String toString() { - return super.toString()+", job=["+job+"]"; + return super.toString()+", startTime="+startTime+", endTime="+endTime+", job=["+job+"]"; } /** diff --git a/spring-batch-execution/.project b/spring-batch-execution/.project index 05226aa73..389dc050a 100644 --- a/spring-batch-execution/.project +++ b/spring-batch-execution/.project @@ -1,20 +1,25 @@ - - spring-batch-execution - Execution tools and implementations of Spring Batch Core interfaces - - spring-batch-core - spring-batch-infrastructure - - - - org.eclipse.jdt.core.javabuilder - - - org.springframework.ide.eclipse.core.springbuilder - - - - org.eclipse.jdt.core.javanature - org.springframework.ide.eclipse.core.springnature - - \ No newline at end of file + + + spring-batch-execution + Execution tools and implementations of Spring Batch Core interfaces + + spring-batch-core + spring-batch-infrastructure + + + + org.eclipse.jdt.core.javabuilder + + + + + org.springframework.ide.eclipse.core.springbuilder + + + + + + org.springframework.ide.eclipse.core.springnature + org.eclipse.jdt.core.javanature + + diff --git a/spring-batch-execution/.springBeans b/spring-batch-execution/.springBeans index a78899bd3..347d9c4a7 100644 --- a/spring-batch-execution/.springBeans +++ b/spring-batch-execution/.springBeans @@ -14,6 +14,7 @@ src/test/resources/org/springframework/batch/execution/bootstrap/support/test-batch-environment-no-launcher.xml src/test/resources/org/springframework/batch/execution/configuration/test-context.xml src/test/resources/org/springframework/batch/execution/scope/scope-tests.xml + src/test/resources/beanRefContext.xml diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/SimpleJobRepository.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/SimpleJobRepository.java index afdb0192e..86da900c4 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/SimpleJobRepository.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/SimpleJobRepository.java @@ -34,6 +34,7 @@ import org.springframework.batch.core.repository.JobRepository; import org.springframework.batch.execution.repository.dao.JobDao; import org.springframework.batch.execution.repository.dao.StepDao; import org.springframework.batch.restart.GenericRestartData; +import org.springframework.transaction.annotation.Isolation; import org.springframework.util.Assert; /** @@ -64,49 +65,96 @@ public class SimpleJobRepository implements JobRepository { /** *

- * Find or Create a JobInstance(@link JobInstance) based on the passed in - * RuntimeInformation and Configuration. JobRuntimeInformation contains the - * following fields which logically identify a job: JobName, JobStream, - * JobRun, and Schedule Date. However, unique identification of a job can - * only come from the database, and therefore must come from JobDao by - * either creating a new job or finding an existing one, which will ensure - * that the id field of the job is populated with the correct value. + * Find or Create a (@link {@link JobExecution}) based on the passed in + * {@link JobIdentifier} and {@link JobConfiguration}. However, unique + * identification of a job can only come from the database, and therefore + * must come from JobDao by either creating a new job or finding an existing + * one, which will ensure that the id of the job is populated with the + * correct value. *

* *

* There are two ways in which the method determines if a job should be * created or an existing one should be returned. The first is - * restartability. The Job's restartPolicy will be checked first. If it is - * not restartable, a new job will be created, regardless of whether or not - * one exists. If it is restartable, the JobDao will be checked to determine - * if the job already exists, if it does, it's steps will be populated - * (there must be at least 1) and it will be returned. If no job is found, a - * new one will be created based on the configuration. + * restartability. The {@link JobConfiguration} restartable property will be + * checked first. If it is not false, a new job will be created, regardless + * of whether or not one exists. If it is true, the {@link JobDao} will be + * checked to determine if the job already exists, if it does, it's steps + * will be populated (there must be at least 1) and a new + * {@link JobExecution} will be returned. If no job is found, a new one will + * be created based on the configuration. + *

+ * + *

+ * A check is made to see if any job executions are already running, and an + * exception will be thrown if one is detected. To detect a running job + * execution we use the {@link JobDao}: + *

    + *
  1. First we find all jobs which match the given {@link JobIdentifier}
    2. + *
  2. What happens then depends on how many existing job instances we + * find: + *
      + *
    • If there are none, or the {@link JobConfiguration} is marked + * restartable, then we create a new {@link JobInstance}
      • + *
    • If there is more than one and the {@link JobConfiguration} is not + * marked as restartable, it is an error. This could be caused by a job + * whose restartable flag has changed to be more strict (true not false) + * after it has been executed at least once.
      • + *
    • If there is precisely one existing {@link JobInstance} then we check + * the {@link JobExecution} instances for that job, and if any of them tells + * us it is running (see {@link JobExecution#isRunning()}) then it is an + * error.
      • + *
    + *
    3. + *
+ * If this method is run in a transaction (as it normally would be) with + * isolation level at {@link Isolation#REPEATABLE_READ} or better, then this + * method should block if another transaction is already executing this + * method for the same {@link JobIdentifier}. The first transaction to + * complete in this scenario should obtain a valid {@link JobExecution}, + * and others will throw {@link JobExecutionAlreadyRunningException} (or + * timeout). There are no such guarantees if the {@link JobDao} does not + * respect the transaction isolation levels (e.g. if using a non-relational + * data-store, or if the platform does not support the higher isolation + * levels). *

- * @throws JobExecutionAlreadyRunningException * * @see JobRepository#findOrCreateJob(JobConfiguration, JobIdentifier) + * * @throws BatchRestartException * if more than one JobInstance if found or if * JobInstance.getJobExecutionCount() is greater than * JobConfiguration.getStartLimit() + * @throws JobExecutionAlreadyRunningException + * if a job execution is found for the given + * {@link JobIdentifier} that is already running + * */ public JobExecution findOrCreateJob(JobConfiguration jobConfiguration, - JobIdentifier runtimeInformation) throws JobExecutionAlreadyRunningException { + JobIdentifier runtimeInformation) + throws JobExecutionAlreadyRunningException { - List jobs; + List jobs = new ArrayList(); + JobInstance job; // Check if a job is restartable, if not, create and return a new job - if (jobConfiguration.isRestartable() == false) { - return createJob(jobConfiguration, runtimeInformation); - } else { - // find all jobs matching the runtime information. + if (jobConfiguration.isRestartable()) { + + /* + * Find all jobs matching the runtime information. + * + * Always do this if the job is restartable, then if this method is + * transactional, and the isolation level is REPEATABLE_READ or + * better, another launcher trying to start the same job in another + * thread or process will block until this transaction has finished. + */ + jobs = jobDao.findJobs(runtimeInformation); } if (jobs.size() == 1) { // One job was found - JobInstance job = (JobInstance) jobs.get(0); + job = (JobInstance) jobs.get(0); job.setSteps(findSteps(jobConfiguration.getStepConfigurations(), job)); job.setJobExecutionCount(jobDao.getJobExecutionCount(job.getId())); @@ -118,27 +166,23 @@ public class SimpleJobRepository implements JobRepository { for (Iterator iterator = executions.iterator(); iterator.hasNext();) { JobExecution execution = (JobExecution) iterator.next(); if (execution.isRunning()) { - throw new JobExecutionAlreadyRunningException("A job execution for this job is already running: "+job); + throw new JobExecutionAlreadyRunningException( + "A job execution for this job is already running: " + + job); } } - /* - * Update the job, then if this method is transactional, and the - * isolation level is SERIALIZABLE, another launcher trying to start - * the same job in another thread or process will lose. - */ - jobDao.update(job); - - return generateJobExecution(job); - } else if (jobs.size() == 0) { // no job found, create one - return createJob(jobConfiguration, runtimeInformation); + job = createJob(jobConfiguration, runtimeInformation); } else { // More than one job found, throw exception throw new BatchRestartException( "Error restarting job, more than one JobInstance found for: " + jobConfiguration.toString()); } + + return generateJobExecution(job); + } private JobExecution generateJobExecution(JobInstance job) { @@ -251,15 +295,14 @@ public class SimpleJobRepository implements JobRepository { * calling {@link JobDao#createJob(JobRuntimeInformation)} and then it's * list of StepConfigurations is passed to the createSteps method. */ - private JobExecution createJob(JobConfiguration jobConfiguration, + private JobInstance createJob(JobConfiguration jobConfiguration, JobIdentifier runtimeInformation) { JobInstance job = jobDao.createJob(runtimeInformation); job .setSteps(createSteps(job, jobConfiguration .getStepConfigurations())); - JobExecution execution = generateJobExecution(job); - return execution; + return job; } /* diff --git a/spring-batch-samples/.springBeans b/spring-batch-samples/.springBeans index 5453e6625..6e45a6b41 100644 --- a/spring-batch-samples/.springBeans +++ b/spring-batch-samples/.springBeans @@ -1,8 +1,10 @@ - - xml - + 1 + + + + src/main/resources/jobs/fixedLengthImportJob.xml src/main/resources/jobs/multilineJob.xml @@ -27,6 +29,9 @@ src/main/resources/jobs/simpleTaskletJob.xml src/main/resources/beanRefContext.xml src/main/resources/jobs/delegatingJob.xml + src/main/resources/jobs/parallelJob.xml + src/main/resources/jobs/rollbackJob.xml + src/test/resources/org/springframework/batch/sample/item/processor/staging-test-context.xml @@ -162,5 +167,14 @@ src/main/resources/simple-container-definition.xml + + + true + false + + src/main/resources/data-source-context.xml + src/main/resources/simple-container-definition.xml + + diff --git a/spring-batch-samples/src/main/resources/alt-data-source-context.xml b/spring-batch-samples/src/main/resources/alt-data-source-context.xml new file mode 100644 index 000000000..d1cc53ac4 --- /dev/null +++ b/spring-batch-samples/src/main/resources/alt-data-source-context.xml @@ -0,0 +1,24 @@ + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/spring-batch-samples/src/main/resources/alt.properties b/spring-batch-samples/src/main/resources/alt.properties new file mode 100644 index 000000000..9d825e9e1 --- /dev/null +++ b/spring-batch-samples/src/main/resources/alt.properties @@ -0,0 +1,3 @@ +batch.schema.script=schema-derby.sql +batch.business.schema.script=business-schema-derby.sql +batch.database.incrementer.class=org.springframework.jdbc.support.incrementer.DerbyMaxValueIncrementer diff --git a/spring-batch-samples/src/main/resources/data-source-context.xml b/spring-batch-samples/src/main/resources/data-source-context.xml index a4b3e627f..73f9cfe94 100644 --- a/spring-batch-samples/src/main/resources/data-source-context.xml +++ b/spring-batch-samples/src/main/resources/data-source-context.xml @@ -14,9 +14,8 @@ - - + @@ -29,11 +28,13 @@ + + diff --git a/spring-batch-samples/src/main/resources/simple-container-definition.xml b/spring-batch-samples/src/main/resources/simple-container-definition.xml index 0ea28fdc0..641fd9450 100644 --- a/spring-batch-samples/src/main/resources/simple-container-definition.xml +++ b/spring-batch-samples/src/main/resources/simple-container-definition.xml @@ -38,7 +38,7 @@ advice-ref="txAdvice" /> - + @@ -58,17 +58,17 @@ - + - - + + - @@ -76,13 +76,13 @@ ref="jobExecutionIncrementer" /> - - +