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

RESOLVED - BATCH-897: Version is not rehydrated from database in JobInstance or JobExecution

Browse Source 
-added concurrent modification checks for JobExecution
-included version in the SQL selects (version value is loaded from db)
This commit is contained in:
robokaso
2008-11-04 12:14:26 +00:00
parent 0f7fd164be
commit 7a4caca1fd
6 changed files with 143 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -15,6 +15,7 @@ import org.springframework.batch.core.JobInstance;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.dao.EmptyResultDataAccessException;
import org.springframework.dao.OptimisticLockingFailureException;
import org.springframework.jdbc.core.RowCallbackHandler;
import org.springframework.jdbc.core.simple.ParameterizedRowMapper;
import org.springframework.jdbc.support.incrementer.DataFieldMaxValueIncrementer;
@@ -41,23 +42,25 @@ public class JdbcJobExecutionDao extends AbstractJdbcBatchMetadataDao implements
+ "END_TIME, STATUS, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE, VERSION, CREATE_TIME, LAST_UPDATED) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
private static final String CHECK_JOB_EXECUTION_EXISTS = "SELECT COUNT(*) FROM %PREFIX%JOB_EXECUTION WHERE JOB_EXECUTION_ID = ?";
private static final String GET_STATUS = "SELECT STATUS 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 = ?, VERSION = ?, CREATE_TIME = ?, LAST_UPDATED = ? where JOB_EXECUTION_ID = ?";
+ " STATUS = ?, CONTINUABLE = ?, EXIT_CODE = ?, EXIT_MESSAGE = ?, VERSION = ?, CREATE_TIME = ?, LAST_UPDATED = ? where JOB_EXECUTION_ID = ? and VERSION = ?";
private static final String FIND_JOB_EXECUTIONS = "SELECT JOB_EXECUTION_ID, START_TIME, END_TIME, STATUS, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE, CREATE_TIME, LAST_UPDATED" +
" from %PREFIX%JOB_EXECUTION where JOB_INSTANCE_ID = ? order by JOB_EXECUTION_ID desc";
private static final String FIND_JOB_EXECUTIONS = "SELECT JOB_EXECUTION_ID, START_TIME, END_TIME, STATUS, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE, CREATE_TIME, LAST_UPDATED, VERSION"
+ " from %PREFIX%JOB_EXECUTION where JOB_INSTANCE_ID = ? order by JOB_EXECUTION_ID desc";
private static final String GET_LAST_EXECUTION = "SELECT JOB_EXECUTION_ID, START_TIME, END_TIME, STATUS, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE, CREATE_TIME, LAST_UPDATED " +
"from %PREFIX%JOB_EXECUTION where JOB_INSTANCE_ID = ? and CREATE_TIME = (SELECT max(CREATE_TIME) from %PREFIX%JOB_EXECUTION where JOB_INSTANCE_ID = ?)";
private static final String GET_LAST_EXECUTION = "SELECT JOB_EXECUTION_ID, START_TIME, END_TIME, STATUS, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE, CREATE_TIME, LAST_UPDATED, VERSION "
+ "from %PREFIX%JOB_EXECUTION where JOB_INSTANCE_ID = ? and CREATE_TIME = (SELECT max(CREATE_TIME) from %PREFIX%JOB_EXECUTION where JOB_INSTANCE_ID = ?)";
private static final String GET_EXECUTION_BY_ID = "SELECT JOB_EXECUTION_ID, START_TIME, END_TIME, STATUS, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE, CREATE_TIME, LAST_UPDATED" +
" from %PREFIX%JOB_EXECUTION where JOB_EXECUTION_ID = ?";
private static final String GET_EXECUTION_BY_ID = "SELECT JOB_EXECUTION_ID, START_TIME, END_TIME, STATUS, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE, CREATE_TIME, LAST_UPDATED, VERSION"
+ " from %PREFIX%JOB_EXECUTION where JOB_EXECUTION_ID = ?";
private static final String GET_RUNNING_EXECUTIONS = "SELECT JOB_EXECUTION_ID, START_TIME, END_TIME, STATUS, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE, CREATE_TIME, LAST_UPDATED, " +
"JOB_INSTANCE_ID from %PREFIX%JOB_EXECUTION where END_TIME is NULL order by JOB_EXECUTION_ID desc";
private static final String GET_RUNNING_EXECUTIONS = "SELECT JOB_EXECUTION_ID, START_TIME, END_TIME, STATUS, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE, CREATE_TIME, LAST_UPDATED, VERSION "
+ "JOB_INSTANCE_ID from %PREFIX%JOB_EXECUTION where END_TIME is NULL order by JOB_EXECUTION_ID desc";
private static final String CURRENT_VERSION_JOB_EXECUTION = "SELECT VERSION FROM %PREFIX%JOB_EXECUTION WHERE JOB_EXECUTION_ID=?";
private int exitMessageLength = DEFAULT_EXIT_MESSAGE_LENGTH;
@@ -151,35 +154,52 @@ public class JdbcJobExecutionDao extends AbstractJdbcBatchMetadataDao implements
validateJobExecution(jobExecution);
jobExecution.incrementVersion();
Assert.notNull(jobExecution.getId(),
"JobExecution ID cannot be null. JobExecution must be saved before it can be updated");
String exitDescription = jobExecution.getExitStatus().getExitDescription();
if (exitDescription != null && exitDescription.length() > exitMessageLength) {
exitDescription = exitDescription.substring(0, exitMessageLength);
logger.debug("Truncating long message before update of JobExecution: " + jobExecution);
Assert.notNull(jobExecution.getVersion(),
"JobExecution version cannot be null. JobExecution must be saved before it can be updated");
synchronized (jobExecution) {
Integer version = jobExecution.getVersion() + 1;
String exitDescription = jobExecution.getExitStatus().getExitDescription();
if (exitDescription != null && exitDescription.length() > exitMessageLength) {
exitDescription = exitDescription.substring(0, exitMessageLength);
logger.debug("Truncating long message before update of JobExecution: " + jobExecution);
}
Object[] parameters = new Object[] { jobExecution.getStartTime(), jobExecution.getEndTime(),
jobExecution.getStatus().toString(), jobExecution.getExitStatus().isContinuable() ? "Y" : "N",
jobExecution.getExitStatus().getExitCode(), exitDescription, version, jobExecution.getCreateTime(),
jobExecution.getLastUpdated(), jobExecution.getId(), jobExecution.getVersion() };
// Check if given JobExecution's Id already exists, if none is found
// it
// 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.");
}
int count = getJdbcTemplate().getJdbcOperations().update(
getQuery(UPDATE_JOB_EXECUTION),
parameters,
new int[] { Types.TIMESTAMP, Types.TIMESTAMP, Types.VARCHAR, Types.CHAR, Types.VARCHAR,
Types.VARCHAR, Types.INTEGER, Types.TIMESTAMP, Types.TIMESTAMP, Types.INTEGER,
Types.INTEGER });
// Avoid concurrent modifications...
if (count == 0) {
int curentVersion = getJdbcTemplate().queryForInt(getQuery(CURRENT_VERSION_JOB_EXECUTION),
new Object[] { jobExecution.getId() });
throw new OptimisticLockingFailureException("Attempt to update step execution id="
+ jobExecution.getId() + " with wrong version (" + jobExecution.getVersion()
+ "), where current version is " + curentVersion);
}
jobExecution.incrementVersion();
}
Object[] parameters = new Object[] { jobExecution.getStartTime(), jobExecution.getEndTime(),
jobExecution.getStatus().toString(), jobExecution.getExitStatus().isContinuable() ? "Y" : "N",
jobExecution.getExitStatus().getExitCode(), exitDescription, jobExecution.getVersion(),
jobExecution.getCreateTime(), jobExecution.getLastUpdated(), jobExecution.getId() };
if (jobExecution.getId() == null) {
throw new IllegalArgumentException("JobExecution ID cannot be null. JobExecution must be saved "
+ "before it can be updated.");
}
// Check if given JobExecution's Id already exists, if none is found it
// 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.");
}
getJdbcTemplate().getJdbcOperations().update(
getQuery(UPDATE_JOB_EXECUTION),
parameters,
new int[] { Types.TIMESTAMP, Types.TIMESTAMP, Types.VARCHAR, Types.CHAR, Types.VARCHAR, Types.VARCHAR,
Types.INTEGER, Types.TIMESTAMP, Types.TIMESTAMP, Types.INTEGER });
}
public JobExecution getLastJobExecution(JobInstance jobInstance) {
@@ -235,9 +255,9 @@ public class JdbcJobExecutionDao extends AbstractJdbcBatchMetadataDao implements
return result;
}
public void synchronizeStatus(JobExecution jobExecution) {
String status = getJdbcTemplate().queryForObject(getQuery(GET_STATUS), String.class, jobExecution.getId());
jobExecution.setStatus(BatchStatus.valueOf(status));
}
@@ -251,31 +271,32 @@ public class JdbcJobExecutionDao extends AbstractJdbcBatchMetadataDao implements
private class JobExecutionRowMapper implements ParameterizedRowMapper<JobExecution> {
private JobInstance jobInstance;
public JobExecutionRowMapper() {
}
public JobExecutionRowMapper(JobInstance jobInstance) {
this.jobInstance = jobInstance;
}
public JobExecution mapRow(ResultSet rs, int rowNum) throws SQLException {
Long id = new Long(rs.getLong(1));
JobExecution jobExecution;
if(jobInstance == null){
JobExecution jobExecution;
if (jobInstance == null) {
jobExecution = new JobExecution(id);
}
else{
else {
jobExecution = new JobExecution(jobInstance, id);
}
jobExecution.setStartTime(rs.getTimestamp(2));
jobExecution.setEndTime(rs.getTimestamp(3));
jobExecution.setStatus(BatchStatus.valueOf(rs.getString(4)));
jobExecution.setExitStatus(new ExitStatus("Y".equals(rs.getString(5)), rs.getString(6), rs.getString(7)));
jobExecution.setCreateTime(rs.getTimestamp(8));
jobExecution.setLastUpdated(rs.getTimestamp(9));
jobExecution.setVersion(rs.getInt(10));
return jobExecution;
}

8
spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JdbcStepExecutionDao.java
View File

@@ -130,12 +130,6 @@ public class JdbcStepExecutionDao extends AbstractJdbcBatchMetadataDao implement
Assert.notNull(stepExecution.getStatus(), "StepExecution status cannot be null.");
}
/*
* (non-Javadoc)
*
* @seeorg.springframework.batch.execution.repository.dao.StepExecutionDao#
* updateStepExecution(org.springframework.batch.core.domain.StepExecution)
*/
public void updateStepExecution(StepExecution stepExecution) {
validateStepExecution(stepExecution);
@@ -151,7 +145,7 @@ public class JdbcStepExecutionDao extends AbstractJdbcBatchMetadataDao implement
// someone is already trying to do it.
synchronized (stepExecution) {
Integer version = new Integer(stepExecution.getVersion().intValue() + 1);
Integer version = stepExecution.getVersion() + 1;
Object[] parameters = new Object[] { stepExecution.getStartTime(), stepExecution.getEndTime(),
stepExecution.getStatus().toString(), stepExecution.getCommitCount(), stepExecution.getReadCount(),
stepExecution.getFilterCount(), stepExecution.getWriteCount(),

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

@@ -12,6 +12,7 @@ import org.apache.commons.lang.SerializationUtils;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobInstance;
import org.springframework.batch.support.transaction.TransactionAwareProxyFactory;
import org.springframework.dao.OptimisticLockingFailureException;
import org.springframework.util.Assert;
/**
@@ -68,9 +69,18 @@ public class MapJobExecutionDao implements JobExecutionDao {
public void updateJobExecution(JobExecution jobExecution) {
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, copy(jobExecution));
JobExecution persistedExecution = executionsById.get(id);
Assert.notNull(persistedExecution, "JobExecution must already be saved");
synchronized (jobExecution) {
if (!persistedExecution.getVersion().equals(jobExecution.getVersion())){
throw new OptimisticLockingFailureException("Attempt to update step execution id=" + id
+ " with wrong version (" + jobExecution.getVersion() + "), where current version is "
+ persistedExecution.getVersion());
}
jobExecution.incrementVersion();
executionsById.put(id, copy(jobExecution));
}
}
public JobExecution getLastJobExecution(JobInstance jobInstance) {

1
spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/AbstractJobDaoTests.java
View File

@@ -190,6 +190,7 @@ public abstract class AbstractJobDaoTests {
// id is invalid
JobExecution execution = new JobExecution(jobInstance, (long) 29432);
execution.incrementVersion();
try {
jobExecutionDao.updateJobExecution(execution);
fail("Expected NoSuchBatchDomainObjectException");

83
spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/AbstractJobExecutionDaoTests.java
View File

@@ -1,9 +1,9 @@
package org.springframework.batch.core.repository.dao;
import static org.junit.Assert.*;
import org.junit.Before;
import org.junit.Test;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.fail;
import java.util.ArrayList;
import java.util.Collections;
@@ -11,22 +11,25 @@ import java.util.Date;
import java.util.List;
import java.util.Set;
import org.junit.Before;
import org.junit.Test;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobInstance;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.dao.OptimisticLockingFailureException;
import org.springframework.test.context.junit4.AbstractTransactionalJUnit4SpringContextTests;
import org.springframework.transaction.annotation.Transactional;
public abstract class AbstractJobExecutionDaoTests extends AbstractTransactionalJUnit4SpringContextTests {
JobExecutionDao dao;
protected JobExecutionDao dao;
JobInstance jobInstance = new JobInstance((long) 1, new JobParameters(), "execTestJob");
protected JobInstance jobInstance = new JobInstance((long) 1, new JobParameters(), "execTestJob");
JobExecution execution = new JobExecution(jobInstance);
protected JobExecution execution = new JobExecution(jobInstance);
/**
* @return tested object ready for use
@@ -63,31 +66,30 @@ public abstract class AbstractJobExecutionDaoTests extends AbstractTransactional
assertEquals(execution, executions.get(0));
assertExecutionsAreEqual(execution, executions.get(0));
}
/**
* Executions should be returned in the reverse order they were saved.
*/
@Transactional
@Test
public void testFindExecutionsOrdering() {
List<JobExecution> execs = new ArrayList<JobExecution>();
for (int i = 0; i < 10; i++) {
JobExecution exec = new JobExecution(jobInstance);
exec.setCreateTime(new Date(i));
execs.add(exec);
dao.saveJobExecution(exec);
}
List<JobExecution> retrieved = dao.findJobExecutions(jobInstance);
Collections.reverse(retrieved);
for (int i = 0; i < 10; i++) {
assertExecutionsAreEqual(execs.get(i), retrieved.get(i));
}
}
/**
@@ -162,7 +164,7 @@ public abstract class AbstractJobExecutionDaoTests extends AbstractTransactional
JobExecution value = dao.getLastJobExecution(jobInstance);
assertNull(value);
}
/**
* Check the execution is returned
*/
@@ -189,7 +191,7 @@ public abstract class AbstractJobExecutionDaoTests extends AbstractTransactional
assertEquals(1, values.size());
JobExecution value = values.iterator().next();
assertEquals(exec, value);
assertEquals(5L, value.getLastUpdated().getTime());
assertEquals(5L, value.getLastUpdated().getTime());
}
/**
@@ -201,7 +203,7 @@ public abstract class AbstractJobExecutionDaoTests extends AbstractTransactional
Set<JobExecution> values = dao.findRunningJobExecutions("no-such-job");
assertEquals(0, values.size());
}
/**
* Check the execution is returned
*/
@@ -233,20 +235,51 @@ public abstract class AbstractJobExecutionDaoTests extends AbstractTransactional
JobExecution value = dao.getJobExecution(54321L);
assertNull(value);
}
/*
* Check to make sure the executions are equal. Normally, comparing the id's is
* sufficient. However, for testing purposes, especially of a dao, we need to make
* sure all the fields are being stored/retrieved correctly.
/**
* Exception should be raised when the version of update argument doesn't
* match the version of persisted entity.
*/
private void assertExecutionsAreEqual(JobExecution lhs, JobExecution rhs){
@Transactional
@Test
public void testConcurrentModificationException() {
JobExecution exec1 = new JobExecution(jobInstance);
dao.saveJobExecution(exec1);
JobExecution exec2 = new JobExecution(jobInstance);
exec2.setId(exec1.getId());
exec2.incrementVersion();
assertEquals((Integer) 0, exec1.getVersion());
assertEquals(exec1.getVersion(), exec2.getVersion());
dao.updateJobExecution(exec1);
assertEquals((Integer) 1, exec1.getVersion());
try {
dao.updateJobExecution(exec2);
fail();
}
catch (OptimisticLockingFailureException e) {
// expected
}
} /*
* Check to make sure the executions are equal. Normally, comparing the id's
* is sufficient. However, for testing purposes, especially of a DAO, we
* need to make sure all the fields are being stored/retrieved correctly.
*/
private void assertExecutionsAreEqual(JobExecution lhs, JobExecution rhs) {
assertEquals(lhs.getId(), rhs.getId());
assertEquals(lhs.getStartTime(), rhs.getStartTime());
assertEquals(lhs.getStatus(), rhs.getStatus());
assertEquals(lhs.getEndTime(), rhs.getEndTime());
assertEquals(lhs.getCreateTime(), rhs.getCreateTime());
assertEquals(lhs.getLastUpdated(), rhs.getLastUpdated());
assertEquals(lhs.getVersion(), rhs.getVersion());
}
}

5
spring-batch-samples/src/test/java/org/springframework/batch/sample/DatabaseShutdownFunctionalTests.java
View File

@@ -16,9 +16,7 @@
package org.springframework.batch.sample;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.*;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -59,6 +57,7 @@ public class DatabaseShutdownFunctionalTests extends AbstractBatchLauncherTests
assertEquals(BatchStatus.STARTED, jobExecution.getStatus());
assertTrue(jobExecution.isRunning());
assertNotNull(jobExecution.getVersion());
jobOperator.stop(jobExecution.getId());