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

IN PROGRESS - BATCH-857: map daos need to be truly transactional for correct restart

Browse Source 
moved transaction manager and proxy up to AbstractJobRepository so that it applies to Map implementation too
This commit is contained in:
robokaso
2008-10-14 12:58:20 +00:00
parent eacf9a41a1
commit 9b1371f6fd
5 changed files with 92 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files

78
spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/AbstractJobRepositoryFactoryBean.java
View File

@@ -1,11 +1,19 @@
package org.springframework.batch.core.repository.support;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.aop.support.DefaultPointcutAdvisor;
import org.springframework.aop.support.NameMatchMethodPointcut;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.dao.ExecutionContextDao;
import org.springframework.batch.core.repository.dao.JobExecutionDao;
import org.springframework.batch.core.repository.dao.JobInstanceDao;
import org.springframework.batch.core.repository.dao.StepExecutionDao;
import org.springframework.batch.support.PropertiesConverter;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.interceptor.TransactionInterceptor;
import org.springframework.util.Assert;
/**
* A {@link FactoryBean} that automates the creation of a
@@ -19,7 +27,20 @@ import org.springframework.beans.factory.FactoryBean;
* @author Lucas Ward
* @author Robert Kasanicky
*/
public abstract class AbstractJobRepositoryFactoryBean implements FactoryBean {
public abstract class AbstractJobRepositoryFactoryBean implements FactoryBean, InitializingBean {
private PlatformTransactionManager transactionManager;
private ProxyFactory proxyFactory;
private String isolationLevelForCreate = DEFAULT_ISOLATION_LEVEL;
/**
* Default value for isolation level in create* method.
*/
private static final String DEFAULT_ISOLATION_LEVEL = "ISOLATION_SERIALIZABLE";
/**
* @return fully configured {@link JobInstanceDao} implementation.
@@ -35,7 +56,7 @@ public abstract class AbstractJobRepositoryFactoryBean implements FactoryBean {
* @return fully configured {@link StepExecutionDao} implementation.
*/
protected abstract StepExecutionDao createStepExecutionDao() throws Exception;
/**
* @return fully configured {@link ExecutionContextDao} implementation.
*/
@@ -55,4 +76,57 @@ public abstract class AbstractJobRepositoryFactoryBean implements FactoryBean {
return true;
}
/**
* public setter for the isolation level to be used for the transaction when
* job execution entities are initially created. The default is
* ISOLATION_SERIALIZABLE, which prevents accidental concurrent execution of
* the same job (ISOLATION_REPEATABLE_READ would work as well).
*
* @param isolationLevelForCreate the isolation level name to set
*
* @see SimpleJobRepository#createJobExecution(String,
* org.springframework.batch.core.JobParameters)
*/
public void setIsolationLevelForCreate(String isolationLevelForCreate) {
this.isolationLevelForCreate = isolationLevelForCreate;
}
/**
* Public setter for the {@link PlatformTransactionManager}.
* @param transactionManager the transactionManager to set
*/
public void setTransactionManager(PlatformTransactionManager transactionManager) {
this.transactionManager = transactionManager;
}
private void initializeProxy() throws Exception {
proxyFactory = new ProxyFactory();
TransactionInterceptor advice = new TransactionInterceptor(transactionManager, PropertiesConverter
.stringToProperties("create*=PROPAGATION_REQUIRES_NEW," + isolationLevelForCreate
+ "\n*=PROPAGATION_REQUIRED"));
DefaultPointcutAdvisor advisor = new DefaultPointcutAdvisor(advice);
NameMatchMethodPointcut pointcut = new NameMatchMethodPointcut();
pointcut.addMethodName("*");
advisor.setPointcut(pointcut);
proxyFactory.addAdvisor(advisor);
proxyFactory.setProxyTargetClass(false);
proxyFactory.addInterface(JobRepository.class);
proxyFactory.setTarget(getTarget());
}
public void afterPropertiesSet() throws Exception {
Assert.notNull(transactionManager, "TransactionManager must not be null.");
initializeProxy();
}
private Object getTarget() throws Exception {
return new SimpleJobRepository(createJobInstanceDao(), createJobExecutionDao(), createStepExecutionDao(),
createExecutionContextDao());
}
public Object getObject() throws Exception {
return proxyFactory.getProxy();
}
}

83
spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/JobRepositoryFactoryBean.java
View File

@@ -20,10 +20,6 @@ import javax.sql.DataSource;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.aop.support.DefaultPointcutAdvisor;
import org.springframework.aop.support.NameMatchMethodPointcut;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.dao.AbstractJdbcBatchMetadataDao;
import org.springframework.batch.core.repository.dao.ExecutionContextDao;
import org.springframework.batch.core.repository.dao.JdbcExecutionContextDao;
@@ -36,13 +32,10 @@ import org.springframework.batch.core.repository.dao.StepExecutionDao;
import org.springframework.batch.item.database.support.DataFieldMaxValueIncrementerFactory;
import org.springframework.batch.item.database.support.DefaultDataFieldMaxValueIncrementerFactory;
import org.springframework.batch.support.DatabaseType;
import org.springframework.batch.support.PropertiesConverter;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.jdbc.core.simple.SimpleJdbcOperations;
import org.springframework.jdbc.core.simple.SimpleJdbcTemplate;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.interceptor.TransactionInterceptor;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
@@ -56,17 +49,8 @@ import org.springframework.util.StringUtils;
* @author Lucas Ward
*/
public class JobRepositoryFactoryBean extends AbstractJobRepositoryFactoryBean implements InitializingBean {
protected static final Log logger = LogFactory.getLog(JobRepositoryFactoryBean.class);
/**
* Default value for isolation level in create* method.
*/
private static final String DEFAULT_ISOLATION_LEVEL = "ISOLATION_SERIALIZABLE";
private ProxyFactory proxyFactory;
private String isolationLevelForCreate = DEFAULT_ISOLATION_LEVEL;
private DataSource dataSource;
@@ -77,11 +61,9 @@ public class JobRepositoryFactoryBean extends AbstractJobRepositoryFactoryBean i
private String tablePrefix = AbstractJdbcBatchMetadataDao.DEFAULT_TABLE_PREFIX;
private DataFieldMaxValueIncrementerFactory incrementerFactory;
private PlatformTransactionManager transactionManager;
private int exitMessageLength = AbstractJdbcBatchMetadataDao.DEFAULT_EXIT_MESSAGE_LENGTH;
/**
* Public setter for the exit message length in database. Do not set this if
* you haven't modified the schema. Note this value will be used for both
@@ -93,29 +75,6 @@ public class JobRepositoryFactoryBean extends AbstractJobRepositoryFactoryBean i
this.exitMessageLength = exitMessageLength;
}
/**
* Public setter for the isolation level to be used for the transaction when
* job execution entities are initially created. The default is
* ISOLATION_SERIALIZABLE, which prevents accidental concurrent execution of
* the same job (ISOLATION_REPEATABLE_READ would work as well).
*
* @param isolationLevelForCreate the isolation level name to set
*
* @see SimpleJobRepository#createJobExecution(String,
* org.springframework.batch.core.JobParameters)
*/
public void setIsolationLevelForCreate(String isolationLevelForCreate) {
this.isolationLevelForCreate = isolationLevelForCreate;
}
/**
* Public setter for the {@link PlatformTransactionManager}.
* @param transactionManager the transactionManager to set
*/
public void setTransactionManager(PlatformTransactionManager transactionManager) {
this.transactionManager = transactionManager;
}
/**
* Public setter for the {@link DataSource}.
* @param dataSource a {@link DataSource}
@@ -126,7 +85,8 @@ public class JobRepositoryFactoryBean extends AbstractJobRepositoryFactoryBean i
/**
* Sets the database type.
* @param dbType as specified by {@link DefaultDataFieldMaxValueIncrementerFactory}
* @param dbType as specified by
* {@link DefaultDataFieldMaxValueIncrementerFactory}
*/
public void setDatabaseType(String dbType) {
this.databaseType = dbType;
@@ -146,16 +106,15 @@ public class JobRepositoryFactoryBean extends AbstractJobRepositoryFactoryBean i
public void afterPropertiesSet() throws Exception {
Assert.notNull(transactionManager, "TransactionManager must not be null.");
Assert.notNull(dataSource, "DataSource must not be null.");
jdbcTemplate = new SimpleJdbcTemplate(dataSource);
if (incrementerFactory == null) {
incrementerFactory = new DefaultDataFieldMaxValueIncrementerFactory(dataSource);
}
if(databaseType == null){
if (databaseType == null) {
databaseType = DatabaseType.fromMetaData(dataSource).name();
logger.info("No database type set, using meta data indicating: " + databaseType);
}
@@ -163,28 +122,9 @@ public class JobRepositoryFactoryBean extends AbstractJobRepositoryFactoryBean i
Assert.isTrue(incrementerFactory.isSupportedIncrementerType(databaseType), "'" + databaseType
+ "' is an unsupported database type. The supported database types are "
+ StringUtils.arrayToCommaDelimitedString(incrementerFactory.getSupportedIncrementerTypes()));
initializeProxy();
}
protected void initializeProxy() throws Exception {
proxyFactory = new ProxyFactory();
TransactionInterceptor advice = new TransactionInterceptor(transactionManager, PropertiesConverter
.stringToProperties("create*=PROPAGATION_REQUIRES_NEW," + isolationLevelForCreate
+ "\n*=PROPAGATION_REQUIRED"));
DefaultPointcutAdvisor advisor = new DefaultPointcutAdvisor(advice);
NameMatchMethodPointcut pointcut = new NameMatchMethodPointcut();
pointcut.addMethodName("*");
advisor.setPointcut(pointcut);
proxyFactory.addAdvisor(advisor);
proxyFactory.setProxyTargetClass(false);
proxyFactory.addInterface(JobRepository.class);
proxyFactory.setTarget(getTarget());
}
private Object getTarget() throws Exception {
return new SimpleJobRepository(createJobInstanceDao(), createJobExecutionDao(), createStepExecutionDao(), createExecutionContextDao());
super.afterPropertiesSet();
}
@Override
@@ -230,7 +170,4 @@ public class JobRepositoryFactoryBean extends AbstractJobRepositoryFactoryBean i
return dao;
}
public Object getObject() throws Exception {
return proxyFactory.getProxy();
}
}

5
spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/MapJobRepositoryFactoryBean.java
View File

@@ -39,9 +39,4 @@ public class MapJobRepositoryFactoryBean extends AbstractJobRepositoryFactoryBea
return new MapExecutionContextDao();
}
public Object getObject() throws Exception {
return new SimpleJobRepository(createJobInstanceDao(), createJobExecutionDao(), createStepExecutionDao(),
createExecutionContextDao());
}
}

4
spring-batch-core/src/test/java/org/springframework/batch/core/repository/support/JobRepositoryFactoryBeanTests.java
View File

@@ -104,11 +104,13 @@ public class JobRepositoryFactoryBeanTests {
@Test
public void testMissingTransactionManager() throws Exception {
factory.setDatabaseType("mockDb");
factory.setTransactionManager(null);
try {
expect(incrementerFactory.isSupportedIncrementerType(null)).andReturn(true);
expect(incrementerFactory.isSupportedIncrementerType("mockDb")).andReturn(true);
expect(incrementerFactory.getSupportedIncrementerTypes()).andReturn(new String[0]);
replay(incrementerFactory);
factory.afterPropertiesSet();
fail();
}

3
spring-batch-core/src/test/java/org/springframework/batch/core/repository/support/MapJobRepositoryFactoryBeanTests.java
View File

@@ -8,6 +8,7 @@ import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.job.JobSupport;
import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
/**
* Tests for {@link MapJobRepositoryFactoryBean}.
@@ -22,6 +23,8 @@ public class MapJobRepositoryFactoryBeanTests {
*/
@Test
public void testCreateRepository() throws Exception {
tested.setTransactionManager(new ResourcelessTransactionManager());
tested.afterPropertiesSet();
JobRepository repository = (JobRepository) tested.getObject();
Job job = new JobSupport("jobName");
JobParameters jobParameters = new JobParameters();