diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/JobRepositoryFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/JobRepositoryFactoryBean.java index 3ec585b88..d01d6d3e8 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/JobRepositoryFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/JobRepositoryFactoryBean.java @@ -18,6 +18,10 @@ package org.springframework.batch.core.repository.support; import javax.sql.DataSource; +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.JdbcJobExecutionDao; import org.springframework.batch.core.repository.dao.JdbcJobInstanceDao; @@ -27,21 +31,30 @@ import org.springframework.batch.core.repository.dao.JobInstanceDao; 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.PropertiesConverter; import org.springframework.beans.factory.FactoryBean; import org.springframework.beans.factory.InitializingBean; import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.interceptor.TransactionInterceptor; import org.springframework.util.Assert; import org.springframework.util.StringUtils; /** - * A {@link FactoryBean} that automates the creation of a {@link SimpleJobRepository}. Requires the - * user to describe what kind of database they are using. + * A {@link FactoryBean} that automates the creation of a + * {@link SimpleJobRepository}. Requires the user to describe what kind of + * database they are using. * * @author Ben Hale * @author Lucas Ward */ public class JobRepositoryFactoryBean implements FactoryBean, InitializingBean { + /** + * Default value for isolation level in create* method. + */ + private static final String DEFAULT_ISOLATION_LEVEL = "ISOLATION_SERIALIZABLE"; + private DataSource dataSource; private String databaseType; @@ -50,6 +63,35 @@ public class JobRepositoryFactoryBean implements FactoryBean, InitializingBean { private DataFieldMaxValueIncrementerFactory incrementerFactory; + private PlatformTransactionManager transactionManager; + + private ProxyFactory proxyFactory; + + private String isolationLevelForCreate = DEFAULT_ISOLATION_LEVEL; + + /** + * Public setter for the {@link PlatformTransactionManager}. + * @param transactionManager the transactionManager to set + */ + public void setTransactionManager(PlatformTransactionManager transactionManager) { + this.transactionManager = transactionManager; + } + + /** + * 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(org.springframework.batch.core.Job, + * org.springframework.batch.core.JobParameters) + */ + public void setIsolationLevelForCreate(String isolationLevelForCreate) { + this.isolationLevelForCreate = isolationLevelForCreate; + } + public void setDataSource(DataSource dataSource) { this.dataSource = dataSource; } @@ -67,18 +109,39 @@ public class JobRepositoryFactoryBean implements FactoryBean, InitializingBean { } public void afterPropertiesSet() throws Exception { - Assert.notNull(dataSource, "Datasource must not be null."); + Assert.notNull(dataSource, "DataSource must not be null."); + Assert.notNull(transactionManager, "TransactionManager must not be null."); if (incrementerFactory == null) { incrementerFactory = new DefaultDataFieldMaxValueIncrementerFactory(dataSource); } Assert.isTrue(incrementerFactory.isSupportedIncrementerType(databaseType), "'" + databaseType - + "' is an unsupported database type. The supported database types are " - + StringUtils.arrayToCommaDelimitedString(incrementerFactory.getSupportedIncrementerTypes())); + + "' is an unsupported database type. The supported database types are " + + StringUtils.arrayToCommaDelimitedString(incrementerFactory.getSupportedIncrementerTypes())); + + initializeProxy(); } - public Object getObject() throws Exception { + 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()); + } + + /** + * @return a SimpleJobRepository + */ + private SimpleJobRepository getTarget() throws Exception { JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource); JobInstanceDao jobInstanceDao = createJobInstanceDao(jdbcTemplate); JobExecutionDao jobExecutionDao = createJobExecutionDao(jdbcTemplate); @@ -86,8 +149,25 @@ public class JobRepositoryFactoryBean implements FactoryBean, InitializingBean { return new SimpleJobRepository(jobInstanceDao, jobExecutionDao, stepExecutionDao); } + /** + * Get a concrete {@link JobRepository}. The repository will be a proxy + * containing transaction advice using the supplied transaction manager. + * + * @return a {@link JobRepository} + * @see org.springframework.beans.factory.FactoryBean#getObject() + */ + public Object getObject() throws Exception { + return proxyFactory.getProxy(); + } + + /** + * The type of object to be returned from {@link #getObject()}. + * + * @return JobRepository.class + * @see org.springframework.beans.factory.FactoryBean#getObjectType() + */ public Class getObjectType() { - return SimpleJobRepository.class; + return JobRepository.class; } public boolean isSingleton() { @@ -106,7 +186,8 @@ public class JobRepositoryFactoryBean implements FactoryBean, InitializingBean { private JobExecutionDao createJobExecutionDao(JdbcTemplate jdbcTemplate) throws Exception { JdbcJobExecutionDao dao = new JdbcJobExecutionDao(); dao.setJdbcTemplate(jdbcTemplate); - dao.setJobExecutionIncrementer(incrementerFactory.getIncrementer(databaseType, tablePrefix + "JOB_EXECUTION_SEQ")); + dao.setJobExecutionIncrementer(incrementerFactory.getIncrementer(databaseType, tablePrefix + + "JOB_EXECUTION_SEQ")); dao.setTablePrefix(tablePrefix); dao.afterPropertiesSet(); return dao; @@ -115,7 +196,8 @@ public class JobRepositoryFactoryBean implements FactoryBean, InitializingBean { private StepExecutionDao createStepExecutionDao(JdbcTemplate jdbcTemplate) throws Exception { JdbcStepExecutionDao dao = new JdbcStepExecutionDao(); dao.setJdbcTemplate(jdbcTemplate); - dao.setStepExecutionIncrementer(incrementerFactory.getIncrementer(databaseType, tablePrefix + "STEP_EXECUTION_SEQ")); + dao.setStepExecutionIncrementer(incrementerFactory.getIncrementer(databaseType, tablePrefix + + "STEP_EXECUTION_SEQ")); dao.setTablePrefix(tablePrefix); dao.afterPropertiesSet(); return dao; diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/AbstractStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/AbstractStepFactoryBean.java index 5bb9ec9c9..52add37c0 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/AbstractStepFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/AbstractStepFactoryBean.java @@ -19,6 +19,7 @@ import org.springframework.batch.core.Step; import org.springframework.batch.core.repository.JobRepository; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.item.validator.Validator; import org.springframework.beans.factory.BeanNameAware; import org.springframework.beans.factory.FactoryBean; import org.springframework.transaction.PlatformTransactionManager; @@ -33,7 +34,7 @@ import org.springframework.util.Assert; * */ public abstract class AbstractStepFactoryBean implements FactoryBean, BeanNameAware { - + private String name; private int startLimit = Integer.MAX_VALUE; @@ -50,6 +51,8 @@ public abstract class AbstractStepFactoryBean implements FactoryBean, BeanNameAw private boolean singleton = true; + private Validator jobRepositoryValidator = new TransactionInterceptorValidator(1); + /** * */ @@ -160,8 +163,8 @@ public abstract class AbstractStepFactoryBean implements FactoryBean, BeanNameAw Assert.notNull(getItemReader(), "ItemReader must be provided"); Assert.notNull(getItemWriter(), "ItemWriter must be provided"); - Assert.notNull(jobRepository, "JobRepository must be provided"); Assert.notNull(transactionManager, "TransactionManager must be provided"); + jobRepositoryValidator .validate(jobRepository); step.setItemHandler(new SimpleItemHandler(itemReader, itemWriter)); step.setTransactionManager(transactionManager); diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/TransactionInterceptorValidator.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/TransactionInterceptorValidator.java new file mode 100644 index 000000000..b6e75abaf --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/TransactionInterceptorValidator.java @@ -0,0 +1,87 @@ +/* + * 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.core.step.item; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.aop.Advisor; +import org.springframework.aop.framework.Advised; +import org.springframework.batch.item.validator.ValidationException; +import org.springframework.batch.item.validator.Validator; +import org.springframework.transaction.interceptor.TransactionInterceptor; +import org.springframework.util.Assert; + +/** + * Simple validator for internal use only (package private to make it testable). + * Asserts that its argument has no more than the specified number of + * transaction interceptors in its advice chain. + * + * @author Dave Syer + * + */ +class TransactionInterceptorValidator implements Validator { + + protected Log logger = LogFactory.getLog(getClass()); + + private final int maxCount; + + /** + * @param maxCount + */ + public TransactionInterceptorValidator(int maxCount) { + super(); + this.maxCount = maxCount; + } + + /** + * Assert that the object passed in has no more than the maximum number of + * transaction interceptors in its advice chain. + * + * @see org.springframework.batch.item.validator.Validator#validate(java.lang.Object) + */ + public void validate(Object value) throws ValidationException { + Assert.notNull(value, "JobRepository must be provided"); + Assert.state(countTransactionInterceptors(value) <= maxCount, + "JobRepository has more than one transaction interceptor. " + + "Do not declare a separate transaction advice if using the JobRepositoryFactoryBean."); + } + + /** + * @param object an Object, possibly advised + * @return the number of transaction interceptors in the advice chain + */ + private int countTransactionInterceptors(Object object) { + int count = 0; + Object target = object; + while (target instanceof Advised) { + Advised advised = (Advised) target; + Advisor[] interceptors = advised.getAdvisors(); + for (int i = 0; i < interceptors.length; i++) { + if (interceptors[i].getAdvice() instanceof TransactionInterceptor) { + count++; + } + } + try { + target = advised.getTargetSource().getTarget(); + } + catch (Exception e) { + logger.warn("Target could not be obtained from advised instance.", e); + } + } + return count; + } + +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/repository/support/JobRepositoryFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/repository/support/JobRepositoryFactoryBeanTests.java index edf3a5b19..81a79c778 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/repository/support/JobRepositoryFactoryBeanTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/repository/support/JobRepositoryFactoryBeanTests.java @@ -15,15 +15,22 @@ */ package org.springframework.batch.core.repository.support; +import java.sql.Connection; + import javax.sql.DataSource; +import junit.framework.AssertionFailedError; import junit.framework.TestCase; import org.easymock.MockControl; -import org.springframework.batch.core.repository.support.JobRepositoryFactoryBean; +import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.job.JobSupport; +import org.springframework.batch.core.repository.JobRepository; import org.springframework.batch.item.database.support.DataFieldMaxValueIncrementerFactory; import org.springframework.dao.DataAccessException; import org.springframework.jdbc.support.incrementer.DataFieldMaxValueIncrementer; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.support.DefaultTransactionDefinition; /** * @author Lucas Ward @@ -31,23 +38,32 @@ import org.springframework.jdbc.support.incrementer.DataFieldMaxValueIncrementer */ public class JobRepositoryFactoryBeanTests extends TestCase { - JobRepositoryFactoryBean factory; + private JobRepositoryFactoryBean factory; - MockControl incrementerControl = MockControl.createControl(DataFieldMaxValueIncrementerFactory.class); + private MockControl incrementerControl = MockControl.createControl(DataFieldMaxValueIncrementerFactory.class); - DataFieldMaxValueIncrementerFactory incrementerFactory; + private DataFieldMaxValueIncrementerFactory incrementerFactory; - DataSource dataSource; - - String tablePrefix = "TEST_BATCH_PREFIX_"; + private DataSource dataSource; + + private PlatformTransactionManager transactionManager; + + private String tablePrefix = "TEST_BATCH_PREFIX_"; + + private MockControl txControl; + + private MockControl dataSourceControl; protected void setUp() throws Exception { super.setUp(); factory = new JobRepositoryFactoryBean(); - MockControl dataSourceControl = MockControl.createControl(DataSource.class); + dataSourceControl = MockControl.createControl(DataSource.class); dataSource = (DataSource) dataSourceControl.getMock(); + txControl = MockControl.createControl(PlatformTransactionManager.class); + transactionManager = (PlatformTransactionManager) txControl.getMock(); factory.setDataSource(dataSource); + factory.setTransactionManager(transactionManager); incrementerFactory = (DataFieldMaxValueIncrementerFactory) incrementerControl.getMock(); factory.setIncrementerFactory(incrementerFactory); factory.setTablePrefix(tablePrefix); @@ -66,14 +82,44 @@ public class JobRepositoryFactoryBeanTests extends TestCase { } catch (IllegalArgumentException ex) { // expected + String message = ex.getMessage(); + assertTrue("Wrong message: " + message, message.indexOf("unsupported database type") >= 0); + } + } + + public void testMissingDataSource() throws Exception { + + factory.setDataSource(null); + try { + factory.afterPropertiesSet(); + fail(); + } + catch (IllegalArgumentException ex) { + // expected + String message = ex.getMessage(); + assertTrue("Wrong message: " + message, message.indexOf("DataSource") >= 0); + } + } + + public void testMissingTransactionManager() throws Exception { + + factory.setTransactionManager(null); + try { + factory.afterPropertiesSet(); + fail(); + } + catch (IllegalArgumentException ex) { + // expected + String message = ex.getMessage(); + assertTrue("Wrong message: " + message, message.indexOf("TransactionManager") >= 0); } } public void testInvalidDatabaseType() throws Exception { - factory.setDatabaseType("invalid type"); + factory.setDatabaseType("foo"); try { - incrementerFactory.isSupportedIncrementerType("invalid type"); + incrementerFactory.isSupportedIncrementerType("foo"); incrementerControl.setReturnValue(false); incrementerFactory.getSupportedIncrementerTypes(); incrementerControl.setReturnValue(new String[0]); @@ -83,13 +129,19 @@ public class JobRepositoryFactoryBeanTests extends TestCase { } catch (IllegalArgumentException ex) { // expected + String message = ex.getMessage(); + assertTrue("Wrong message: " + message, message.indexOf("foo") >= 0); } } public void testCreateRepository() throws Exception { - String databaseType = "databaseType"; + String databaseType = "foo"; factory.setDatabaseType(databaseType); + incrementerFactory.isSupportedIncrementerType("foo"); + incrementerControl.setReturnValue(true); + incrementerFactory.getSupportedIncrementerTypes(); + incrementerControl.setReturnValue(new String[0]); incrementerFactory.getIncrementer(databaseType, tablePrefix + "JOB_SEQ"); incrementerControl.setReturnValue(new StubIncrementer()); incrementerFactory.getIncrementer(databaseType, tablePrefix + "JOB_EXECUTION_SEQ"); @@ -98,11 +150,79 @@ public class JobRepositoryFactoryBeanTests extends TestCase { incrementerControl.setReturnValue(new StubIncrementer()); incrementerControl.replay(); + factory.afterPropertiesSet(); factory.getObject(); incrementerControl.verify(); } + public void testTransactionAttributesForCreateMethodNullHypothesis() throws Exception { + testCreateRepository(); + JobRepository repository = (JobRepository) factory.getObject(); + DefaultTransactionDefinition transactionDefinition = new DefaultTransactionDefinition( + DefaultTransactionDefinition.PROPAGATION_REQUIRES_NEW); + txControl.expectAndReturn(transactionManager.getTransaction(transactionDefinition), null); + txControl.replay(); + try { + repository.createJobExecution(new JobSupport("job"), new JobParameters()); + // we expect an exception from the txControl because we provided the + // wrong meta data + fail("Expected IllegalArgumentException"); + } + catch (AssertionFailedError e) { + // expected exception from txControl - wrong isolation level used in + // comparison + assertEquals("Unexpected method call", e.getMessage().substring(3, 25)); + } + } + + public void testTransactionAttributesForCreateMethod() throws Exception { + testCreateRepository(); + JobRepository repository = (JobRepository) factory.getObject(); + DefaultTransactionDefinition transactionDefinition = new DefaultTransactionDefinition( + DefaultTransactionDefinition.PROPAGATION_REQUIRES_NEW); + transactionDefinition.setIsolationLevel(DefaultTransactionDefinition.ISOLATION_SERIALIZABLE); + txControl.expectAndReturn(transactionManager.getTransaction(transactionDefinition), null); + dataSourceControl.expectAndReturn(dataSource.getConnection(), MockControl.createControl(Connection.class) + .getMock()); + dataSourceControl.replay(); + txControl.replay(); + try { + repository.createJobExecution(new JobSupport("job"), new JobParameters()); + // we expect an exception but not from the txControl because we + // provided the correct meta data + fail("Expected IllegalArgumentException"); + } + catch (IllegalArgumentException e) { + // expected exception from DataSourceUtils + assertEquals("No Statement specified", e.getMessage()); + } + } + + public void testSetTransactionAttributesForCreateMethod() throws Exception { + factory.setIsolationLevelForCreate("ISOLATION_READ_UNCOMMITTED"); + testCreateRepository(); + JobRepository repository = (JobRepository) factory.getObject(); + DefaultTransactionDefinition transactionDefinition = new DefaultTransactionDefinition( + DefaultTransactionDefinition.PROPAGATION_REQUIRES_NEW); + transactionDefinition.setIsolationLevel(DefaultTransactionDefinition.ISOLATION_READ_UNCOMMITTED); + txControl.expectAndReturn(transactionManager.getTransaction(transactionDefinition), null); + dataSourceControl.expectAndReturn(dataSource.getConnection(), MockControl.createControl(Connection.class) + .getMock()); + dataSourceControl.replay(); + txControl.replay(); + try { + repository.createJobExecution(new JobSupport("job"), new JobParameters()); + // we expect an exception but not from the txControl because we + // provided the correct meta data + fail("Expected IllegalArgumentException"); + } + catch (IllegalArgumentException e) { + // expected exception from DataSourceUtils + assertEquals("No Statement specified", e.getMessage()); + } + } + private static class StubIncrementer implements DataFieldMaxValueIncrementer { public int nextIntValue() throws DataAccessException { diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/TransactionInterceptorValidatorTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/TransactionInterceptorValidatorTests.java new file mode 100644 index 000000000..7ad14afa4 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/TransactionInterceptorValidatorTests.java @@ -0,0 +1,65 @@ +/* + * 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.core.step.item; + +import org.springframework.aop.framework.ProxyFactory; +import org.springframework.batch.core.repository.JobRepository; +import org.springframework.transaction.interceptor.TransactionInterceptor; + +import junit.framework.TestCase; + +/** + * @author Dave Syer + * + */ +public class TransactionInterceptorValidatorTests extends TestCase { + + private TransactionInterceptorValidator validator = new TransactionInterceptorValidator(1); + + public void testValidateNull() { + try { + validator.validate(null); + fail("Expected IllegalArgumentException"); + } catch (IllegalArgumentException e) { + String message = e.getMessage(); + assertTrue("Wrong message: "+message, message.indexOf("JobRepository")>=0); + } + } + + public void testValidateWithNoInterceptors() { + validator.validate(new Object()); + } + + public void testValidateAdvisedWithOneInterceptor() { + validator.validate(ProxyFactory.getProxy(JobRepository.class, new TransactionInterceptor())); + } + + public void testValidateAdvisedWithTwoInterceptors() { + Object target = ProxyFactory.getProxy(JobRepository.class, new TransactionInterceptor()); + ProxyFactory factory = new ProxyFactory(); + factory.setTarget(target); + factory.addInterface(JobRepository.class); + factory.addAdvice(new TransactionInterceptor()); + try { + validator.validate(factory.getProxy()); + fail("Expected IllegalStateException"); + } catch (IllegalStateException e) { + String message = e.getMessage(); + assertTrue("Wrong message: "+message, message.indexOf("JobRepository")>=0); + } + } + +} diff --git a/spring-batch-samples/src/main/resources/simple-job-launcher-context.xml b/spring-batch-samples/src/main/resources/simple-job-launcher-context.xml index f6f83b670..4a332dae6 100644 --- a/spring-batch-samples/src/main/resources/simple-job-launcher-context.xml +++ b/spring-batch-samples/src/main/resources/simple-job-launcher-context.xml @@ -20,20 +20,8 @@ - - - - - - - - - - - + p:databaseType="${environment}" p:dataSource-ref="dataSource" p:transactionManager-ref="transactionManager"/>