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

RESOLVED - BATCH-329 All daos now increment the version field on entity save/update

Browse Source
This commit is contained in:
robokaso
2008-03-08 20:58:27 +00:00
parent ed6f991a76
commit 5f9350e1c1
9 changed files with 166 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -39,12 +39,12 @@ public class JdbcJobExecutionDao extends AbstractJdbcBatchMetadataDao implements
+ "where JOB_INSTANCE_ID = ?";
private static final String SAVE_JOB_EXECUTION = "INSERT into %PREFIX%JOB_EXECUTION(JOB_EXECUTION_ID, JOB_INSTANCE_ID, START_TIME, "
+ "END_TIME, STATUS, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE) values (?, ?, ?, ?, ?, ?, ?, ?)";
+ "END_TIME, STATUS, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE, VERSION) values (?, ?, ?, ?, ?, ?, ?, ?, ?)";
private static final String CHECK_JOB_EXECUTION_EXISTS = "SELECT COUNT(*) FROM %PREFIX%JOB_EXECUTION WHERE JOB_EXECUTION_ID = ?";
private static final String UPDATE_JOB_EXECUTION = "UPDATE %PREFIX%JOB_EXECUTION set START_TIME = ?, END_TIME = ?, "
+ " STATUS = ?, CONTINUABLE = ?, EXIT_CODE = ?, EXIT_MESSAGE = ? where JOB_EXECUTION_ID = ?";
+ " STATUS = ?, CONTINUABLE = ?, EXIT_CODE = ?, EXIT_MESSAGE = ?, VERSION = ? where JOB_EXECUTION_ID = ?";
private static final String FIND_JOB_EXECUTIONS = "SELECT JOB_EXECUTION_ID, START_TIME, END_TIME, STATUS, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE from %PREFIX%JOB_EXECUTION"
+ " where JOB_INSTANCE_ID = ?";
@@ -90,16 +90,18 @@ public class JdbcJobExecutionDao extends AbstractJdbcBatchMetadataDao implements
validateJobExecution(jobExecution);
jobExecution.incrementVersion();
jobExecution.setId(new Long(jobExecutionIncrementer.nextLongValue()));
Object[] parameters = new Object[] { jobExecution.getId(), jobExecution.getJobId(),
jobExecution.getStartTime(), jobExecution.getEndTime(), jobExecution.getStatus().toString(),
jobExecution.getExitStatus().isContinuable() ? "Y" : "N", jobExecution.getExitStatus().getExitCode(),
jobExecution.getExitStatus().getExitDescription() };
jobExecution.getExitStatus().getExitDescription(), jobExecution.getVersion() };
getJdbcTemplate().update(
getQuery(SAVE_JOB_EXECUTION),
parameters,
new int[] { Types.INTEGER, Types.INTEGER, Types.TIMESTAMP, Types.TIMESTAMP, Types.VARCHAR, Types.CHAR,
Types.VARCHAR, Types.VARCHAR });
Types.VARCHAR, Types.VARCHAR, Types.INTEGER });
}
/**
@@ -128,6 +130,8 @@ public class JdbcJobExecutionDao extends AbstractJdbcBatchMetadataDao implements
validateJobExecution(jobExecution);
jobExecution.incrementVersion();
String exitDescription = jobExecution.getExitStatus().getExitDescription();
if (exitDescription != null && exitDescription.length() > EXIT_MESSAGE_LENGTH) {
exitDescription = exitDescription.substring(0, EXIT_MESSAGE_LENGTH);
@@ -135,7 +139,8 @@ public class JdbcJobExecutionDao extends AbstractJdbcBatchMetadataDao implements
}
Object[] parameters = new Object[] { jobExecution.getStartTime(), jobExecution.getEndTime(),
jobExecution.getStatus().toString(), jobExecution.getExitStatus().isContinuable() ? "Y" : "N",
jobExecution.getExitStatus().getExitCode(), exitDescription, jobExecution.getId() };
jobExecution.getExitStatus().getExitCode(), exitDescription, jobExecution.getVersion(),
jobExecution.getId() };
if (jobExecution.getId() == null) {
throw new IllegalArgumentException("JobExecution ID cannot be null. JobExecution must be saved "
@@ -146,15 +151,14 @@ public class JdbcJobExecutionDao extends AbstractJdbcBatchMetadataDao implements
// is invalid and
// an exception should be thrown.
if (getJdbcTemplate().queryForInt(getQuery(CHECK_JOB_EXECUTION_EXISTS), new Object[] { jobExecution.getId() }) != 1) {
throw new NoSuchObjectException("Invalid JobExecution, ID " + jobExecution.getId()
+ " not found.");
throw new NoSuchObjectException("Invalid JobExecution, ID " + jobExecution.getId() + " not found.");
}
getJdbcTemplate().update(
getQuery(UPDATE_JOB_EXECUTION),
parameters,
new int[] { Types.TIMESTAMP, Types.TIMESTAMP, Types.VARCHAR, Types.CHAR, Types.VARCHAR, Types.VARCHAR,
Types.INTEGER });
Types.INTEGER, Types.INTEGER });
}
/**

16
spring-batch-core/src/main/java/org/springframework/batch/execution/repository/dao/JdbcJobInstanceDao.java
View File

@@ -32,8 +32,8 @@ import org.springframework.util.Assert;
*/
public class JdbcJobInstanceDao extends AbstractJdbcBatchMetadataDao implements JobInstanceDao, InitializingBean {
private static final String CREATE_JOB_INSTANCE = "INSERT into %PREFIX%JOB_INSTANCE(JOB_INSTANCE_ID, JOB_NAME, JOB_KEY)"
+ " values (?, ?, ?)";
private static final String CREATE_JOB_INSTANCE = "INSERT into %PREFIX%JOB_INSTANCE(JOB_INSTANCE_ID, JOB_NAME, JOB_KEY, VERSION)"
+ " values (?, ?, ?, ?)";
private static final String CREATE_JOB_PARAMETERS = "INSERT into %PREFIX%JOB_PARAMS(JOB_INSTANCE_ID, KEY_NAME, TYPE_CD, "
+ "STRING_VAL, DATE_VAL, LONG_VAL, DOUBLE_VAL) values (?, ?, ?, ?, ?, ?, ?)";
@@ -60,21 +60,23 @@ public class JdbcJobInstanceDao extends AbstractJdbcBatchMetadataDao implements
Assert.state(getJobInstance(job, jobParameters) == null, "JobInstance must not already exist");
Long jobId = new Long(jobIncrementer.nextLongValue());
Object[] parameters = new Object[] { jobId, job.getName(), createJobKey(jobParameters) };
JobInstance jobInstance = new JobInstance(jobId, jobParameters, job);
jobInstance.incrementVersion();
Object[] parameters = new Object[] { jobId, job.getName(), createJobKey(jobParameters), jobInstance.getVersion() };
getJdbcTemplate().update(getQuery(CREATE_JOB_INSTANCE), parameters,
new int[] { Types.INTEGER, Types.VARCHAR, Types.VARCHAR });
new int[] { Types.INTEGER, Types.VARCHAR, Types.VARCHAR, Types.INTEGER });
insertJobParameters(jobId, jobParameters);
JobInstance jobInstance = new JobInstance(jobId, jobParameters, job);
return jobInstance;
}
private String createJobKey(JobParameters jobParameters) {
Map props = jobParameters.getParameters();
// Start with non-empty string for Oracle:
StringBuffer stringBuffer = new StringBuffer("key:");
StringBuffer stringBuffer = new StringBuffer();
for (Iterator it = props.entrySet().iterator(); it.hasNext();) {
Entry entry = (Entry) it.next();
stringBuffer.append(entry.toString() + ";");

59
spring-batch-core/src/main/java/org/springframework/batch/execution/repository/dao/MapJobExecutionDao.java
View File

@@ -4,66 +4,71 @@ import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobInstance;
import org.springframework.batch.support.transaction.TransactionAwareProxyFactory;
import org.springframework.util.Assert;
/**
* In-memory implementation of {@link JobExecutionDao}.
*
*/
public class MapJobExecutionDao implements JobExecutionDao {
private static Map executionsByJobInstanceId = TransactionAwareProxyFactory.createTransactionalMap();
private static Map executionsById = TransactionAwareProxyFactory.createTransactionalMap();
private static long currentId;
public static void clear() {
executionsByJobInstanceId.clear();
executionsById.clear();
}
public int getJobExecutionCount(JobInstance jobInstance) {
Set executions = (Set) executionsByJobInstanceId.get(jobInstance.getId());
if (executions == null) {
return 0;
int count = 0;
for (Iterator iterator = executionsById.values().iterator(); iterator.hasNext();) {
JobExecution exec = (JobExecution) iterator.next();
if (exec.getJobInstance().equals(jobInstance)) {
count++;
}
}
return executions.size();
return count;
}
public void saveJobExecution(JobExecution jobExecution) {
Set executions = (Set) executionsByJobInstanceId.get(jobExecution.getJobId());
if (executions == null) {
executions = TransactionAwareProxyFactory.createTransactionalSet();
executionsByJobInstanceId.put(jobExecution.getJobId(), executions);
}
executions.add(jobExecution);
jobExecution.setId(new Long(currentId++));
Assert.isTrue(jobExecution.getId() == null);
Long newId = new Long(currentId++);
jobExecution.setId(newId);
jobExecution.incrementVersion();
executionsById.put(newId, jobExecution);
}
public List findJobExecutions(JobInstance jobInstance) {
Set executions = (Set) executionsByJobInstanceId.get(jobInstance.getId());
if (executions == null) {
return new ArrayList();
}
else {
return new ArrayList(executions);
List executions = new ArrayList();
for (Iterator iterator = executionsById.values().iterator(); iterator.hasNext();) {
JobExecution exec = (JobExecution) iterator.next();
if (exec.getJobInstance().equals(jobInstance)) {
executions.add(exec);
}
}
return executions;
}
public void updateJobExecution(JobExecution jobExecution) {
// no-op
Long id = jobExecution.getId();
Assert.notNull(id, "JobExecution is expected to have an id (should be saved already)");
Assert.notNull(executionsById.get(id), "JobExecution must already be saved");
jobExecution.incrementVersion();
executionsById.put(id, jobExecution);
}
public JobExecution getLastJobExecution(JobInstance jobInstance) {
Set executions = (Set) executionsByJobInstanceId.get(jobInstance.getId());
if (executions == null) {
return null;
}
JobExecution lastExec = null;
for (Iterator iterator = executions.iterator(); iterator.hasNext();) {
for (Iterator iterator = executionsById.values().iterator(); iterator.hasNext();) {
JobExecution exec = (JobExecution) iterator.next();
if (!exec.getJobInstance().equals(jobInstance)) {
continue;
}
if (lastExec == null) {
lastExec = exec;
}

4
spring-batch-core/src/main/java/org/springframework/batch/execution/repository/dao/MapJobInstanceDao.java
View File

@@ -9,6 +9,9 @@ import org.springframework.batch.core.JobParameters;
import org.springframework.batch.support.transaction.TransactionAwareProxyFactory;
import org.springframework.util.Assert;
/**
* In-memory implementation of {@link JobInstanceDao}.
*/
public class MapJobInstanceDao implements JobInstanceDao {
private static Collection jobInstances = TransactionAwareProxyFactory.createTransactionalList();
@@ -24,6 +27,7 @@ public class MapJobInstanceDao implements JobInstanceDao {
Assert.state(getJobInstance(job, jobParameters) == null, "JobInstance must not already exist");
JobInstance jobInstance = new JobInstance(new Long(currentId++), jobParameters, job);
jobInstance.incrementVersion();
jobInstances.add(jobInstance);
return jobInstance;

22
spring-batch-core/src/main/java/org/springframework/batch/execution/repository/dao/MapStepExecutionDao.java
View File

@@ -51,8 +51,8 @@ public class MapStepExecutionDao implements StepExecutionDao {
}
public void saveStepExecution(StepExecution stepExecution) {
Assert.state(stepExecution.getId() == null);
Assert.state(stepExecution.getVersion() == null);
Assert.isTrue(stepExecution.getId() == null);
Assert.isTrue(stepExecution.getVersion() == null);
Assert.notNull(stepExecution.getJobExecutionId(), "JobExecution must be saved already.");
Map executions = (Map) executionsByJobExecutionId.get(stepExecution.getJobExecutionId());
@@ -60,8 +60,8 @@ public class MapStepExecutionDao implements StepExecutionDao {
executions = TransactionAwareProxyFactory.createTransactionalMap();
executionsByJobExecutionId.put(stepExecution.getJobExecutionId(), executions);
}
stepExecution.incrementVersion();
stepExecution.setId(new Long(currentId++));
stepExecution.incrementVersion();
executions.put(stepExecution.getStepName(), stepExecution);
}
@@ -75,14 +75,16 @@ public class MapStepExecutionDao implements StepExecutionDao {
StepExecution persistedExecution = (StepExecution) executions.get(stepExecution.getStepName());
Assert.notNull(persistedExecution, "step execution is expected to be already saved");
if (!persistedExecution.getVersion().equals(stepExecution.getVersion())) {
throw new OptimisticLockingFailureException("Attempt to update step execution id=" + stepExecution.getId()
+ " with wrong version (" + stepExecution.getVersion() + "), where current version is "
+ persistedExecution.getVersion());
}
synchronized (stepExecution) {
if (!persistedExecution.getVersion().equals(stepExecution.getVersion())) {
throw new OptimisticLockingFailureException("Attempt to update step execution id="
+ stepExecution.getId() + " with wrong version (" + stepExecution.getVersion()
+ "), where current version is " + persistedExecution.getVersion());
}
stepExecution.incrementVersion();
executions.put(stepExecution.getStepName(), stepExecution);
stepExecution.incrementVersion();
executions.put(stepExecution.getStepName(), stepExecution);
}
}
public StepExecution getStepExecution(JobExecution jobExecution, Step step) {