Added support for specifying a TransactionManager

Before this commit, Spring Cloud Task did not correctly configure the
provided PlatformTransactionManager with the TaskRepository.  This
commit addressis this bug.

Resolves #652

Polished PR on Merge

spring-cloud-task-core/src/main/java/org/springframework/cloud/task/configuration/SimpleTaskAutoConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2019 the original author or authors.
+ * Copyright 2015-2020 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.
@@ -28,7 +28,6 @@ import org.apache.commons.logging.LogFactory;
 import org.springframework.aop.scope.ScopedProxyUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.ApplicationArguments;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.cloud.task.repository.TaskExplorer;
@@ -89,8 +88,7 @@ public class SimpleTaskAutoConfiguration {
 	}
 
 	@Bean
-	@ConditionalOnMissingBean
-	public PlatformTransactionManager transactionManager() {
+	public PlatformTransactionManager springCloudTaskTransactionManager() {
 		return this.platformTransactionManager;
 	}
 

spring-cloud-task-core/src/main/java/org/springframework/cloud/task/repository/TaskRepository.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2019 the original author or authors.
+ * Copyright 2015-2020 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.
@@ -26,6 +26,7 @@ import org.springframework.transaction.annotation.Transactional;
  * information.
  *
  * @author Glenn Renfro
+ * @author Michael Minella
  */
 public interface TaskRepository {
 
@@ -37,7 +38,7 @@ public interface TaskRepository {
 	 * @param exitMessage to be stored for the task.
 	 * @return the updated {@link TaskExecution}
 	 */
-	@Transactional
+	@Transactional("springCloudTaskTransactionManager")
 	TaskExecution completeTaskExecution(long executionId, Integer exitCode, Date endTime,
 			String exitMessage);
 
@@ -51,7 +52,7 @@ public interface TaskRepository {
 	 * @return the updated {@link TaskExecution}
 	 * @since 1.1.0
 	 */
-	@Transactional
+	@Transactional("springCloudTaskTransactionManager")
 	TaskExecution completeTaskExecution(long executionId, Integer exitCode, Date endTime,
 			String exitMessage, String errorMessage);
 
@@ -64,7 +65,7 @@ public interface TaskRepository {
 	 * TaskExecution's taskExecutionId will also contain the id that was used to store the
 	 * TaskExecution.
 	 */
-	@Transactional
+	@Transactional("springCloudTaskTransactionManager")
 	TaskExecution createTaskExecution(TaskExecution taskExecution);
 
 	/**
@@ -75,7 +76,7 @@ public interface TaskRepository {
 	 * @param name task name to be associated with the task execution.
 	 * @return the initial {@link TaskExecution}
 	 */
-	@Transactional
+	@Transactional("springCloudTaskTransactionManager")
 	TaskExecution createTaskExecution(String name);
 
 	/**
@@ -85,7 +86,7 @@ public interface TaskRepository {
 	 * launching, etc).
 	 * @return the initial {@link TaskExecution}
 	 */
-	@Transactional
+	@Transactional("springCloudTaskTransactionManager")
 	TaskExecution createTaskExecution();
 
 	/**
@@ -97,7 +98,7 @@ public interface TaskRepository {
 	 * @param externalExecutionId id assigned to the task by the platform.
 	 * @return TaskExecution created based on the parameters.
 	 */
-	@Transactional
+	@Transactional("springCloudTaskTransactionManager")
 	TaskExecution startTaskExecution(long executionid, String taskName, Date startTime,
 			List<String> arguments, String externalExecutionId);
 
@@ -106,7 +107,7 @@ public interface TaskRepository {
 	 * @param executionid to the task execution to be updated.
 	 * @param externalExecutionId id assigned to the task by the platform.
 	 */
-	@Transactional
+	@Transactional("springCloudTaskTransactionManager")
 	void updateExternalExecutionId(long executionid, String externalExecutionId);
 
 	/**
@@ -120,7 +121,7 @@ public interface TaskRepository {
 	 * @return A TaskExecution that contains the information available at the beginning of
 	 * a TaskExecution.
 	 */
-	@Transactional
+	@Transactional("springCloudTaskTransactionManager")
 	TaskExecution startTaskExecution(long executionid, String taskName, Date startTime,
 			List<String> arguments, String externalExecutionId, Long parentExecutionId);
 

spring-cloud-task-core/src/test/java/org/springframework/cloud/task/configuration/RepositoryTransactionManagerConfigurationTests.java

@@ -0,0 +1,228 @@
+/*
+ * Copyright 2020-2020 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.cloud.task.configuration;
+
+import java.util.ArrayList;
+import java.util.Date;
+
+import javax.sql.DataSource;
+
+import org.junit.Test;
+
+import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.autoconfigure.context.PropertyPlaceholderAutoConfiguration;
+import org.springframework.boot.jdbc.DataSourceBuilder;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.cloud.task.listener.TaskLifecycleListener;
+import org.springframework.cloud.task.repository.TaskExecution;
+import org.springframework.cloud.task.repository.TaskRepository;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.datasource.DataSourceTransactionManager;
+import org.springframework.test.jdbc.JdbcTestUtils;
+import org.springframework.test.util.ReflectionTestUtils;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.support.DefaultTransactionStatus;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Michael Minella
+ */
+public class RepositoryTransactionManagerConfigurationTests {
+
+	@Test
+	public void testZeroCustomTransactionManagerConfiguration() {
+		ApplicationContextRunner applicationContextRunner = new ApplicationContextRunner()
+				.withConfiguration(
+						AutoConfigurations.of(PropertyPlaceholderAutoConfiguration.class,
+								SimpleTaskAutoConfiguration.class,
+								ZeroTransactionManagerConfiguration.class))
+				.withPropertyValues("application.name=transactionManagerTask");
+
+		applicationContextRunner.run((context) -> {
+			DataSource dataSource = context.getBean("dataSource", DataSource.class);
+
+			int taskExecutionCount = JdbcTestUtils
+					.countRowsInTable(new JdbcTemplate(dataSource), "TASK_EXECUTION");
+
+			assertThat(taskExecutionCount).isEqualTo(1);
+		});
+	}
+
+	@Test
+	public void testSingleCustomTransactionManagerConfiguration() {
+		testConfiguration(SingleTransactionManagerConfiguration.class);
+	}
+
+	@Test
+	public void testMultipleCustomTransactionManagerConfiguration() {
+		testConfiguration(MultipleTransactionManagerConfiguration.class);
+	}
+
+	private void testConfiguration(Class configurationClass) {
+		ApplicationContextRunner applicationContextRunner = new ApplicationContextRunner()
+				.withConfiguration(
+						AutoConfigurations.of(PropertyPlaceholderAutoConfiguration.class,
+								SimpleTaskAutoConfiguration.class, configurationClass))
+				.withPropertyValues("application.name=transactionManagerTask");
+
+		applicationContextRunner.run((context) -> {
+			DataSource dataSource = context.getBean("dataSource", DataSource.class);
+
+			int taskExecutionCount = JdbcTestUtils
+					.countRowsInTable(new JdbcTemplate(dataSource), "TASK_EXECUTION");
+
+			// Verify that the create call was rolled back
+			assertThat(taskExecutionCount).isEqualTo(0);
+
+			// Execute a new create call so that things close cleanly
+			TaskRepository taskRepository = context.getBean("taskRepository",
+					TaskRepository.class);
+
+			TaskExecution taskExecution = taskRepository
+					.createTaskExecution("transactionManagerTask");
+			taskExecution = taskRepository.startTaskExecution(
+					taskExecution.getExecutionId(), taskExecution.getTaskName(),
+					new Date(), new ArrayList<>(0), null);
+
+			TaskLifecycleListener listener = context.getBean(TaskLifecycleListener.class);
+
+			ReflectionTestUtils.setField(listener, "taskExecution", taskExecution);
+		});
+	}
+
+	@EnableTask
+	@Configuration
+	public static class ZeroTransactionManagerConfiguration {
+
+		@Bean
+		public TaskConfigurer taskConfigurer(DataSource dataSource) {
+			return new DefaultTaskConfigurer(dataSource);
+		}
+
+		@Bean
+		public DataSource dataSource() {
+			DataSourceBuilder dsb = DataSourceBuilder.create()
+					.url("jdbc:h2:mem:testdb;DB_CLOSE_ON_EXIT=FALSE");
+			dsb.driverClassName("org.h2.Driver");
+			return dsb.build();
+		}
+
+	}
+
+	@EnableTask
+	@Configuration
+	public static class SingleTransactionManagerConfiguration {
+
+		@Bean
+		public TaskConfigurer taskConfigurer(DataSource dataSource,
+				PlatformTransactionManager transactionManager) {
+			return new DefaultTaskConfigurer(dataSource) {
+				@Override
+				public PlatformTransactionManager getTransactionManager() {
+					return transactionManager;
+				}
+			};
+		}
+
+		@Bean
+		public DataSource dataSource() {
+			DataSourceBuilder dsb = DataSourceBuilder.create()
+					.url("jdbc:h2:mem:testdb;DB_CLOSE_ON_EXIT=FALSE");
+			dsb.driverClassName("org.h2.Driver");
+			return dsb.build();
+		}
+
+		@Bean
+		public DataSourceTransactionManager transactionManager(DataSource dataSource) {
+			return new TestDataSourceTransactionManager(dataSource);
+		}
+
+	}
+
+	@EnableTask
+	@Configuration
+	public static class MultipleTransactionManagerConfiguration {
+
+		@Bean
+		public TaskConfigurer taskConfigurer(DataSource dataSource,
+				PlatformTransactionManager transactionManager) {
+			return new DefaultTaskConfigurer(dataSource) {
+				@Override
+				public PlatformTransactionManager getTransactionManager() {
+					return transactionManager;
+				}
+			};
+		}
+
+		@Bean
+		public DataSource dataSource() {
+			DataSourceBuilder dsb = DataSourceBuilder.create()
+					.url("jdbc:h2:mem:testdb;DB_CLOSE_ON_EXIT=FALSE");
+			dsb.driverClassName("org.h2.Driver");
+			return dsb.build();
+		}
+
+		@Bean
+		public DataSourceTransactionManager transactionManager(DataSource dataSource) {
+			return new TestDataSourceTransactionManager(dataSource);
+		}
+
+		@Bean
+		public DataSource dataSource2() {
+			DataSourceBuilder dsb = DataSourceBuilder.create()
+					.url("jdbc:h2:mem:testdb2;DB_CLOSE_ON_EXIT=FALSE");
+			dsb.driverClassName("org.h2.Driver");
+			return dsb.build();
+		}
+
+		@Bean
+		public DataSourceTransactionManager transactionManager2(DataSource dataSource2) {
+			return new DataSourceTransactionManager(dataSource2);
+		}
+
+	}
+
+	private static class TestDataSourceTransactionManager
+			extends DataSourceTransactionManager {
+
+		protected TestDataSourceTransactionManager(DataSource dataSource) {
+			super(dataSource);
+		}
+
+		private int count = 0;
+
+		@Override
+		protected void doCommit(DefaultTransactionStatus status) {
+
+			if (count == 0) {
+				// Rollback the finish of the task
+				super.doRollback(status);
+			}
+			else {
+				// Commit the start of the task
+				super.doCommit(status);
+			}
+
+			count++;
+		}
+
+	}
+
+}