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

RESOLVED - BATCH-999: JobExecution ExecutionContext should not be persisted by Step?

Browse Source 
javadoc
This commit is contained in:
robokaso
2009-01-20 14:34:49 +00:00
parent 9c9fe8f6ca
commit 82c106800d
4 changed files with 67 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files

78
spring-batch-core/src/main/java/org/springframework/batch/core/repository/JobRepository.java
View File

@@ -22,42 +22,64 @@ import org.springframework.batch.core.JobInstance;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.repository.dao.JobExecutionDao;
import org.springframework.batch.core.repository.dao.JobInstanceDao;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.transaction.annotation.Isolation;
/**
* <p>
* Repository for storing batch {@link JobExecution} and {@link StepExecution}s.
* Before using any methods, a {@link JobExecution} must first be obtained using
* the createJobExecution method. Once a {@link JobExecution} is obtained, they
* can be updated.
* Repository responsible for persistence of batch metadata entities.
* </p>
*
* @see JobInstance
* @see JobExecution
* @see StepExecution
*
* @author Lucas Ward
* @author Dave Syer
*
* @author Robert Kasanicky
*/
public interface JobRepository {
/**
* Check if an instance of this job already exists with the parameters provided.
* Check if an instance of this job already exists with the parameters
* provided.
*
* @param jobName the name of the job
* @param jobParameters the parameters to match
* @return true if a {@link JobInstance} already exists for this job name and job parameters
* @return true if a {@link JobInstance} already exists for this job name
* and job parameters
*/
boolean isJobInstanceExists(String jobName, JobParameters jobParameters);
/**
* Find or create a {@link JobExecution} for a given {@link Job} and
* {@link JobParameters}. If the {@link Job} was already executed with these
* {@link JobParameters}, its persisted values (including ID) will be
* returned in a new {@link JobInstance}, associated with the
* {@link JobExecution}. If no previous instance is found, the execution
* will be associated with a new {@link JobInstance}
* @param jobName the name of the job that is to be executed
* <p>
* Create a {@link JobExecution} for a given {@link Job} and
* {@link JobParameters}. If matching {@link JobInstance} already exists,
* the job must be restartable and it's last JobExecution must *not* be
* completed. If matching {@link JobInstance} does not exist yet it will be
* created.
* </p>
*
* <p>
* 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 it (for
* the same {@link JobParameters} and job name). The first transaction to
* complete in this scenario obtains a valid {@link JobExecution}, and
* others throw {@link JobExecutionAlreadyRunningException} (or timeout).
* There are no such guarantees if the {@link JobInstanceDao} and
* {@link JobExecutionDao} do 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).
* </p>
*
* @param jobName the name of the job that is to be executed </p>
*
* @param jobParameters the runtime parameters for the job
*
* @return a valid job {@link JobExecution} for the arguments provided
* @return a valid {@link JobExecution} for the arguments provided
* @throws JobExecutionAlreadyRunningException if there is a
* {@link JobExecution} already running for the job instance with the
* provided job and parameters.
@@ -68,11 +90,11 @@ public interface JobRepository {
* found and was already completed successfully.
*
*/
JobExecution createJobExecution(String jobName, JobParameters jobParameters) throws JobExecutionAlreadyRunningException,
JobRestartException, JobInstanceAlreadyCompleteException;
JobExecution createJobExecution(String jobName, JobParameters jobParameters)
throws JobExecutionAlreadyRunningException, JobRestartException, JobInstanceAlreadyCompleteException;
/**
* Update the {@link JobExecution}.
* Update the {@link JobExecution} (but not its {@link ExecutionContext}).
*
* Preconditions: {@link JobExecution} must contain a valid
* {@link JobInstance} and be saved (have an id assigned).
@@ -82,11 +104,10 @@ public interface JobRepository {
void update(JobExecution jobExecution);
/**
* Save the {@link StepExecution}. ID will be assigned - it is not permitted
* that an ID be assigned before calling this method. Instead, it should be
* left blank, to be assigned by a {@link JobRepository}. The
* {@link ExecutionContext} of the {@link StepExecution} is <em>not</em>
* saved: see {@link #updateExecutionContext(StepExecution)}.
* Save the {@link StepExecution} and its {@link ExecutionContext}. ID will
* be assigned - it is not permitted that an ID be assigned before calling
* this method. Instead, it should be left blank, to be assigned by a
* {@link JobRepository}.
*
* Preconditions: {@link StepExecution} must have a valid {@link Step}.
*
@@ -95,7 +116,7 @@ public interface JobRepository {
void add(StepExecution stepExecution);
/**
* Update the {@link StepExecution}.
* Update the {@link StepExecution} (but not its {@link ExecutionContext}).
*
* Preconditions: {@link StepExecution} must be saved (have an id assigned).
*
@@ -104,11 +125,10 @@ public interface JobRepository {
void update(StepExecution stepExecution);
/**
* Persist the {@link ExecutionContext} of the given {@link StepExecution}
* and enclosing {@link JobExecution}.
* Persist the updated {@link ExecutionContext}s of the given
* {@link StepExecution} and corresponding {@link JobExecution}.
*
* @param stepExecution the {@link StepExecution} containing the
* {@link ExecutionContext} to be saved.
* @param stepExecution
*/
void updateExecutionContext(StepExecution stepExecution);
@@ -125,7 +145,7 @@ public interface JobRepository {
int getStepExecutionCount(JobInstance jobInstance, String stepName);
/**
* @param jobName the name of the job that might have run
* @param jobName the name of the job that might have run
* @param jobParameters parameters identifying the {@link JobInstance}
* @return the last execution of job if exists, null otherwise
*/

23
spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JdbcExecutionContextDao.java
View File

@@ -70,10 +70,6 @@ public class JdbcExecutionContextDao extends AbstractJdbcBatchMetadataDao implem
private ExecutionContextStringSerializer serializer;
/**
* @param jobExecution
* @return execution context associated with the given jobExecution.
*/
public ExecutionContext getExecutionContext(JobExecution jobExecution) {
Long executionId = jobExecution.getId();
Assert.notNull(executionId, "ExecutionId must not be null.");
@@ -88,10 +84,6 @@ public class JdbcExecutionContextDao extends AbstractJdbcBatchMetadataDao implem
}
}
/**
* @param stepExecution
* @return execution context associated with the given stepExecution.
*/
public ExecutionContext getExecutionContext(StepExecution stepExecution) {
Long executionId = stepExecution.getId();
Assert.notNull(executionId, "ExecutionId must not be null.");
@@ -106,11 +98,6 @@ public class JdbcExecutionContextDao extends AbstractJdbcBatchMetadataDao implem
}
}
/**
* Persist or update the execution context associated with the given
* jobExecution
* @param jobExecution
*/
public void updateExecutionContext(final JobExecution jobExecution) {
Long executionId = jobExecution.getId();
ExecutionContext executionContext = jobExecution.getExecutionContext();
@@ -122,11 +109,6 @@ public class JdbcExecutionContextDao extends AbstractJdbcBatchMetadataDao implem
persistSerializedContext(executionId, serializedContext, UPDATE_JOB_EXECUTION_CONTEXT);
}
/**
* Persist or update the execution context associated with the given
* stepExecution
* @param stepExecution
*/
public void updateExecutionContext(final StepExecution stepExecution) {
Long executionId = stepExecution.getId();
@@ -173,6 +155,11 @@ public class JdbcExecutionContextDao extends AbstractJdbcBatchMetadataDao implem
((XStreamExecutionContextStringSerializer) serializer).afterPropertiesSet();
}
/**
* @param executionId
* @param serializedContext
* @param sql with parameters (shortContext, longContext, executionId)
*/
private void persistSerializedContext(final Long executionId, String serializedContext, String sql) {
final String shortContext;

20
spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapExecutionContextDao.java
View File

@@ -24,22 +24,29 @@ import org.springframework.batch.core.StepExecution;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.support.transaction.TransactionAwareProxyFactory;
/**
* In-memory implementation of {@link ExecutionContextDao} backed by static
* maps.
*
* @author Robert Kasanicky
*/
public class MapExecutionContextDao implements ExecutionContextDao {
private static Map<Long, ExecutionContext> contextsByStepExecutionId = TransactionAwareProxyFactory.createTransactionalMap();
private static Map<Long, ExecutionContext> contextsByStepExecutionId = TransactionAwareProxyFactory
.createTransactionalMap();
private static Map<Long, ExecutionContext> contextsByJobExecutionId = TransactionAwareProxyFactory
.createTransactionalMap();
private static Map<Long, ExecutionContext> contextsByJobExecutionId = TransactionAwareProxyFactory.createTransactionalMap();
public static void clear() {
contextsByJobExecutionId.clear();
contextsByStepExecutionId.clear();
}
private static ExecutionContext copy(ExecutionContext original) {
return (ExecutionContext) SerializationUtils.deserialize(SerializationUtils.serialize(original));
}
public ExecutionContext getExecutionContext(StepExecution stepExecution) {
return copy(contextsByStepExecutionId.get(stepExecution.getId()));
}
@@ -47,14 +54,13 @@ public class MapExecutionContextDao implements ExecutionContextDao {
public void updateExecutionContext(StepExecution stepExecution) {
contextsByStepExecutionId.put(stepExecution.getId(), copy(stepExecution.getExecutionContext()));
}
public ExecutionContext getExecutionContext(JobExecution jobExecution) {
return copy(contextsByJobExecutionId.get(jobExecution.getId()));
}
public void updateExecutionContext(JobExecution jobExecution) {
contextsByJobExecutionId.put(jobExecution.getId(), copy(jobExecution.getExecutionContext()));
}
public void saveExecutionContext(JobExecution jobExecution) {

87
spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/SimpleJobRepository.java
View File

@@ -21,7 +21,6 @@ import java.util.Date;
import java.util.List;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobInstance;
import org.springframework.batch.core.JobParameters;
@@ -35,7 +34,6 @@ import org.springframework.batch.core.repository.dao.JobExecutionDao;
import org.springframework.batch.core.repository.dao.JobInstanceDao;
import org.springframework.batch.core.repository.dao.StepExecutionDao;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.transaction.annotation.Isolation;
import org.springframework.util.Assert;
/**
@@ -81,71 +79,10 @@ public class SimpleJobRepository implements JobRepository {
this.ecDao = ecDao;
}
/**
* @see JobRepository#isJobInstanceExists(String, JobParameters)
*/
public boolean isJobInstanceExists(String jobName, JobParameters jobParameters) {
return jobInstanceDao.getJobInstance(jobName, jobParameters) != null;
}
/**
* <p>
* Create a {@link JobExecution} based on the passed in {@link Job} and
* {@link JobParameters}. However, unique identification of a job can only
* come from the database, and therefore must come from JobDao by either
* creating a new job instance or finding an existing one, which will ensure
* that the id of the job instance is populated with the correct value.
* </p>
*
* <p>
* 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 {@link Job} restartable property will be checked
* first. If it is false, a new job will be created, regardless of whether
* or not one exists. If it is true, the {@link JobInstanceDao} 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 instance is found, a new
* one will be created.
* </p>
*
* <p>
* 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 JobExecutionDao}:
* <ol>
* <li>First we find all jobs which match the given {@link JobParameters}
* and job name</li>
* <li>What happens then depends on how many existing job instances we find:
* <ul>
* <li>If there are none, or the {@link Job} is marked restartable, then we
* create a new {@link JobInstance}</li>
* <li>If there is more than one and the {@link Job} 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)
* <em>after</em> it has been executed at least once.</li>
* <li>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.</li>
* </ul>
* </li>
* </ol>
* 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 it (for
* the same {@link JobParameters} and job name). The first transaction to
* complete in this scenario obtains a valid {@link JobExecution}, and
* others throw {@link JobExecutionAlreadyRunningException} (or timeout).
* There are no such guarantees if the {@link JobInstanceDao} and
* {@link JobExecutionDao} do 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).
* </p>
*
* @see JobRepository#createJobExecution(String, JobParameters)
*
*/
public JobExecution createJobExecution(String jobName, JobParameters jobParameters)
throws JobExecutionAlreadyRunningException, JobRestartException, JobInstanceAlreadyCompleteException {
@@ -202,15 +139,6 @@ public class SimpleJobRepository implements JobRepository {
}
/**
* Update a JobExecution. A JobExecution is considered one 'execution' of a
* particular job. Therefore, it must have it's jobId field set before it is
* passed into this method. It also has it's own unique identifier, because
* it must be updatable separately.
*
* @param jobExecution to be stored.
* @throws IllegalArgumentException if jobExecution is null.
*/
public void update(JobExecution jobExecution) {
Assert.notNull(jobExecution, "JobExecution cannot be null.");
@@ -221,12 +149,6 @@ public class SimpleJobRepository implements JobRepository {
jobExecutionDao.updateJobExecution(jobExecution);
}
/**
* Save the {@link StepExecution}.
*
* Preconditions: step name must be given and associated
* {@link JobExecution} must already be saved (have an id assigned).
*/
public void add(StepExecution stepExecution) {
validateStepExecution(stepExecution);
@@ -235,12 +157,6 @@ public class SimpleJobRepository implements JobRepository {
ecDao.saveExecutionContext(stepExecution);
}
/**
* Update the {@link StepExecution}.
*
* Preconditions: step name must be given and associated
* {@link JobExecution} must already be saved (have an id assigned).
*/
public void update(StepExecution stepExecution) {
validateStepExecution(stepExecution);
Assert.notNull(stepExecution.getId(), "StepExecution must already be saved (have an id assigned)");
@@ -261,9 +177,6 @@ public class SimpleJobRepository implements JobRepository {
ecDao.updateExecutionContext(stepExecution);
}
/**
* @return the last execution of the step within given job instance
*/
public StepExecution getLastStepExecution(JobInstance jobInstance, String stepName) {
List<JobExecution> jobExecutions = jobExecutionDao.findJobExecutions(jobInstance);
List<StepExecution> stepExecutions = new ArrayList<StepExecution>(jobExecutions.size());