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

Added tablePrefix property to SqlJobDao (defaults to "BATCH_").

Browse Source 
OPEN - issue BATCH-93: Need to be able to configure table names! 
http://opensource.atlassian.com/projects/spring/browse/BATCH-93
This commit is contained in:
dsyer
2007-08-28 13:16:21 +00:00
parent 26401ff11e
commit 26d808cb9e
3 changed files with 328 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -31,6 +31,7 @@ import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.jdbc.support.incrementer.DataFieldMaxValueIncrementer;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
* SQL implementation of {@link JobDao}. Uses sequences (via Spring's
@@ -47,28 +48,35 @@ import org.springframework.util.Assert;
*/
public class SqlJobDao implements JobDao, InitializingBean {
/**
* Default value for the table prefix property.
*/
public static final String DEFAULT_TABLE_PREFIX = "BATCH_";
private static String tablePrefix = DEFAULT_TABLE_PREFIX;
// Job SQL statements
private static final String CREATE_JOB = "INSERT into BATCH_JOB(ID, JOB_NAME, JOB_STREAM, SCHEDULE_DATE, JOB_RUN)"
private static final String CREATE_JOB = "INSERT into %PREFIX%JOB(ID, JOB_NAME, JOB_STREAM, SCHEDULE_DATE, JOB_RUN)"
+ " values (?, ?, ?, ?, ?)";
private static final String FIND_JOBS = "SELECT ID, STATUS from BATCH_JOB where JOB_NAME = ? and "
private static final String FIND_JOBS = "SELECT ID, STATUS from %PREFIX%JOB where JOB_NAME = ? and "
+ "JOB_STREAM = ? and SCHEDULE_DATE = ? and JOB_RUN = ?";
private static final String UPDATE_JOB = "UPDATE BATCH_JOB set STATUS = ? where ID = ?";
private static final String UPDATE_JOB = "UPDATE %PREFIX%JOB set STATUS = ? where ID = ?";
private static final String GET_JOB_EXECUTION_COUNT = "SELECT count(ID) from BATCH_JOB_EXECUTION "
private static final String GET_JOB_EXECUTION_COUNT = "SELECT count(ID) from %PREFIX%JOB_EXECUTION "
+ "where JOB_ID = ?";
// Job Execution SqlStatements
private static final String UPDATE_JOB_EXECUTION = "UPDATE BATCH_JOB_EXECUTION set START_TIME = ?, END_TIME = ?, "
private static final String UPDATE_JOB_EXECUTION = "UPDATE %PREFIX%JOB_EXECUTION set START_TIME = ?, END_TIME = ?, "
+ " STATUS = ? where ID = ?";
private static final String SAVE_JOB_EXECUTION = "INSERT into BATCH_JOB_EXECUTION(ID, JOB_ID, START_TIME, END_TIME, STATUS)"
private static final String SAVE_JOB_EXECUTION = "INSERT into %PREFIX%JOB_EXECUTION(ID, JOB_ID, START_TIME, END_TIME, STATUS)"
+ " values (?, ?, ?, ?, ?)";
private static final String CHECK_JOB_EXECUTION_EXISTS = "SELECT COUNT(*) FROM BATCH_JOB_EXECUTION WHERE ID=?";
private static final String CHECK_JOB_EXECUTION_EXISTS = "SELECT COUNT(*) FROM %PREFIX%JOB_EXECUTION WHERE ID=?";
private static final String FIND_JOB_EXECUTIONS = "SELECT ID, START_TIME, END_TIME, STATUS from BATCH_JOB_EXECUTION"
private static final String FIND_JOB_EXECUTIONS = "SELECT ID, START_TIME, END_TIME, STATUS from %PREFIX%JOB_EXECUTION"
+ " where JOB_ID = ?";
private JdbcTemplate jdbcTemplate;
@@ -77,6 +85,18 @@ public class SqlJobDao implements JobDao, InitializingBean {
private DataFieldMaxValueIncrementer jobExecutionIncrementer;
/**
* Public setter for the table prefix property. This will be prefixed to all
* the table names before queries are executed. Defaults to
* {@value #DEFAULT_TABLE_PREFIX}.
*
* @param tablePrefix
* the tablePrefix to set
*/
public void setTablePrefix(String tablePrefix) {
SqlJobDao.tablePrefix = tablePrefix;
}
/**
* In this sql implementation a job id is obtained by asking the
* jobIncrementer (which is likely a sequence) for the nextLong, and then
@@ -94,12 +114,10 @@ public class SqlJobDao implements JobDao, InitializingBean {
ScheduledJobIdentifier defaultJobId = getScheduledJobIdentifier(jobIdentifier);
Long jobId = new Long(jobIncrementer.nextLongValue());
Object[] parameters = new Object[] { jobId,
defaultJobId.getName(),
defaultJobId.getJobStream(),
defaultJobId.getScheduleDate(),
Object[] parameters = new Object[] { jobId, defaultJobId.getName(),
defaultJobId.getJobStream(), defaultJobId.getScheduleDate(),
new Long(defaultJobId.getJobRun()) };
jdbcTemplate.update(CREATE_JOB, parameters);
jdbcTemplate.update(getCreateJobQuery(), parameters);
JobInstance job = new JobInstance(jobId);
job.setIdentifier(jobIdentifier);
@@ -107,8 +125,8 @@ public class SqlJobDao implements JobDao, InitializingBean {
}
/**
* The BATCH_JOB table is queried for <strong>any</strong> jobs that match
* the given identifier, adding them to a list via the RowMapper callback.
* The job table is queried for <strong>any</strong> jobs that match the
* given identifier, adding them to a list via the RowMapper callback.
*
* @see JobDao#findJobs(JobIdentifier)
* @throws IllegalArgumentException
@@ -135,7 +153,7 @@ public class SqlJobDao implements JobDao, InitializingBean {
}
};
return jdbcTemplate.query(FIND_JOBS, parameters, rowMapper);
return jdbcTemplate.query(getFindJobsQuery(), parameters, rowMapper);
}
/**
@@ -151,7 +169,7 @@ public class SqlJobDao implements JobDao, InitializingBean {
Object[] parameters = new Object[] { job.getStatus().toString(),
job.getId() };
jdbcTemplate.update(UPDATE_JOB, parameters);
jdbcTemplate.update(getUpdateJobQuery(), parameters);
}
/**
@@ -173,7 +191,7 @@ public class SqlJobDao implements JobDao, InitializingBean {
Object[] parameters = new Object[] { jobExecution.getId(),
jobExecution.getJobId(), jobExecution.getStartTime(),
jobExecution.getEndTime(), jobExecution.getStatus().toString() };
jdbcTemplate.update(SAVE_JOB_EXECUTION, parameters);
jdbcTemplate.update(getSaveJobExecutionQuery(), parameters);
}
/**
@@ -201,14 +219,14 @@ public class SqlJobDao implements JobDao, InitializingBean {
// 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,
if (jdbcTemplate.queryForInt(getCheckJobExecutionExistsQuery(),
new Object[] { jobExecution.getId() }) != 1) {
throw new NoSuchBatchDomainObjectException(
"Invalid JobExecution, ID " + jobExecution.getId()
+ " not found.");
}
jdbcTemplate.update(UPDATE_JOB_EXECUTION, parameters);
jdbcTemplate.update(getUpdateJobExecutionQuery(), parameters);
}
/**
@@ -222,7 +240,7 @@ public class SqlJobDao implements JobDao, InitializingBean {
Object[] parameters = new Object[] { jobId };
return jdbcTemplate.queryForInt(GET_JOB_EXECUTION_COUNT, parameters);
return jdbcTemplate.queryForInt(getJobExecutionCountQuery(), parameters);
}
public List findJobExecutions(JobInstance job) {
@@ -247,10 +265,46 @@ public class SqlJobDao implements JobDao, InitializingBean {
};
return jdbcTemplate.query(FIND_JOB_EXECUTIONS, new Object[] { jobId },
return jdbcTemplate.query(getFindJobExecutionsQuery(), new Object[] { jobId },
rowMapper);
}
private String getQuery(String base) {
return StringUtils.replace(base, "%PREFIX%", tablePrefix);
}
private String getCreateJobQuery() {
return getQuery(CREATE_JOB);
}
private String getFindJobsQuery() {
return getQuery(FIND_JOBS);
}
private String getUpdateJobQuery() {
return getQuery(UPDATE_JOB);
}
private String getSaveJobExecutionQuery() {
return getQuery(SAVE_JOB_EXECUTION);
}
private String getUpdateJobExecutionQuery() {
return getQuery(UPDATE_JOB_EXECUTION);
}
private String getCheckJobExecutionExistsQuery() {
return getQuery(CHECK_JOB_EXECUTION_EXISTS);
}
private String getJobExecutionCountQuery() {
return getQuery(GET_JOB_EXECUTION_COUNT);
}
private String getFindJobExecutionsQuery() {
return getQuery(FIND_JOB_EXECUTIONS);
}
public void setJdbcTemplate(JdbcTemplate jdbcTemplate) {
this.jdbcTemplate = jdbcTemplate;
}

251
execution/src/test/java/org/springframework/batch/execution/repository/dao/BaseJobDaoTests.java Normal file
View File

@@ -0,0 +1,251 @@
/*
* Copyright 2006-2007 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.execution.repository.dao;
import java.sql.ResultSet;
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 BaseJobDaoTests extends
AbstractTransactionalDataSourceSpringContextTests {
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 String[] getConfigLocations() {
return new String[] { ClassUtils.addResourcePathToPackagePath(
getClass(), "sql-dao-test.xml") };
}
/*
* Because AbstractTransactionalSpringContextTests is used, this method will
* be called by Spring to set the JobRepository.
*/
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"));
// Create job.
job = jobDao.createJob(jobRuntimeInformation);
// Create an execution
jobExecutionStartTime = new Timestamp(System.currentTimeMillis());
jobExecution = new JobExecution(job.getId());
jobExecution.setStartTime(jobExecutionStartTime);
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());
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() {
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() {
ScheduledJobIdentifier runtimeInformation = null;
try {
jobDao.findJobs(runtimeInformation);
fail();
} catch (IllegalArgumentException ex) {
// expected
}
}
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);
JobInstance tempJob = (JobInstance) jobs.get(0);
assertTrue(job.equals(tempJob));
assertEquals(tempJob.getStatus(), BatchStatus.COMPLETED);
}
public void testUpdateJobWithNullId() {
JobInstance testJob = new JobInstance(null);
try {
jobDao.update(testJob);
fail();
} catch (IllegalArgumentException ex) {
// expected
}
}
public void testUpdateNullJob() {
JobInstance testJob = null;
try {
jobDao.update(testJob);
} catch (IllegalArgumentException ex) {
// expected
}
}
public void testUpdateJobExecution() {
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)));
}
public void testUpdateInvalidJobExecution() {
JobExecution execution = new JobExecution(job.getId());
// id is invalid
execution.setId(new Long(29432));
try {
jobDao.update(execution);
fail();
} catch (NoSuchBatchDomainObjectException ex) {
// expected
}
}
public void testUpdateNullIdJobExection() {
JobExecution execution = new JobExecution(job.getId());
try {
jobDao.update(execution);
fail();
} catch (IllegalArgumentException ex) {
// expected
}
}
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
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
assertEquals(jobDao.getJobExecutionCount(testJob.getId()), 0);
}
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)));
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);
}
}

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

@@ -24,7 +24,7 @@ import org.hibernate.SessionFactory;
import org.springframework.batch.core.domain.BatchStatus;
import org.springframework.util.ClassUtils;
public class HibernateJobDaoTests extends SqlJobDaoTests {
public class HibernateJobDaoTests extends BaseJobDaoTests {
private SessionFactory sessionFactory;