Add support for transaction customization in SimpleJobOperator

This commit adds support to configure and create a transactional
proxy around SimpleJobOperator through a factory bean.

The usage of `@Transactional` was removed in favor of
the programmatic way of customizing the proxy through
the factory bean.

Resolves #1078

spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/JobOperatorFactoryBean.java

@@ -0,0 +1,161 @@
+/*
+ * Copyright 2022 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
+ *
+ *      https://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.launch.support;
+
+import java.util.Properties;
+
+import org.springframework.aop.framework.ProxyFactory;
+import org.springframework.batch.core.configuration.JobRegistry;
+import org.springframework.batch.core.converter.JobParametersConverter;
+import org.springframework.batch.core.explore.JobExplorer;
+import org.springframework.batch.core.launch.JobLauncher;
+import org.springframework.batch.core.launch.JobOperator;
+import org.springframework.batch.core.repository.JobRepository;
+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;
+
+/**
+ * Convenient factory bean that creates a transactional proxy around a
+ * {@link JobOperator}.
+ *
+ * @see JobOperator
+ * @see SimpleJobOperator
+ * @author Mahmoud Ben Hassine
+ * @since 5.0
+ */
+public class JobOperatorFactoryBean implements FactoryBean<JobOperator>, InitializingBean {
+
+	private static final String TRANSACTION_ISOLATION_LEVEL_PREFIX = "ISOLATION_";
+
+	private static final String TRANSACTION_PROPAGATION_PREFIX = "PROPAGATION_";
+
+	private PlatformTransactionManager transactionManager;
+
+	private JobRegistry jobRegistry;
+
+	private JobLauncher jobLauncher;
+
+	private JobRepository jobRepository;
+
+	private JobExplorer jobExplorer;
+
+	private JobParametersConverter jobParametersConverter;
+
+	private ProxyFactory proxyFactory = new ProxyFactory();
+
+	@Override
+	public void afterPropertiesSet() throws Exception {
+		Assert.notNull(this.transactionManager, "TransactionManager must not be null");
+		Assert.notNull(this.jobLauncher, "JobLauncher must not be null");
+		Assert.notNull(this.jobRegistry, "JobLocator must not be null");
+		Assert.notNull(this.jobExplorer, "JobExplorer must not be null");
+		Assert.notNull(this.jobRepository, "JobRepository must not be null");
+	}
+
+	/**
+	 * Setter for the job registry.
+	 * @param jobRegistry the job registry to set
+	 */
+	public void setJobRegistry(JobRegistry jobRegistry) {
+		this.jobRegistry = jobRegistry;
+	}
+
+	/**
+	 * Setter for the job launcher.
+	 * @param jobLauncher the job launcher to set
+	 */
+	public void setJobLauncher(JobLauncher jobLauncher) {
+		this.jobLauncher = jobLauncher;
+	}
+
+	/**
+	 * Setter for the job repository.
+	 * @param jobRepository the job repository to set
+	 */
+	public void setJobRepository(JobRepository jobRepository) {
+		this.jobRepository = jobRepository;
+	}
+
+	/**
+	 * Setter for the job explorer.
+	 * @param jobExplorer the job explorer to set
+	 */
+	public void setJobExplorer(JobExplorer jobExplorer) {
+		this.jobExplorer = jobExplorer;
+	}
+
+	/**
+	 * Setter for the job parameters converter.
+	 * @param jobParametersConverter the job parameters converter to set
+	 */
+	public void setJobParametersConverter(JobParametersConverter jobParametersConverter) {
+		this.jobParametersConverter = jobParametersConverter;
+	}
+
+	/**
+	 * Setter for the transaction manager.
+	 * @param transactionManager the transaction manager to set
+	 */
+	public void setTransactionManager(PlatformTransactionManager transactionManager) {
+		this.transactionManager = transactionManager;
+	}
+
+	@Override
+	public Class<?> getObjectType() {
+		return JobOperator.class;
+	}
+
+	@Override
+	public boolean isSingleton() {
+		return true;
+	}
+
+	@Override
+	public JobOperator getObject() throws Exception {
+		Properties transactionAttributes = new Properties();
+		String transactionProperties = String.join(",", TRANSACTION_PROPAGATION_PREFIX + Propagation.REQUIRED,
+				TRANSACTION_ISOLATION_LEVEL_PREFIX + Isolation.DEFAULT);
+		transactionAttributes.setProperty("stop*", transactionProperties);
+		NameMatchTransactionAttributeSource transactionAttributeSource = new NameMatchTransactionAttributeSource();
+		transactionAttributeSource.setProperties(transactionAttributes);
+		TransactionInterceptor advice = new TransactionInterceptor((TransactionManager) this.transactionManager,
+				transactionAttributeSource);
+		this.proxyFactory.addAdvice(advice);
+		this.proxyFactory.setProxyTargetClass(false);
+		this.proxyFactory.addInterface(JobOperator.class);
+		this.proxyFactory.setTarget(getTarget());
+		return (JobOperator) this.proxyFactory.getProxy(getClass().getClassLoader());
+	}
+
+	private SimpleJobOperator getTarget() throws Exception {
+		SimpleJobOperator simpleJobOperator = new SimpleJobOperator();
+		simpleJobOperator.setJobRegistry(this.jobRegistry);
+		simpleJobOperator.setJobExplorer(this.jobExplorer);
+		simpleJobOperator.setJobRepository(this.jobRepository);
+		simpleJobOperator.setJobLauncher(this.jobLauncher);
+		simpleJobOperator.setJobParametersConverter(this.jobParametersConverter);
+		simpleJobOperator.afterPropertiesSet();
+		return simpleJobOperator;
+	}
+
+}

spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/SimpleJobOperator.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2006-2021 the original author or authors.
+ * Copyright 2006-2022 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.
@@ -74,6 +74,10 @@ import org.springframework.util.Assert;
  * <li>{@link JobRegistry}
  * </ul>
  *
+ * This class can be instantiated with a {@link JobOperatorFactoryBean} to create a
+ * transactional proxy around the job operator.
+ *
+ * @see JobOperatorFactoryBean
  * @author Dave Syer
  * @author Lucas Ward
  * @author Will Schipp
@@ -369,7 +373,6 @@ public class SimpleJobOperator implements JobOperator, InitializingBean {
 	 * @see org.springframework.batch.core.launch.JobOperator#stop(java.lang.Long)
 	 */
 	@Override
-	@Transactional
 	public boolean stop(long executionId) throws NoSuchJobExecutionException, JobExecutionNotRunningException {
 
 		JobExecution jobExecution = findExecutionById(executionId);

spring-batch-core/src/test/java/org/springframework/batch/core/launch/support/JobOperatorFactoryBeanTests.java

@@ -0,0 +1,87 @@
+/*
+ * Copyright 2022 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
+ *
+ *      https://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.launch.support;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import org.springframework.aop.Advisor;
+import org.springframework.aop.framework.Advised;
+import org.springframework.batch.core.configuration.JobRegistry;
+import org.springframework.batch.core.converter.JobParametersConverter;
+import org.springframework.batch.core.explore.JobExplorer;
+import org.springframework.batch.core.launch.JobLauncher;
+import org.springframework.batch.core.launch.JobOperator;
+import org.springframework.batch.core.repository.JobRepository;
+import org.springframework.test.util.AopTestUtils;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.interceptor.TransactionAttributeSource;
+import org.springframework.transaction.interceptor.TransactionInterceptor;
+
+/**
+ * Test class for {@link JobOperatorFactoryBean}.
+ *
+ * @author Mahmoud Ben Hassine
+ */
+class JobOperatorFactoryBeanTests {
+
+	private PlatformTransactionManager transactionManager = Mockito.mock(PlatformTransactionManager.class);
+
+	private JobRepository jobRepository = Mockito.mock(JobRepository.class);
+
+	private JobLauncher jobLauncher = Mockito.mock(JobLauncher.class);
+
+	private JobRegistry jobRegistry = Mockito.mock(JobRegistry.class);
+
+	private JobExplorer jobExplorer = Mockito.mock(JobExplorer.class);
+
+	private JobParametersConverter jobParametersConverter = Mockito.mock(JobParametersConverter.class);
+
+	@Test
+	public void testJobOperatorCreation() throws Exception {
+		// given
+		JobOperatorFactoryBean jobOperatorFactoryBean = new JobOperatorFactoryBean();
+		jobOperatorFactoryBean.setTransactionManager(this.transactionManager);
+		jobOperatorFactoryBean.setJobLauncher(this.jobLauncher);
+		jobOperatorFactoryBean.setJobExplorer(this.jobExplorer);
+		jobOperatorFactoryBean.setJobRegistry(this.jobRegistry);
+		jobOperatorFactoryBean.setJobRepository(this.jobRepository);
+		jobOperatorFactoryBean.setJobParametersConverter(this.jobParametersConverter);
+
+		// when
+		JobOperator jobOperator = jobOperatorFactoryBean.getObject();
+
+		// then
+		Assertions.assertNotNull(jobOperator);
+		Object targetObject = AopTestUtils.getTargetObject(jobOperator);
+		Assertions.assertInstanceOf(SimpleJobOperator.class, targetObject);
+		Assertions.assertEquals(this.transactionManager, getTransactionManagerSetOnJobOperator(jobOperator));
+	}
+
+	private PlatformTransactionManager getTransactionManagerSetOnJobOperator(JobOperator jobOperator) {
+		Advised target = (Advised) jobOperator; // proxy created by
+												// AbstractJobOperatorFactoryBean
+		Advisor[] advisors = target.getAdvisors();
+		for (Advisor advisor : advisors) {
+			if (advisor.getAdvice() instanceof TransactionInterceptor transactionInterceptor) {
+				return (PlatformTransactionManager) transactionInterceptor.getTransactionManager();
+			}
+		}
+		return null;
+	}
+
+}