Add transaction support in JobExplorerFactoryBean

This commit adds support to create a transactional
proxy around the JobExplorer created by the
JobExplorerFactoryBean.

Resolves #1307

Author: fmbenhassine

spring-batch-core/src/main/java/org/springframework/batch/core/configuration/annotation/BatchRegistrar.java

@@ -136,6 +136,9 @@ private void registerJobExplorer(BeanDefinitionRegistry registry, EnableBatchPro
 		String dataSourceRef = batchAnnotation.dataSourceRef();
 		beanDefinitionBuilder.addPropertyReference("dataSource", dataSourceRef);
 
+		String transactionManagerRef = batchAnnotation.transactionManagerRef();
+		beanDefinitionBuilder.addPropertyReference("transactionManager", transactionManagerRef);
+
 		// set optional properties
 		String executionContextSerializerRef = batchAnnotation.executionContextSerializerRef();
 		if (registry.containsBeanDefinition(executionContextSerializerRef)) {

spring-batch-core/src/main/java/org/springframework/batch/core/configuration/support/DefaultBatchConfiguration.java

@@ -157,6 +157,7 @@ public JobLauncher jobLauncher() throws BatchConfigurationException {
 	public JobExplorer jobExplorer() throws BatchConfigurationException {
 		JobExplorerFactoryBean jobExplorerFactoryBean = new JobExplorerFactoryBean();
 		jobExplorerFactoryBean.setDataSource(getDataSource());
+		jobExplorerFactoryBean.setTransactionManager(getTransactionManager());
 		jobExplorerFactoryBean.setJdbcOperations(getJdbcOperations());
 		jobExplorerFactoryBean.setCharset(getCharset());
 		jobExplorerFactoryBean.setTablePrefix(getTablePrefix());

spring-batch-core/src/main/java/org/springframework/batch/core/explore/support/AbstractJobExplorerFactoryBean.java

@@ -16,12 +16,23 @@
 
 package org.springframework.batch.core.explore.support;
 
+import java.util.Properties;
+
+import org.springframework.aop.framework.ProxyFactory;
 import org.springframework.batch.core.explore.JobExplorer;
 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.beans.factory.FactoryBean;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.TransactionManager;
+import org.springframework.transaction.annotation.Isolation;
+import org.springframework.transaction.annotation.Propagation;
+import org.springframework.transaction.interceptor.NameMatchTransactionAttributeSource;
+import org.springframework.transaction.interceptor.TransactionInterceptor;
+import org.springframework.util.Assert;
 
 /**
  * A {@link FactoryBean} that automates the creation of a {@link SimpleJobExplorer}. It
@@ -32,7 +43,15 @@
  * @author Mahmoud Ben Hassine
  * @since 2.0
  */
-public abstract class AbstractJobExplorerFactoryBean implements FactoryBean<JobExplorer> {
+public abstract class AbstractJobExplorerFactoryBean implements FactoryBean<JobExplorer>, InitializingBean {
+
+	private static final String TRANSACTION_ISOLATION_LEVEL_PREFIX = "ISOLATION_";
+
+	private static final String TRANSACTION_PROPAGATION_PREFIX = "PROPAGATION_";
+
+	private PlatformTransactionManager transactionManager;
+
+	private ProxyFactory proxyFactory = new ProxyFactory();
 
 	/**
 	 * Creates a job instance data access object (DAO).
@@ -62,6 +81,30 @@ public abstract class AbstractJobExplorerFactoryBean implements FactoryBean<JobE
 	 */
 	protected abstract ExecutionContextDao createExecutionContextDao() throws Exception;
 
+	/**
+	 * Public setter for the {@link PlatformTransactionManager}.
+	 * @param transactionManager the transactionManager to set
+	 * @since 5.0
+	 */
+	public void setTransactionManager(PlatformTransactionManager transactionManager) {
+		this.transactionManager = transactionManager;
+	}
+
+	/**
+	 * The transaction manager used in this factory. Useful to inject into steps and jobs,
+	 * to ensure that they are using the same instance.
+	 * @return the transactionManager
+	 * @since 5.0
+	 */
+	public PlatformTransactionManager getTransactionManager() {
+		return this.transactionManager;
+	}
+
+	@Override
+	public void afterPropertiesSet() throws Exception {
+		Assert.notNull(this.transactionManager, "TransactionManager must not be null.");
+	}
+
 	/**
 	 * Returns the type of object to be returned from {@link #getObject()}.
 	 * @return {@code JobExplorer.class}
@@ -77,4 +120,27 @@ public boolean isSingleton() {
 		return true;
 	}
 
+	@Override
+	public JobExplorer getObject() throws Exception {
+		Properties transactionAttributes = new Properties();
+		String transactionProperties = String.join(",", TRANSACTION_PROPAGATION_PREFIX + Propagation.SUPPORTS,
+				TRANSACTION_ISOLATION_LEVEL_PREFIX + Isolation.READ_COMMITTED);
+		transactionAttributes.setProperty("get*", transactionProperties);
+		transactionAttributes.setProperty("find*", transactionProperties);
+		NameMatchTransactionAttributeSource transactionAttributeSource = new NameMatchTransactionAttributeSource();
+		transactionAttributeSource.setProperties(transactionAttributes);
+		TransactionInterceptor advice = new TransactionInterceptor((TransactionManager) transactionManager,
+				transactionAttributeSource);
+		proxyFactory.addAdvice(advice);
+		proxyFactory.setProxyTargetClass(false);
+		proxyFactory.addInterface(JobExplorer.class);
+		proxyFactory.setTarget(getTarget());
+		return (JobExplorer) proxyFactory.getProxy(getClass().getClassLoader());
+	}
+
+	private JobExplorer getTarget() throws Exception {
+		return new SimpleJobExplorer(createJobInstanceDao(), createJobExecutionDao(), createStepExecutionDao(),
+				createExecutionContextDao());
+	}
+
 }

spring-batch-core/src/main/java/org/springframework/batch/core/explore/support/JobExplorerFactoryBean.java

@@ -147,11 +147,8 @@ public void afterPropertiesSet() throws Exception {
 		if (serializer == null) {
 			serializer = new Jackson2ExecutionContextStringSerializer();
 		}
-	}
 
-	private JobExplorer getTarget() throws Exception {
-		return new SimpleJobExplorer(createJobInstanceDao(), createJobExecutionDao(), createStepExecutionDao(),
-				createExecutionContextDao());
+		super.afterPropertiesSet();
 	}
 
 	@Override
@@ -196,9 +193,4 @@ protected StepExecutionDao createStepExecutionDao() throws Exception {
 		return dao;
 	}
 
-	@Override
-	public JobExplorer getObject() throws Exception {
-		return getTarget();
-	}
-
 }

spring-batch-core/src/test/java/org/springframework/batch/core/explore/support/JobExplorerFactoryBeanTests.java

@@ -29,10 +29,12 @@
 import org.springframework.jdbc.core.JdbcOperations;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.test.util.ReflectionTestUtils;
+import org.springframework.transaction.PlatformTransactionManager;
 
 /**
  * @author Dave Syer
  * @author Will Schipp
+ * @author Mahmoud Ben Hassine
  *
  */
 class JobExplorerFactoryBeanTests {
@@ -46,7 +48,9 @@ void setUp() {
 
 		factory = new JobExplorerFactoryBean();
 		DataSource dataSource = mock(DataSource.class);
+		PlatformTransactionManager transactionManager = mock(PlatformTransactionManager.class);
 		factory.setDataSource(dataSource);
+		factory.setTransactionManager(transactionManager);
 		factory.setTablePrefix(tablePrefix);
 
 	}
@@ -78,6 +82,16 @@ void testMissingDataSource() {
 
 	}
 
+	@Test
+	void testMissingTransactionManager() {
+
+		factory.setTransactionManager(null);
+		Exception exception = assertThrows(IllegalArgumentException.class, factory::afterPropertiesSet);
+		String message = exception.getMessage();
+		assertTrue(message.contains("TransactionManager"), "Wrong message: " + message);
+
+	}
+
 	@Test
 	void testCreateExplorer() throws Exception {
 

spring-batch-core/src/test/java/org/springframework/batch/core/explore/support/SimpleJobExplorerIntegrationTests.java

@@ -86,6 +86,7 @@ public JobExplorer jobExplorer() throws Exception {
 		public JobExplorerFactoryBean jobExplorerFactoryBean() {
 			JobExplorerFactoryBean jobExplorerFactoryBean = new JobExplorerFactoryBean();
 			jobExplorerFactoryBean.setDataSource(dataSource());
+			jobExplorerFactoryBean.setTransactionManager(transactionManager(dataSource()));
 			return jobExplorerFactoryBean;
 		}
 

spring-batch-core/src/test/java/org/springframework/batch/core/job/SimpleJobTests.java

@@ -105,6 +105,7 @@ void setUp() throws Exception {
 		this.jobRepository = repositoryFactoryBean.getObject();
 		JobExplorerFactoryBean explorerFactoryBean = new JobExplorerFactoryBean();
 		explorerFactoryBean.setDataSource(embeddedDatabase);
+		explorerFactoryBean.setTransactionManager(transactionManager);
 		explorerFactoryBean.afterPropertiesSet();
 		this.jobExplorer = explorerFactoryBean.getObject();
 		job = new SimpleJob();

spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/FlowJobTests.java

@@ -76,16 +76,18 @@ void setUp() throws Exception {
 		EmbeddedDatabase embeddedDatabase = new EmbeddedDatabaseBuilder()
 				.addScript("/org/springframework/batch/core/schema-drop-hsqldb.sql")
 				.addScript("/org/springframework/batch/core/schema-hsqldb.sql").build();
+		JdbcTransactionManager transactionManager = new JdbcTransactionManager(embeddedDatabase);
 		JobRepositoryFactoryBean factory = new JobRepositoryFactoryBean();
 		factory.setDataSource(embeddedDatabase);
-		factory.setTransactionManager(new JdbcTransactionManager(embeddedDatabase));
+		factory.setTransactionManager(transactionManager);
 		factory.afterPropertiesSet();
 		this.jobRepository = factory.getObject();
 		job.setJobRepository(this.jobRepository);
 		this.jobExecution = this.jobRepository.createJobExecution("job", new JobParameters());
 
 		JobExplorerFactoryBean explorerFactoryBean = new JobExplorerFactoryBean();
 		explorerFactoryBean.setDataSource(embeddedDatabase);
+		explorerFactoryBean.setTransactionManager(transactionManager);
 		explorerFactoryBean.afterPropertiesSet();
 		this.jobExplorer = explorerFactoryBean.getObject();
 	}

spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/RemoteStepExecutionAggregatorTests.java

@@ -53,13 +53,15 @@ void init() throws Exception {
 		EmbeddedDatabase embeddedDatabase = new EmbeddedDatabaseBuilder()
 				.addScript("/org/springframework/batch/core/schema-drop-hsqldb.sql")
 				.addScript("/org/springframework/batch/core/schema-hsqldb.sql").build();
+		JdbcTransactionManager transactionManager = new JdbcTransactionManager(embeddedDatabase);
 		JobRepositoryFactoryBean factory = new JobRepositoryFactoryBean();
 		factory.setDataSource(embeddedDatabase);
-		factory.setTransactionManager(new JdbcTransactionManager(embeddedDatabase));
+		factory.setTransactionManager(transactionManager);
 		factory.afterPropertiesSet();
 		JobRepository jobRepository = factory.getObject();
 		JobExplorerFactoryBean explorerFactoryBean = new JobExplorerFactoryBean();
 		explorerFactoryBean.setDataSource(embeddedDatabase);
+		explorerFactoryBean.setTransactionManager(transactionManager);
 		explorerFactoryBean.afterPropertiesSet();
 		aggregator.setJobExplorer(explorerFactoryBean.getObject());
 		jobExecution = jobRepository.createJobExecution("job", new JobParameters());

spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/support/JobRegistryIntegrationTests-context.xml

@@ -30,6 +30,7 @@
 
 	<bean id="jobExplorer" class="org.springframework.batch.core.explore.support.JobExplorerFactoryBean">
 		<property name="dataSource" ref="dataSource"/>
+		<property name="transactionManager" ref="transactionManager"/>
 	</bean>
 
 	<bean id="tasklet" class="org.springframework.batch.core.configuration.xml.FailingTasklet"/>