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

Use id instead of name to locate existing step execution (more uniform interface).

Browse Source
This commit is contained in:
dsyer
2009-01-03 10:16:59 +00:00
parent 71ec0a602e
commit c668a1f5db
10 changed files with 72 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
spring-batch-core/src/main/java/org/springframework/batch/core/explore/JobExplorer.java
View File

@@ -47,11 +47,10 @@ public interface JobExplorer {
/**
* @param jobExecutionId the parent job execution id
* @param stepName the step name identifier for the required
* {@link StepExecution}
* @param stepExecutionId the step execution id
* @return the {@link StepExecution} with this id, or null if not found
*/
StepExecution getStepExecution(Long jobExecutionId, String stepName);
StepExecution getStepExecution(Long jobExecutionId, Long stepExecutionId);
/**
* @param instanceId

8
spring-batch-core/src/main/java/org/springframework/batch/core/explore/support/SimpleJobExplorer.java
View File

@@ -101,12 +101,8 @@ public class SimpleJobExplorer implements JobExplorer {
/* (non-Javadoc)
* @see org.springframework.batch.core.explore.JobExplorer#getStepExecution(java.lang.Long)
*/
public StepExecution getStepExecution(Long executionId, String stepName) {
JobExecution jobExecution = getJobExecution(executionId);
if (jobExecution==null) {
return null;
}
return stepExecutionDao.getStepExecution(jobExecution, stepName);
public StepExecution getStepExecution(Long jobExecutionId, Long executionId) {
return stepExecutionDao.getStepExecution(getJobExecution(jobExecutionId), executionId);
}
/* (non-Javadoc)

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

@@ -54,7 +54,7 @@ public class JdbcStepExecutionDao extends AbstractJdbcBatchMetadataDao implement
private static final String GET_STEP_EXECUTIONS = GET_RAW_STEP_EXECUTIONS + " order by STEP_EXECUTION_ID";
private static final String GET_STEP_EXECUTION = GET_RAW_STEP_EXECUTIONS + " and STEP_NAME = ?";
private static final String GET_STEP_EXECUTION = GET_RAW_STEP_EXECUTIONS + " and STEP_EXECUTION_ID = ?";
private static final String CURRENT_VERSION_STEP_EXECUTION = "SELECT VERSION FROM %PREFIX%STEP_EXECUTION WHERE STEP_EXECUTION_ID=?";
@@ -190,9 +190,9 @@ public class JdbcStepExecutionDao extends AbstractJdbcBatchMetadataDao implement
}
}
public StepExecution getStepExecution(JobExecution jobExecution, String stepName) {
public StepExecution getStepExecution(JobExecution jobExecution, Long stepExecutionId) {
List<StepExecution> executions = getJdbcTemplate().query(getQuery(GET_STEP_EXECUTION),
new StepExecutionRowMapper(jobExecution), jobExecution.getId(), stepName);
new StepExecutionRowMapper(jobExecution), jobExecution.getId(), stepExecutionId);
Assert.state(executions.size() <= 1,
"There can be at most one step execution with given name for single job execution");

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

@@ -34,15 +34,20 @@ import org.springframework.util.Assert;
*/
public class MapStepExecutionDao implements StepExecutionDao {
private static Map<Long, Map<String, StepExecution>> executionsByJobExecutionId = TransactionAwareProxyFactory.createTransactionalMap();
private static Map<Long, Map<Long, StepExecution>> executionsByJobExecutionId = TransactionAwareProxyFactory
.createTransactionalMap();
private static Map<Long, StepExecution> executionsByStepExecutionId = TransactionAwareProxyFactory
.createTransactionalMap();
private static long currentId = 0;
public static void clear() {
executionsByJobExecutionId.clear();
executionsByStepExecutionId.clear();
}
private static StepExecution copy(StepExecution original){
private static StepExecution copy(StepExecution original) {
return (StepExecution) SerializationUtils.deserialize(SerializationUtils.serialize(original));
}
@@ -51,47 +56,48 @@ public class MapStepExecutionDao implements StepExecutionDao {
Assert.isTrue(stepExecution.getVersion() == null);
Assert.notNull(stepExecution.getJobExecutionId(), "JobExecution must be saved already.");
Map<String, StepExecution> executions = executionsByJobExecutionId.get(stepExecution.getJobExecutionId());
Map<Long, StepExecution> executions = executionsByJobExecutionId.get(stepExecution.getJobExecutionId());
if (executions == null) {
executions = TransactionAwareProxyFactory.createTransactionalMap();
executionsByJobExecutionId.put(stepExecution.getJobExecutionId(), executions);
}
stepExecution.setId(currentId++);
stepExecution.incrementVersion();
executions.put(stepExecution.getStepName(), copy(stepExecution));
StepExecution copy = copy(stepExecution);
executions.put(stepExecution.getId(), copy);
executionsByStepExecutionId.put(stepExecution.getId(), copy);
}
public void updateStepExecution(StepExecution stepExecution) {
Assert.notNull(stepExecution.getJobExecutionId());
Map<String, StepExecution> executions = executionsByJobExecutionId.get(stepExecution.getJobExecutionId());
Map<Long, StepExecution> executions = executionsByJobExecutionId.get(stepExecution.getJobExecutionId());
Assert.notNull(executions, "step executions for given job execution are expected to be already saved");
StepExecution persistedExecution = (StepExecution) executions.get(stepExecution.getStepName());
StepExecution persistedExecution = executionsByStepExecutionId.get(stepExecution.getId());
Assert.notNull(persistedExecution, "step execution is expected to be already saved");
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());
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(), copy(stepExecution));
StepExecution copy = copy(stepExecution);
executions.put(stepExecution.getId(), copy);
executionsByStepExecutionId.put(stepExecution.getId(), copy);
}
}
public StepExecution getStepExecution(JobExecution jobExecution, String stepName) {
Map<String, StepExecution> executions = executionsByJobExecutionId.get(jobExecution.getId());
if (executions == null) {
return null;
}
return copy(executions.get(stepName));
public StepExecution getStepExecution(JobExecution jobExecution, Long stepExecutionId) {
return executionsByStepExecutionId.get(stepExecutionId);
}
public List<StepExecution> getStepExecutions(JobExecution jobExecution) {
Map<String, StepExecution> executions = executionsByJobExecutionId.get(jobExecution.getId());
Map<Long, StepExecution> executions = executionsByJobExecutionId.get(jobExecution.getId());
if (executions == null || executions.isEmpty()) {
return Collections.emptyList();
}
@@ -102,9 +108,9 @@ public class MapStepExecutionDao implements StepExecutionDao {
return Long.signum(o2.getId() - o1.getId());
}
});
List<StepExecution> copy = new ArrayList<StepExecution>(result.size());
for(StepExecution exec : result) {
for (StepExecution exec : result) {
copy.add(copy(exec));
}
return copy;

9
spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/StepExecutionDao.java
View File

@@ -28,14 +28,13 @@ public interface StepExecutionDao {
void updateStepExecution(StepExecution stepExecution);
/**
* Retrieve a {@link StepExecution} from its parent {@link JobExecution} and
* step name.
* Retrieve a {@link StepExecution} from its id.
*
* @param jobExecution the parent job execution
* @param stepName the name of the step that was used to create the step execution
* @param jobExecution the parent {@link JobExecution}
* @param stepExecutionId the step execution id
* @return a {@link StepExecution}
*/
StepExecution getStepExecution(JobExecution jobExecution, String stepName);
StepExecution getStepExecution(JobExecution jobExecution, Long stepExecutionId);
/**
* Retrieve all the {@link StepExecution} for the parent {@link JobExecution}.

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

@@ -85,9 +85,9 @@ public class SimpleJobRepository implements JobRepository {
* @see JobRepository#isJobInstanceExists(String, JobParameters)
*/
public boolean isJobInstanceExists(String jobName, JobParameters jobParameters) {
return jobInstanceDao.getJobInstance(jobName, jobParameters)!=null;
return jobInstanceDao.getJobInstance(jobName, jobParameters) != null;
}
/**
* <p>
* Create a {@link JobExecution} based on the passed in {@link Job} and
@@ -244,7 +244,6 @@ public class SimpleJobRepository implements JobRepository {
public void update(StepExecution stepExecution) {
validateStepExecution(stepExecution);
Assert.notNull(stepExecution.getId(), "StepExecution must already be saved (have an id assigned)");
stepExecution.setLastUpdated(new Date(System.currentTimeMillis()));
stepExecutionDao.updateStepExecution(stepExecution);
@@ -277,9 +276,11 @@ public class SimpleJobRepository implements JobRepository {
List<JobExecution> jobExecutions = jobExecutionDao.findJobExecutions(jobInstance);
List<StepExecution> stepExecutions = new ArrayList<StepExecution>(jobExecutions.size());
for (JobExecution jobExecution : jobExecutions) {
StepExecution stepExecution = stepExecutionDao.getStepExecution(jobExecution, stepName);
if (stepExecution != null) {
stepExecutions.add(stepExecution);
List<StepExecution> allStepExecutions = stepExecutionDao.getStepExecutions(jobExecution);
for (StepExecution stepExecution : allStepExecutions) {
if (stepName.equals(stepExecution.getStepName())) {
stepExecutions.add(stepExecution);
}
}
}
StepExecution latest = null;
@@ -305,24 +306,28 @@ public class SimpleJobRepository implements JobRepository {
int count = 0;
List<JobExecution> jobExecutions = jobExecutionDao.findJobExecutions(jobInstance);
for (JobExecution jobExecution : jobExecutions) {
if (stepExecutionDao.getStepExecution(jobExecution, stepName) != null) {
count++;
List<StepExecution> allStepExecutions = stepExecutionDao.getStepExecutions(jobExecution);
for (StepExecution stepExecution : allStepExecutions) {
if (stepName.equals(stepExecution.getStepName())) {
count++;
}
}
}
return count;
}
/*
* Check to determine whether or not the JobExecution that is the parent of the provided
* StepExecution has been interrupted. If, after synchronizing the status with the database,
* the status has been updated to STOPPING, then the job has been interrupted.
/*
* Check to determine whether or not the JobExecution that is the parent of
* the provided StepExecution has been interrupted. If, after synchronizing
* the status with the database, the status has been updated to STOPPING,
* then the job has been interrupted.
*
* @param stepExecution
*/
private void checkForInterruption(StepExecution stepExecution){
private void checkForInterruption(StepExecution stepExecution) {
JobExecution jobExecution = stepExecution.getJobExecution();
jobExecutionDao.synchronizeStatus(jobExecution);
if(jobExecution.getStatus() == BatchStatus.STOPPING){
if (jobExecution.getStatus() == BatchStatus.STOPPING) {
stepExecution.setTerminateOnly();
}
}
@@ -333,7 +338,7 @@ public class SimpleJobRepository implements JobRepository {
return null;
}
return jobExecutionDao.getLastJobExecution(jobInstance);
}
}

4
spring-batch-core/src/test/java/org/springframework/batch/core/exlore/support/SimpleJobExplorerTests.java
View File

@@ -86,10 +86,10 @@ public class SimpleJobExplorerTests extends TestCase {
@Test
public void testGetStepExecution() throws Exception {
expect(jobExecutionDao.getJobExecution(123L)).andReturn(jobExecution);
expect(stepExecutionDao.getStepExecution(jobExecution, "foo")).andReturn(null);
expect(stepExecutionDao.getStepExecution(jobExecution, 123L)).andReturn(null);
expect(stepExecutionDao.getStepExecutions(jobExecution)).andReturn(null);
replay(jobExecutionDao, stepExecutionDao);
jobExplorer.getStepExecution(123L,"foo");
jobExplorer.getStepExecution(jobExecution.getId(), 123L);
verify(jobExecutionDao, stepExecutionDao);
}

8
spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/AbstractStepExecutionDaoTests.java
View File

@@ -104,7 +104,7 @@ public abstract class AbstractStepExecutionDaoTests extends AbstractTransactiona
stepExecution.setWriteCount(13);
dao.saveStepExecution(stepExecution);
StepExecution retrieved = dao.getStepExecution(jobExecution, step.getName());
StepExecution retrieved = dao.getStepExecution(jobExecution, stepExecution.getId());
assertStepExecutionsAreEqual(stepExecution, retrieved);
assertNotNull(retrieved.getVersion());
@@ -118,7 +118,7 @@ public abstract class AbstractStepExecutionDaoTests extends AbstractTransactiona
@Transactional
@Test
public void testSaveAndGetNonExistentExecution() {
assertNull(dao.getStepExecution(jobExecution, "not-existing step"));
assertNull(dao.getStepExecution(jobExecution, 45677L));
}
@Transactional
@@ -138,7 +138,7 @@ public abstract class AbstractStepExecutionDaoTests extends AbstractTransactiona
@Transactional
@Test
public void testGetForNotExistingJobExecution() {
assertNull(dao.getStepExecution(new JobExecution(jobInstance, (long) 777), step.getName()));
assertNull(dao.getStepExecution(new JobExecution(jobInstance, (long) 777), 11L));
}
/**
@@ -189,7 +189,7 @@ public abstract class AbstractStepExecutionDaoTests extends AbstractTransactiona
dao.updateStepExecution(stepExecution);
assertEquals(versionAfterSave + 1, stepExecution.getVersion().intValue());
StepExecution retrieved = dao.getStepExecution(jobExecution, step.getName());
StepExecution retrieved = dao.getStepExecution(jobExecution, stepExecution.getId());
assertEquals(stepExecution, retrieved);
assertEquals(stepExecution.getLastUpdated(), retrieved.getLastUpdated());
assertEquals(BatchStatus.STOPPED, retrieved.getStatus());

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

@@ -44,14 +44,14 @@ public class JdbcStepExecutionDaoTests extends AbstractStepExecutionDaoTests {
((JdbcStepExecutionDao) dao).setExitMessageLength(250);
dao.saveStepExecution(stepExecution);
StepExecution retrievedAfterSave = dao.getStepExecution(jobExecution, step.getName());
StepExecution retrievedAfterSave = dao.getStepExecution(jobExecution, stepExecution.getId());
assertTrue("Exit description should be truncated", retrievedAfterSave.getExitStatus().getExitDescription()
.length() < stepExecution.getExitStatus().getExitDescription().length());
dao.updateStepExecution(stepExecution);
StepExecution retrievedAfterUpdate = dao.getStepExecution(jobExecution, step.getName());
StepExecution retrievedAfterUpdate = dao.getStepExecution(jobExecution, stepExecution.getId());
assertTrue("Exit description should be truncated", retrievedAfterUpdate.getExitStatus().getExitDescription()
.length() < stepExecution.getExitStatus().getExitDescription().length());

12
spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/MapStepExecutionDaoTests.java
View File

@@ -1,16 +1,16 @@
package org.springframework.batch.core.repository.dao;
import static org.junit.Assert.*;
import static org.junit.Assert.assertNull;
import java.util.Date;
import org.junit.Test;
import org.junit.internal.runners.JUnit4ClassRunner;
import org.junit.runner.RunWith;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.SimpleJobRepository;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.internal.runners.JUnit4ClassRunner;
@RunWith(JUnit4ClassRunner.class)
public class MapStepExecutionDaoTests extends AbstractStepExecutionDaoTests {
@@ -40,14 +40,14 @@ public class MapStepExecutionDaoTests extends AbstractStepExecutionDaoTests {
tested.saveStepExecution(stepExecution);
stepExecution.setEndTime(new Date());
StepExecution retrieved = tested.getStepExecution(jobExecution, "stepName");
StepExecution retrieved = tested.getStepExecution(jobExecution, stepExecution.getId());
assertNull(retrieved.getEndTime());
stepExecution.setEndTime(null);
tested.updateStepExecution(stepExecution);
stepExecution.setEndTime(new Date());
StepExecution stored = tested.getStepExecution(jobExecution, "stepName");
StepExecution stored = tested.getStepExecution(jobExecution, stepExecution.getId());
assertNull(stored.getEndTime());
}