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

Sql and Hibernate JobDao implementations are now able to cope with Simple- and ScheduledJobIdentifier strategies

Browse Source 
RESOLVED - issue BATCH-58: Make JobStream and JobRun optional properties of Job (e.g. through join to optional table) 
http://opensource.atlassian.com/projects/spring/browse/BATCH-58
This commit is contained in:
dsyer
2007-08-28 11:28:17 +00:00
parent 4c2d473874
commit 61249e9702
3 changed files with 202 additions and 143 deletions
Download Patch File Download Diff File Expand all files Collapse all files

135
execution/src/main/java/org/springframework/batch/execution/repository/dao/SqlJobDao.java
View File

@@ -34,12 +34,13 @@ import org.springframework.util.Assert;
/**
* SQL implementation of {@link JobDao}. Uses sequences (via Spring's
*
* @link DataFieldMaxValueIncrementer abstraction) to create all primary keys
* before inserting a new row. Objects are checked to ensure all mandatory
* fields to be stored are not null. If any are found to be null, an
* IllegalArgumentException will be thrown. This could be left to JdbcTemplate,
* however, the exception will be fairly vague, and fails to highlight which
* field caused the exception.
* before inserting a new row. Objects are checked to ensure all mandatory
* fields to be stored are not null. If any are found to be null, an
* IllegalArgumentException will be thrown. This could be left to
* JdbcTemplate, however, the exception will be fairly vague, and fails to
* highlight which field caused the exception.
*
* @author Lucas Ward
* @author Dave Syer
@@ -83,21 +84,25 @@ public class SqlJobDao implements JobDao, InitializingBean {
* date) into an INSERT statement.
*
* @see JobDao#createJob(JobIdentifier)
* @throws IllegalArgumentException if any JobRuntimeInformation fields are
* null.
* @throws IllegalArgumentException
* if any JobRuntimeInformation fields are null.
*/
public JobInstance createJob(JobIdentifier jobIdentifier) {
ScheduledJobIdentifier jobRuntimeInformation = (ScheduledJobIdentifier) jobIdentifier;
validateJobRuntimeInformation(jobRuntimeInformation);
validateJobIdentifier(jobIdentifier);
ScheduledJobIdentifier defaultJobId = getScheduledJobIdentifier(jobIdentifier);
Long jobId = new Long(jobIncrementer.nextLongValue());
Object[] parameters = new Object[] { jobId, jobRuntimeInformation.getName(),
jobRuntimeInformation.getJobStream(), jobRuntimeInformation.getScheduleDate(),
new Long(jobRuntimeInformation.getJobRun()) };
Object[] parameters = new Object[] { jobId,
defaultJobId.getName(),
defaultJobId.getJobStream(),
defaultJobId.getScheduleDate(),
new Long(defaultJobId.getJobRun()) };
jdbcTemplate.update(CREATE_JOB, parameters);
JobInstance job = new JobInstance(jobId);
job.setIdentifier(jobIdentifier);
return job;
}
@@ -106,16 +111,18 @@ public class SqlJobDao implements JobDao, InitializingBean {
* the given identifier, adding them to a list via the RowMapper callback.
*
* @see JobDao#findJobs(JobIdentifier)
* @throws IllegalArgumentException if any JobRuntimeInformation fields are
* null.
* @throws IllegalArgumentException
* if any JobRuntimeInformation fields are null.
*/
public List findJobs(final JobIdentifier jobIdentifier) {
ScheduledJobIdentifier defaultJobId = (ScheduledJobIdentifier) jobIdentifier;
validateJobRuntimeInformation(defaultJobId);
validateJobIdentifier(jobIdentifier);
Object[] parameters = new Object[] { defaultJobId.getName(), defaultJobId.getJobStream(),
defaultJobId.getScheduleDate(), new Integer(defaultJobId.getJobRun()) };
ScheduledJobIdentifier defaultJobId = getScheduledJobIdentifier(jobIdentifier);
Object[] parameters = new Object[] { defaultJobId.getName(),
defaultJobId.getJobStream(), defaultJobId.getScheduleDate(),
new Integer(defaultJobId.getJobRun()) };
RowMapper rowMapper = new RowMapper() {
public Object mapRow(ResultSet rs, int rowNum) throws SQLException {
@@ -133,7 +140,8 @@ public class SqlJobDao implements JobDao, InitializingBean {
/**
* @see JobDao#update(JobInstance)
* @throws IllegalArgumentException if Job, Job.status, or job.id is null
* @throws IllegalArgumentException
* if Job, Job.status, or job.id is null
*/
public void update(JobInstance job) {
@@ -141,7 +149,8 @@ public class SqlJobDao implements JobDao, InitializingBean {
Assert.notNull(job.getStatus(), "Job Status cannot be Null");
Assert.notNull(job.getId(), "Job ID cannot be null");
Object[] parameters = new Object[] { job.getStatus().toString(), job.getId() };
Object[] parameters = new Object[] { job.getStatus().toString(),
job.getId() };
jdbcTemplate.update(UPDATE_JOB, parameters);
}
@@ -152,16 +161,18 @@ public class SqlJobDao implements JobDao, InitializingBean {
* via a SQL INSERT statement.
*
* @see JobDao#save(JobExecution)
* @throws IllegalArgumentException if jobExecution is null, as well as any
* of it's fields to be persisted.
* @throws IllegalArgumentException
* if jobExecution is null, as well as any of it's fields to be
* persisted.
*/
public void save(JobExecution jobExecution) {
validateJobExecution(jobExecution);
jobExecution.setId(new Long(jobExecutionIncrementer.nextLongValue()));
Object[] parameters = new Object[] { jobExecution.getId(), jobExecution.getJobId(),
jobExecution.getStartTime(), jobExecution.getEndTime(), jobExecution.getStatus().toString() };
Object[] parameters = new Object[] { jobExecution.getId(),
jobExecution.getJobId(), jobExecution.getStartTime(),
jobExecution.getEndTime(), jobExecution.getStatus().toString() };
jdbcTemplate.update(SAVE_JOB_EXECUTION, parameters);
}
@@ -177,20 +188,24 @@ public class SqlJobDao implements JobDao, InitializingBean {
validateJobExecution(jobExecution);
Object[] parameters = new Object[] { jobExecution.getStartTime(), jobExecution.getEndTime(),
jobExecution.getStatus().toString(), jobExecution.getId() };
Object[] parameters = new Object[] { jobExecution.getStartTime(),
jobExecution.getEndTime(), jobExecution.getStatus().toString(),
jobExecution.getId() };
if (jobExecution.getId() == null) {
throw new IllegalArgumentException("JobExecution ID cannot be null. JobExecution must be saved "
+ "before it can be updated.");
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 (jdbcTemplate.queryForInt(CHECK_JOB_EXECUTION_EXISTS, new Object[] { jobExecution.getId() }) != 1) {
throw new NoSuchBatchDomainObjectException("Invalid JobExecution, ID " + jobExecution.getId()
+ " not found.");
if (jdbcTemplate.queryForInt(CHECK_JOB_EXECUTION_EXISTS,
new Object[] { jobExecution.getId() }) != 1) {
throw new NoSuchBatchDomainObjectException(
"Invalid JobExecution, ID " + jobExecution.getId()
+ " not found.");
}
jdbcTemplate.update(UPDATE_JOB_EXECUTION, parameters);
@@ -198,7 +213,8 @@ public class SqlJobDao implements JobDao, InitializingBean {
/**
* @see JobDao#getJobExecutionCount(JobInstance)
* @throws IllegalArgumentException if jobId is null.
* @throws IllegalArgumentException
* if jobId is null.
*/
public int getJobExecutionCount(Long jobId) {
@@ -231,7 +247,8 @@ public class SqlJobDao implements JobDao, InitializingBean {
};
return jdbcTemplate.query(FIND_JOB_EXECUTIONS, new Object[] { jobId }, rowMapper);
return jdbcTemplate.query(FIND_JOB_EXECUTIONS, new Object[] { jobId },
rowMapper);
}
public void setJdbcTemplate(JdbcTemplate jdbcTemplate) {
@@ -242,12 +259,14 @@ public class SqlJobDao implements JobDao, InitializingBean {
this.jobIncrementer = jobIncrementer;
}
public void setJobExecutionIncrementer(DataFieldMaxValueIncrementer jobExecutionIncrementer) {
public void setJobExecutionIncrementer(
DataFieldMaxValueIncrementer jobExecutionIncrementer) {
this.jobExecutionIncrementer = jobExecutionIncrementer;
}
/*
* (non-Javadoc)
*
* @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
*
* Ensure jdbcTemplate and incrementers have been provided.
@@ -256,7 +275,8 @@ public class SqlJobDao implements JobDao, InitializingBean {
Assert.notNull(jdbcTemplate, "JdbcTemplate cannot be null");
Assert.notNull(jobIncrementer, "JobIncrementor cannot be null");
Assert.notNull(jobExecutionIncrementer, "JobExecutionIncrementer cannot be null");
Assert.notNull(jobExecutionIncrementer,
"JobExecutionIncrementer cannot be null");
}
/*
@@ -268,24 +288,49 @@ public class SqlJobDao implements JobDao, InitializingBean {
private void validateJobExecution(JobExecution jobExecution) {
Assert.notNull(jobExecution);
Assert.notNull(jobExecution.getJobId(), "JobExecution Job-Id cannot be null.");
Assert.notNull(jobExecution.getStartTime(), "JobExecution start time cannot be null.");
Assert.notNull(jobExecution.getStatus(), "JobExecution status cannot be null.");
Assert.notNull(jobExecution.getJobId(),
"JobExecution Job-Id cannot be null.");
Assert.notNull(jobExecution.getStartTime(),
"JobExecution start time cannot be null.");
Assert.notNull(jobExecution.getStatus(),
"JobExecution status cannot be null.");
}
/*
* Validate JobRuntimeInformation. Due to differing requirements, it is
* acceptable for any field to be blank, however null fields may cause odd
* and vague exception reports from the database driver.
*
* TODO: remove dependency on ScheduledJobIdentifier
*/
private void validateJobRuntimeInformation(ScheduledJobIdentifier jobRuntimeInformation) {
private void validateJobIdentifier(JobIdentifier jobIdentifier) {
Assert.notNull(jobRuntimeInformation, "JobRuntimeInformation cannot be null.");
Assert.notNull(jobRuntimeInformation.getName(), "JobRuntimeInformation name cannot be null.");
Assert.notNull(jobRuntimeInformation.getJobStream(), "JobRuntimeInformation JobStream cannot be null.");
Assert.notNull(jobRuntimeInformation.getScheduleDate(), "JobRuntimeInformation ScheduleDate cannot be null.");
Assert.notNull(jobIdentifier, "JobRuntimeInformation cannot be null.");
Assert.notNull(jobIdentifier.getName(),
"JobRuntimeInformation name cannot be null.");
if (jobIdentifier instanceof ScheduledJobIdentifier) {
ScheduledJobIdentifier jobRuntimeInformation = (ScheduledJobIdentifier) jobIdentifier;
Assert.notNull(jobRuntimeInformation.getJobStream(),
"JobRuntimeInformation JobStream cannot be null.");
Assert.notNull(jobRuntimeInformation.getScheduleDate(),
"JobRuntimeInformation ScheduleDate cannot be null.");
}
}
/**
* Convert a {@link JobIdentifier} to a {@link ScheduledJobIdentifier} by
* supplying additional fields with null values, as necessary.
*
* @param jobIdentifier
* a {@link JobIdentifier}
* @return a {@link ScheduledJobIdentifier} with the same name
*/
private ScheduledJobIdentifier getScheduledJobIdentifier(
JobIdentifier jobIdentifier) {
if (jobIdentifier instanceof ScheduledJobIdentifier) {
return (ScheduledJobIdentifier) jobIdentifier;
}
return new ScheduledJobIdentifier(jobIdentifier.getName());
}
}

15
execution/src/test/java/org/springframework/batch/execution/repository/dao/HibernateJobDaoTests.java
View File

@@ -22,7 +22,6 @@ import java.util.Map;
import org.hibernate.SessionFactory;
import org.springframework.batch.core.domain.BatchStatus;
import org.springframework.batch.core.runtime.SimpleJobIdentifier;
import org.springframework.util.ClassUtils;
public class HibernateJobDaoTests extends SqlJobDaoTests {
@@ -50,18 +49,4 @@ public class HibernateJobDaoTests extends SqlJobDaoTests {
assertEquals(jobExecution.getEndTime(), ((Map)executions.get(0)).get("END_TIME"));
}
public void testJobWithSimpleJobIdentifier() throws Exception {
SimpleJobIdentifier jobIdentifier = new SimpleJobIdentifier("Job1");
// Create job.
job = jobDao.createJob(jobIdentifier);
sessionFactory.getCurrentSession().flush();
List jobs = jdbcTemplate.queryForList("SELECT * FROM BATCH_JOB where ID=?", new Object[] {job.getId()});
assertEquals(1, jobs.size());
assertEquals(job.getName(), ((Map)jobs.get(0)).get("JOB_NAME"));
}
}

195
execution/src/test/java/org/springframework/batch/execution/repository/dao/SqlJobDaoTests.java
View File

@@ -21,33 +21,38 @@ import java.sql.SQLException;
import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.util.List;
import java.util.Map;
import org.springframework.batch.core.domain.BatchStatus;
import org.springframework.batch.core.domain.JobExecution;
import org.springframework.batch.core.domain.JobInstance;
import org.springframework.batch.core.repository.NoSuchBatchDomainObjectException;
import org.springframework.batch.core.runtime.SimpleJobIdentifier;
import org.springframework.batch.execution.runtime.ScheduledJobIdentifier;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.test.AbstractTransactionalDataSourceSpringContextTests;
import org.springframework.util.ClassUtils;
public class SqlJobDaoTests extends AbstractTransactionalDataSourceSpringContextTests {
public class SqlJobDaoTests extends
AbstractTransactionalDataSourceSpringContextTests {
private static final String GET_JOB_EXECUTION = "SELECT JOB_ID, START_TIME, END_TIME, STATUS from "
+ "BATCH_JOB_EXECUTION where ID = ?";
private static final String GET_JOB_EXECUTION = "SELECT JOB_ID, START_TIME, END_TIME, STATUS from " +
"BATCH_JOB_EXECUTION where ID = ?";
protected JobDao jobDao;
protected ScheduledJobIdentifier jobRuntimeInformation;
protected JobInstance job;
protected JobExecution jobExecution;
protected Timestamp jobExecutionStartTime = new Timestamp(System.currentTimeMillis());
protected Timestamp jobExecutionStartTime = new Timestamp(System
.currentTimeMillis());
protected String[] getConfigLocations() {
return new String[] { ClassUtils.addResourcePathToPackagePath(getClass(), "sql-dao-test.xml") };
return new String[] { ClassUtils.addResourcePathToPackagePath(
getClass(), "sql-dao-test.xml") };
}
/*
@@ -57,17 +62,18 @@ public class SqlJobDaoTests extends AbstractTransactionalDataSourceSpringContext
public void setJobRepositoryDao(JobDao jobRepositoryDao) {
this.jobDao = jobRepositoryDao;
}
protected void onSetUpInTransaction() throws Exception {
jobRuntimeInformation = new ScheduledJobIdentifier("Job1");
jobRuntimeInformation.setName("Job1");
jobRuntimeInformation.setJobStream("TestStream");
jobRuntimeInformation.setJobRun(1);
jobRuntimeInformation.setScheduleDate(new SimpleDateFormat("yyyyMMdd").parse("20070505"));
jobRuntimeInformation.setScheduleDate(new SimpleDateFormat("yyyyMMdd")
.parse("20070505"));
// Create job.
job = jobDao.createJob(jobRuntimeInformation);
// Create an execution
jobExecutionStartTime = new Timestamp(System.currentTimeMillis());
jobExecution = new JobExecution(job.getId());
@@ -75,49 +81,53 @@ public class SqlJobDaoTests extends AbstractTransactionalDataSourceSpringContext
jobExecution.setStatus(BatchStatus.STARTED);
jobDao.save(jobExecution);
}
public void testVersionIsNotNullForJob() throws Exception {
int version = jdbcTemplate.queryForInt("select version from BATCH_JOB where ID="+job.getId());
assertEquals(0, version);
}
public void testVersionIsNotNullForJobExecution() throws Exception {
int version = jdbcTemplate.queryForInt("select version from BATCH_JOB_EXECUTION where ID="+jobExecution.getId());
int version = jdbcTemplate
.queryForInt("select version from BATCH_JOB where ID="
+ job.getId());
assertEquals(0, version);
}
public void testFindNonExistentJob(){
public void testVersionIsNotNullForJobExecution() throws Exception {
int version = jdbcTemplate
.queryForInt("select version from BATCH_JOB_EXECUTION where ID="
+ jobExecution.getId());
assertEquals(0, version);
}
public void testFindNonExistentJob() {
// No job should be found since it hasn't been created.
List jobs = jobDao.findJobs(new ScheduledJobIdentifier("Job2"));
assertTrue(jobs.size() == 0);
}
public void testFindJob(){
public void testFindJob() {
List jobs = jobDao.findJobs(jobRuntimeInformation);
assertTrue(jobs.size() == 1);
JobInstance tempJob = (JobInstance) jobs.get(0);
assertTrue(job.equals(tempJob));
assertEquals(jobRuntimeInformation, tempJob.getIdentifier());
}
public void testFindJobWithNullRuntime(){
public void testFindJobWithNullRuntime() {
ScheduledJobIdentifier runtimeInformation = null;
try{
try {
jobDao.findJobs(runtimeInformation);
fail();
}catch(IllegalArgumentException ex){
//expected
} catch (IllegalArgumentException ex) {
// expected
}
}
public void testUpdateJob(){
public void testUpdateJob() {
// Update the returned job with a new status
job.setStatus(BatchStatus.COMPLETED);
jobDao.update(job);
// The job just updated should be found, with the saved status.
List jobs = jobDao.findJobs(jobRuntimeInformation);
assertTrue(jobs.size() == 1);
@@ -125,25 +135,25 @@ public class SqlJobDaoTests extends AbstractTransactionalDataSourceSpringContext
assertTrue(job.equals(tempJob));
assertEquals(tempJob.getStatus(), BatchStatus.COMPLETED);
}
public void testUpdateJobWithNullId(){
public void testUpdateJobWithNullId() {
JobInstance testJob = new JobInstance(null);
try{
try {
jobDao.update(testJob);
fail();
}catch(IllegalArgumentException ex){
//expected
} catch (IllegalArgumentException ex) {
// expected
}
}
public void testUpdateNullJob(){
JobInstance testJob = null;
try{
public void testUpdateNullJob() {
JobInstance testJob = null;
try {
jobDao.update(testJob);
}catch(IllegalArgumentException ex){
//expected
} catch (IllegalArgumentException ex) {
// expected
}
}
@@ -152,71 +162,90 @@ public class SqlJobDaoTests extends AbstractTransactionalDataSourceSpringContext
jobExecution.setStatus(BatchStatus.COMPLETED);
jobExecution.setEndTime(new Timestamp(System.currentTimeMillis()));
jobDao.update(jobExecution);
List executions = retrieveJobExecution(jobExecution.getId());
assertEquals(executions.size(), 1);
assertEquals(jobExecution, ((JobExecution)executions.get(0)));
assertEquals(jobExecution, ((JobExecution) executions.get(0)));
}
public void testUpdateInvalidJobExecution(){
public void testUpdateInvalidJobExecution() {
JobExecution execution = new JobExecution(job.getId());
//id is invalid
// id is invalid
execution.setId(new Long(29432));
try{
try {
jobDao.update(execution);
fail();
}catch(NoSuchBatchDomainObjectException ex){
//expected
} catch (NoSuchBatchDomainObjectException ex) {
// expected
}
}
public void testUpdateNullIdJobExection(){
public void testUpdateNullIdJobExection() {
JobExecution execution = new JobExecution(job.getId());
try{
try {
jobDao.update(execution);
fail();
}catch(IllegalArgumentException ex){
//expected
} catch (IllegalArgumentException ex) {
// expected
}
}
public void testIncrementExecutionCount(){
public void testIncrementExecutionCount() {
// 1 JobExection already added in setup
assertEquals(jobDao.getJobExecutionCount(job.getId()), 1);
// Save new JobExecution for same job
JobExecution testJobExecution = new JobExecution(job.getId());
jobDao.save(testJobExecution);
//JobExecutionCount should be incremented by 1
// JobExecutionCount should be incremented by 1
assertEquals(jobDao.getJobExecutionCount(job.getId()), 2);
}
public void testZeroExecutionCount(){
JobInstance testJob = jobDao.createJob(new ScheduledJobIdentifier("TestJob"));
//no jobExecutions saved for new job, count should be 0
public void testZeroExecutionCount() {
JobInstance testJob = jobDao.createJob(new ScheduledJobIdentifier(
"TestJob"));
// no jobExecutions saved for new job, count should be 0
assertEquals(jobDao.getJobExecutionCount(testJob.getId()), 0);
}
private List retrieveJobExecution(final Long id){
RowMapper rowMapper = new RowMapper(){
public void testJobWithSimpleJobIdentifier() throws Exception {
SimpleJobIdentifier jobIdentifier = new SimpleJobIdentifier("Job1");
// Create job.
job = jobDao.createJob(jobIdentifier);
// sessionFactory.getCurrentSession().flush();
List jobs = jdbcTemplate.queryForList(
"SELECT * FROM BATCH_JOB where ID=?", new Object[] { job
.getId() });
assertEquals(1, jobs.size());
assertEquals(job.getName(), ((Map) jobs.get(0)).get("JOB_NAME"));
}
private List retrieveJobExecution(final Long id) {
RowMapper rowMapper = new RowMapper() {
public Object mapRow(ResultSet rs, int rowNum) throws SQLException {
JobExecution execution = new JobExecution(new Long(rs.getLong(1)));
JobExecution execution = new JobExecution(new Long(rs
.getLong(1)));
execution.setStartTime(rs.getTimestamp(2));
execution.setEndTime(rs.getTimestamp(3));
execution.setStatus(BatchStatus.getStatus(rs.getString(4)));
execution.setId(id);
return execution;
}
};
return jdbcTemplate.query(GET_JOB_EXECUTION, new Object[]{id}, rowMapper);
return jdbcTemplate.query(GET_JOB_EXECUTION, new Object[] { id },
rowMapper);
}
}