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

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

Browse Source 
distinguish between save and update in ExecutionContextDao interface
initially save EC together with the execution it belongs to
This commit is contained in:
robokaso
2009-01-20 13:32:14 +00:00
parent 784e640490
commit c08bdad717
7 changed files with 113 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/ExecutionContextDao.java
View File

@@ -40,14 +40,30 @@ public interface ExecutionContextDao {
ExecutionContext getExecutionContext(StepExecution stepExecution);
/**
* Persist the execution context associated with the given jobExecution
* Persist the execution context associated with the given jobExecution,
* persistent entry for the context should not exist yet.
* @param jobExecution
*/
void persistExecutionContext(final JobExecution jobExecution);
void saveExecutionContext(final JobExecution jobExecution);
/**
* Persist the execution context associated with the given stepExecution
* Persist the execution context associated with the given stepExecution,
* persistent entry for the context should not exist yet.
* @param stepExecution
*/
void persistExecutionContext(final StepExecution stepExecution);
void saveExecutionContext(final StepExecution stepExecution);
/**
* Persist the updates of execution context associated with the given
* jobExecution. Persistent entry should already exist for this context.
* @param jobExecution
*/
void updateExecutionContext(final JobExecution jobExecution);
/**
* Persist the updates of execution context associated with the given
* stepExecution. Persistent entry should already exist for this context.
* @param stepExecution
*/
void updateExecutionContext(final StepExecution stepExecution);
}

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

@@ -49,31 +49,26 @@ public class JdbcExecutionContextDao extends AbstractJdbcBatchMetadataDao implem
private static final String COUNT_JOB_EXECUTION_CONTEXT = "SELECT COUNT(*) FROM %PREFIX%JOB_EXECUTION_CONTEXT "
+ "WHERE JOB_EXECUTION_ID = ?";
private static final String FIND_JOB_EXECUTION_CONTEXT = "SELECT SHORT_CONTEXT, SERIALIZED_CONTEXT " +
"FROM %PREFIX%JOB_EXECUTION_CONTEXT WHERE JOB_EXECUTION_ID = ?";
private static final String FIND_JOB_EXECUTION_CONTEXT = "SELECT SHORT_CONTEXT, SERIALIZED_CONTEXT "
+ "FROM %PREFIX%JOB_EXECUTION_CONTEXT WHERE JOB_EXECUTION_ID = ?";
private static final String INSERT_JOB_EXECUTION_CONTEXT = "INSERT INTO %PREFIX%JOB_EXECUTION_CONTEXT " +
"(JOB_EXECUTION_ID, SHORT_CONTEXT, SERIALIZED_CONTEXT) " +
"VALUES(?, ?, ?)";
private static final String INSERT_JOB_EXECUTION_CONTEXT = "INSERT INTO %PREFIX%JOB_EXECUTION_CONTEXT "
+ "(JOB_EXECUTION_ID, SHORT_CONTEXT, SERIALIZED_CONTEXT) " + "VALUES(?, ?, ?)";
private static final String UPDATE_JOB_EXECUTION_CONTEXT = "UPDATE %PREFIX%JOB_EXECUTION_CONTEXT " +
"SET SHORT_CONTEXT = ?, SERIALIZED_CONTEXT = ? " +
"WHERE JOB_EXECUTION_ID = ?";
private static final String UPDATE_JOB_EXECUTION_CONTEXT = "UPDATE %PREFIX%JOB_EXECUTION_CONTEXT "
+ "SET SHORT_CONTEXT = ?, SERIALIZED_CONTEXT = ? " + "WHERE JOB_EXECUTION_ID = ?";
private static final String COUNT_STEP_EXECUTION_CONTEXT = "SELECT COUNT(*) FROM %PREFIX%STEP_EXECUTION_CONTEXT "
+ "WHERE STEP_EXECUTION_ID = ?";
private static final String FIND_STEP_EXECUTION_CONTEXT = "SELECT SHORT_CONTEXT, SERIALIZED_CONTEXT " +
"FROM %PREFIX%STEP_EXECUTION_CONTEXT WHERE STEP_EXECUTION_ID = ?";
private static final String FIND_STEP_EXECUTION_CONTEXT = "SELECT SHORT_CONTEXT, SERIALIZED_CONTEXT "
+ "FROM %PREFIX%STEP_EXECUTION_CONTEXT WHERE STEP_EXECUTION_ID = ?";
private static final String INSERT_STEP_EXECUTION_CONTEXT = "INSERT INTO %PREFIX%STEP_EXECUTION_CONTEXT "
+ "(STEP_EXECUTION_ID, SHORT_CONTEXT, SERIALIZED_CONTEXT) " + "VALUES(?, ?, ?)";
private static final String INSERT_STEP_EXECUTION_CONTEXT = "INSERT INTO %PREFIX%STEP_EXECUTION_CONTEXT " +
"(STEP_EXECUTION_ID, SHORT_CONTEXT, SERIALIZED_CONTEXT) " +
"VALUES(?, ?, ?)";
private static final String UPDATE_STEP_EXECUTION_CONTEXT = "UPDATE %PREFIX%STEP_EXECUTION_CONTEXT " +
"SET SHORT_CONTEXT = ?, SERIALIZED_CONTEXT = ? " +
"WHERE STEP_EXECUTION_ID = ?";
private static final String UPDATE_STEP_EXECUTION_CONTEXT = "UPDATE %PREFIX%STEP_EXECUTION_CONTEXT "
+ "SET SHORT_CONTEXT = ?, SERIALIZED_CONTEXT = ? " + "WHERE STEP_EXECUTION_ID = ?";
private static final int MAX_VARCHAR_LENGTH = 2500;
@@ -90,8 +85,7 @@ public class JdbcExecutionContextDao extends AbstractJdbcBatchMetadataDao implem
Assert.notNull(executionId, "ExecutionId must not be null.");
List<ExecutionContext> results = getJdbcTemplate().query(getQuery(FIND_JOB_EXECUTION_CONTEXT),
new ExecutionContextRowMapper(),
executionId);
new ExecutionContextRowMapper(), executionId);
if (results.size() > 0) {
return results.get(0);
}
@@ -109,8 +103,7 @@ public class JdbcExecutionContextDao extends AbstractJdbcBatchMetadataDao implem
Assert.notNull(executionId, "ExecutionId must not be null.");
List<ExecutionContext> results = getJdbcTemplate().query(getQuery(FIND_STEP_EXECUTION_CONTEXT),
new ExecutionContextRowMapper(),
executionId);
new ExecutionContextRowMapper(), executionId);
if (results.size() > 0) {
return results.get(0);
}
@@ -124,7 +117,7 @@ public class JdbcExecutionContextDao extends AbstractJdbcBatchMetadataDao implem
* jobExecution
* @param jobExecution
*/
public void persistExecutionContext(final JobExecution jobExecution) {
public void updateExecutionContext(final JobExecution jobExecution) {
Long executionId = jobExecution.getId();
ExecutionContext executionContext = jobExecution.getExecutionContext();
Assert.notNull(executionId, "ExecutionId must not be null.");
@@ -140,7 +133,7 @@ public class JdbcExecutionContextDao extends AbstractJdbcBatchMetadataDao implem
* stepExecution
* @param stepExecution
*/
public void persistExecutionContext(final StepExecution stepExecution) {
public void updateExecutionContext(final StepExecution stepExecution) {
Long executionId = stepExecution.getId();
ExecutionContext executionContext = stepExecution.getExecutionContext();
@@ -152,6 +145,14 @@ public class JdbcExecutionContextDao extends AbstractJdbcBatchMetadataDao implem
persistSerializedContext(executionId, serializedContext, false);
}
public void saveExecutionContext(JobExecution jobExecution) {
updateExecutionContext(jobExecution);
}
public void saveExecutionContext(StepExecution stepExecution) {
updateExecutionContext(stepExecution);
}
public void setLobHandler(LobHandler lobHandler) {
this.lobHandler = lobHandler;
}
@@ -160,10 +161,11 @@ public class JdbcExecutionContextDao extends AbstractJdbcBatchMetadataDao implem
public void afterPropertiesSet() throws Exception {
super.afterPropertiesSet();
serializer = new XStreamExecutionContextStringSerializer();
((XStreamExecutionContextStringSerializer)serializer).afterPropertiesSet();
((XStreamExecutionContextStringSerializer) serializer).afterPropertiesSet();
}
private void persistSerializedContext(final Long executionId, String serializedContext, boolean isJobExecutionContext) {
private void persistSerializedContext(final Long executionId, String serializedContext,
boolean isJobExecutionContext) {
String countSql = isJobExecutionContext ? COUNT_JOB_EXECUTION_CONTEXT : COUNT_STEP_EXECUTION_CONTEXT;
String updateSql = isJobExecutionContext ? UPDATE_JOB_EXECUTION_CONTEXT : UPDATE_STEP_EXECUTION_CONTEXT;
String insertSql = isJobExecutionContext ? INSERT_JOB_EXECUTION_CONTEXT : INSERT_STEP_EXECUTION_CONTEXT;
@@ -181,36 +183,33 @@ public class JdbcExecutionContextDao extends AbstractJdbcBatchMetadataDao implem
longContext = null;
}
if (count > 0) {
getJdbcTemplate().getJdbcOperations().update(getQuery(updateSql),
new PreparedStatementSetter() {
public void setValues(PreparedStatement ps) throws SQLException {
ps.setString(1, shortContext);
if (longContext != null) {
lobHandler.getLobCreator().setClobAsString(ps, 2, longContext);
}
else {
ps.setNull(2, Types.CLOB);
}
ps.setLong(3, executionId);
}
});
getJdbcTemplate().getJdbcOperations().update(getQuery(updateSql), new PreparedStatementSetter() {
public void setValues(PreparedStatement ps) throws SQLException {
ps.setString(1, shortContext);
if (longContext != null) {
lobHandler.getLobCreator().setClobAsString(ps, 2, longContext);
}
else {
ps.setNull(2, Types.CLOB);
}
ps.setLong(3, executionId);
}
});
}
else {
getJdbcTemplate().getJdbcOperations().update(getQuery(insertSql),
new PreparedStatementSetter() {
public void setValues(PreparedStatement ps) throws SQLException {
ps.setLong(1, executionId);
ps.setString(2, shortContext);
if (longContext != null) {
lobHandler.getLobCreator().setClobAsString(ps, 3, longContext);
}
else {
ps.setNull(3, Types.CLOB);
}
}
});
getJdbcTemplate().getJdbcOperations().update(getQuery(insertSql), new PreparedStatementSetter() {
public void setValues(PreparedStatement ps) throws SQLException {
ps.setLong(1, executionId);
ps.setString(2, shortContext);
if (longContext != null) {
lobHandler.getLobCreator().setClobAsString(ps, 3, longContext);
}
else {
ps.setNull(3, Types.CLOB);
}
}
});
}
}
@@ -236,4 +235,5 @@ public class JdbcExecutionContextDao extends AbstractJdbcBatchMetadataDao implem
return executionContext;
}
}
}

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

@@ -44,7 +44,7 @@ public class MapExecutionContextDao implements ExecutionContextDao {
return copy(contextsByStepExecutionId.get(stepExecution.getId()));
}
public void persistExecutionContext(StepExecution stepExecution) {
public void updateExecutionContext(StepExecution stepExecution) {
contextsByStepExecutionId.put(stepExecution.getId(), copy(stepExecution.getExecutionContext()));
}
@@ -52,9 +52,17 @@ public class MapExecutionContextDao implements ExecutionContextDao {
return copy(contextsByJobExecutionId.get(jobExecution.getId()));
}
public void persistExecutionContext(JobExecution jobExecution) {
public void updateExecutionContext(JobExecution jobExecution) {
contextsByJobExecutionId.put(jobExecution.getId(), copy(jobExecution.getExecutionContext()));
}
public void saveExecutionContext(JobExecution jobExecution) {
updateExecutionContext(jobExecution);
}
public void saveExecutionContext(StepExecution stepExecution) {
updateExecutionContext(stepExecution);
}
}

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

@@ -196,18 +196,17 @@ public class SimpleJobRepository implements JobRepository {
// Save the JobExecution so that it picks up an ID (useful for clients
// monitoring asynchronous executions):
jobExecutionDao.saveJobExecution(jobExecution);
ecDao.saveExecutionContext(jobExecution);
return jobExecution;
}
/**
* Save or 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. If an id isn't
* found, a new JobExecution is created, if one is found, the current row is
* updated.
* 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.
@@ -233,6 +232,7 @@ public class SimpleJobRepository implements JobRepository {
stepExecution.setLastUpdated(new Date(System.currentTimeMillis()));
stepExecutionDao.saveStepExecution(stepExecution);
ecDao.saveExecutionContext(stepExecution);
}
/**
@@ -256,17 +256,9 @@ public class SimpleJobRepository implements JobRepository {
Assert.notNull(stepExecution.getJobExecutionId(), "StepExecution must belong to persisted JobExecution");
}
/*
* (non-Javadoc)
*
* @see org.springframework.batch.core.repository.JobRepository#
* persistExecutionContext
* (org.springframework.batch.core.domain.StepExecution)
*/
public void updateExecutionContext(StepExecution stepExecution) {
// Until there is an interface change (
ecDao.persistExecutionContext(stepExecution.getJobExecution());
ecDao.persistExecutionContext(stepExecution);
ecDao.updateExecutionContext(stepExecution.getJobExecution());
ecDao.updateExecutionContext(stepExecution);
}
/**

18
spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/AbstractExecutionContextDaoTests.java
View File

@@ -76,7 +76,7 @@ public abstract class AbstractExecutionContextDaoTests extends AbstractTransacti
ExecutionContext ctx = new ExecutionContext(Collections.<String, Object> singletonMap("key", "value"));
jobExecution.setExecutionContext(ctx);
contextDao.persistExecutionContext(jobExecution);
contextDao.updateExecutionContext(jobExecution);
ExecutionContext retrieved = contextDao.getExecutionContext(jobExecution);
assertEquals(ctx, retrieved);
@@ -88,7 +88,7 @@ public abstract class AbstractExecutionContextDaoTests extends AbstractTransacti
ExecutionContext ctx = new ExecutionContext();
jobExecution.setExecutionContext(ctx);
contextDao.persistExecutionContext(jobExecution);
contextDao.updateExecutionContext(jobExecution);
ExecutionContext retrieved = contextDao.getExecutionContext(jobExecution);
assertEquals(ctx, retrieved);
@@ -101,10 +101,10 @@ public abstract class AbstractExecutionContextDaoTests extends AbstractTransacti
ExecutionContext ctx = new ExecutionContext(Collections
.<String, Object> singletonMap("key", "value"));
jobExecution.setExecutionContext(ctx);
contextDao.persistExecutionContext(jobExecution);
contextDao.updateExecutionContext(jobExecution);
ctx.putLong("longKey", 7);
contextDao.persistExecutionContext(jobExecution);
contextDao.updateExecutionContext(jobExecution);
ExecutionContext retrieved = contextDao.getExecutionContext(jobExecution);
assertEquals(ctx, retrieved);
@@ -117,7 +117,7 @@ public abstract class AbstractExecutionContextDaoTests extends AbstractTransacti
ExecutionContext ctx = new ExecutionContext(Collections.<String, Object> singletonMap("key", "value"));
stepExecution.setExecutionContext(ctx);
contextDao.persistExecutionContext(stepExecution);
contextDao.updateExecutionContext(stepExecution);
ExecutionContext retrieved = contextDao.getExecutionContext(stepExecution);
assertEquals(ctx, retrieved);
@@ -129,7 +129,7 @@ public abstract class AbstractExecutionContextDaoTests extends AbstractTransacti
ExecutionContext ctx = new ExecutionContext();
stepExecution.setExecutionContext(ctx);
contextDao.persistExecutionContext(stepExecution);
contextDao.updateExecutionContext(stepExecution);
ExecutionContext retrieved = contextDao.getExecutionContext(stepExecution);
assertEquals(ctx, retrieved);
@@ -141,10 +141,10 @@ public abstract class AbstractExecutionContextDaoTests extends AbstractTransacti
ExecutionContext ctx = new ExecutionContext(Collections.<String, Object> singletonMap("key", "value"));
stepExecution.setExecutionContext(ctx);
contextDao.persistExecutionContext(stepExecution);
contextDao.updateExecutionContext(stepExecution);
ctx.putLong("longKey", 7);
contextDao.persistExecutionContext(stepExecution);
contextDao.updateExecutionContext(stepExecution);
ExecutionContext retrieved = contextDao.getExecutionContext(stepExecution);
assertEquals(ctx, retrieved);
@@ -158,7 +158,7 @@ public abstract class AbstractExecutionContextDaoTests extends AbstractTransacti
ExecutionContext ec = new ExecutionContext();
ec.put("intValue", new Integer(343232));
stepExecution.setExecutionContext(ec);
contextDao.persistExecutionContext(stepExecution);
contextDao.updateExecutionContext(stepExecution);
ExecutionContext restoredEc = contextDao.getExecutionContext(stepExecution);
assertEquals(ec, restoredEc);
}

4
spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/MapExecutionContextDaoTests.java
View File

@@ -45,13 +45,13 @@ public class MapExecutionContextDaoTests extends AbstractExecutionContextDaoTest
StepExecution stepExecution = new StepExecution("stepName", jobExecution);
assertTrue(stepExecution.getExecutionContext().isEmpty());
tested.persistExecutionContext(stepExecution);
tested.updateExecutionContext(stepExecution);
stepExecution.getExecutionContext().put("key","value");
ExecutionContext retrieved = tested.getExecutionContext(stepExecution);
assertTrue(retrieved.isEmpty());
tested.persistExecutionContext(jobExecution);
tested.updateExecutionContext(jobExecution);
jobExecution.getExecutionContext().put("key", "value");
retrieved = tested.getExecutionContext(jobExecution);
assertTrue(retrieved.isEmpty());

19
spring-batch-samples/src/test/java/org/springframework/batch/sample/support/JdbcJobRepositoryTests.java
View File

@@ -50,7 +50,7 @@ import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = {"/simple-job-launcher-context.xml"})
@ContextConfiguration(locations = { "/simple-job-launcher-context.xml" })
public class JdbcJobRepositoryTests {
private JobRepository repository;
@@ -67,9 +67,9 @@ public class JdbcJobRepositoryTests {
private PlatformTransactionManager transactionManager;
/** Logger */
/** Logger */
private final Log logger = LogFactory.getLog(getClass());
@Autowired
public void setDataSource(DataSource dataSource) {
this.simpleJdbcTemplate = new SimpleJdbcTemplate(dataSource);
@@ -100,18 +100,21 @@ public class JdbcJobRepositoryTests {
@AfterTransaction
public void onTearDownAfterTransaction() throws Exception {
for (Long id : jobExecutionIds) {
simpleJdbcTemplate.update("DELETE FROM BATCH_JOB_EXECUTION_CONTEXT where JOB_EXECUTION_ID=?", id);
simpleJdbcTemplate.update("DELETE FROM BATCH_JOB_EXECUTION where JOB_EXECUTION_ID=?", id);
}
for (Long id : jobIds) {
simpleJdbcTemplate.update("DELETE FROM BATCH_JOB_INSTANCE where JOB_INSTANCE_ID=?", id);
}
for (Long id : jobIds) {
int count = simpleJdbcTemplate.queryForInt("SELECT COUNT(*) FROM BATCH_JOB_INSTANCE where JOB_INSTANCE_ID=?", id);
int count = simpleJdbcTemplate.queryForInt(
"SELECT COUNT(*) FROM BATCH_JOB_INSTANCE where JOB_INSTANCE_ID=?", id);
assertEquals(0, count);
}
}
@Transactional @Test
@Transactional
@Test
public void testFindOrCreateJob() throws Exception {
job.setName("foo");
int before = 0;
@@ -121,7 +124,8 @@ public class JdbcJobRepositoryTests {
assertNotNull(execution.getId());
}
@Transactional @Test
@Transactional
@Test
public void testFindOrCreateJobConcurrently() throws Exception {
job.setName("bar");
@@ -199,7 +203,8 @@ public class JdbcJobRepositoryTests {
new TransactionTemplate(transactionManager).execute(new TransactionCallback() {
public Object doInTransaction(org.springframework.transaction.TransactionStatus status) {
try {
JobExecution execution = repository.createJobExecution(job.getName(), new JobParameters());
JobExecution execution = repository.createJobExecution(job.getName(),
new JobParameters());
cacheJobIds(execution);
list.add(execution);
Thread.sleep(1000);