User can now select transaction Isolation for CTR

resolves #4680

Author: cppwfs

result.txt

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2020 the original author or authors.
+ * Copyright 2017-2021 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.
@@ -35,14 +35,15 @@
 import org.springframework.boot.autoconfigure.batch.BasicBatchConfigurer;
 import org.springframework.boot.autoconfigure.batch.BatchProperties;
 import org.springframework.boot.autoconfigure.transaction.TransactionManagerCustomizers;
+import org.springframework.cloud.dataflow.composedtaskrunner.properties.ComposedTaskProperties;
 import org.springframework.cloud.dataflow.composedtaskrunner.support.ComposedTaskException;
 import org.springframework.cloud.dataflow.composedtaskrunner.support.SqlServerSequenceMaxValueIncrementer;
 import org.springframework.jdbc.support.MetaDataAccessException;
 import org.springframework.jdbc.support.incrementer.DataFieldMaxValueIncrementer;
 import org.springframework.util.StringUtils;
 
 /**
- * A BatchConfigurer for CTR that will establish the transaction isolation lavel to READ_COMMITTED.
+ * A BatchConfigurer for CTR that will establish the transaction isolation level to ISOLATION_REPEATABLE_READ by default.
  *
  * @author Glenn Renfro
  */
@@ -54,6 +55,8 @@ public class ComposedBatchConfigurer extends BasicBatchConfigurer {
 
 	private Map<String, DataFieldMaxValueIncrementer> incrementerMap;
 
+	private ComposedTaskProperties composedTaskProperties;
+
 	/**
 	 * Create a new {@link BasicBatchConfigurer} instance.
 	 *
@@ -62,10 +65,12 @@ public class ComposedBatchConfigurer extends BasicBatchConfigurer {
 	 * @param transactionManagerCustomizers transaction manager customizers (or
 	 *                                      {@code null})
 	 */
-	protected ComposedBatchConfigurer(BatchProperties properties, DataSource dataSource, TransactionManagerCustomizers transactionManagerCustomizers) {
+	protected ComposedBatchConfigurer(BatchProperties properties, DataSource dataSource,
+			TransactionManagerCustomizers transactionManagerCustomizers, ComposedTaskProperties composedTaskProperties) {
 		super(properties, dataSource, transactionManagerCustomizers);
 		this.incrementerDataSource = dataSource;
 		incrementerMap = new HashMap<>();
+		this.composedTaskProperties = composedTaskProperties;
 	}
 
 	protected JobRepository createJobRepository() {
@@ -85,7 +90,7 @@ public DataFieldMaxValueIncrementer getIncrementer(String incrementerType, Strin
 		factory.setIncrementerFactory(incrementerFactory);
 		factory.setDataSource(this.incrementerDataSource);
 		factory.setTransactionManager(this.getTransactionManager());
-		factory.setIsolationLevelForCreate("ISOLATION_REPEATABLE_READ");
+		factory.setIsolationLevelForCreate(this.composedTaskProperties.getTransactionIsolationLevel());
 		try {
 			factory.afterPropertiesSet();
 			return factory.getObject();

spring-cloud-dataflow-composed-task-runner/src/main/java/org/springframework/cloud/dataflow/composedtaskrunner/ComposedBatchConfigurer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2020 the original author or authors.
+ * Copyright 2017-2021 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.
@@ -21,7 +21,6 @@
 import org.springframework.batch.core.StepExecutionListener;
 import org.springframework.batch.core.configuration.annotation.BatchConfigurer;
 import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
-import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.batch.BatchProperties;
 import org.springframework.boot.autoconfigure.transaction.TransactionManagerCustomizers;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
@@ -46,22 +45,19 @@
 @Import(org.springframework.cloud.dataflow.composedtaskrunner.StepBeanDefinitionRegistrar.class)
 public class ComposedTaskRunnerConfiguration {
 
-	@Autowired
-	private ComposedTaskProperties properties;
-
 	@Bean
 	public StepExecutionListener composedTaskStepExecutionListener(TaskExplorer taskExplorer){
 		return new org.springframework.cloud.dataflow.composedtaskrunner.ComposedTaskStepExecutionListener(taskExplorer);
 	}
 
 	@Bean
-	public org.springframework.cloud.dataflow.composedtaskrunner.ComposedRunnerJobFactory composedTaskJob() {
+	public org.springframework.cloud.dataflow.composedtaskrunner.ComposedRunnerJobFactory composedTaskJob(ComposedTaskProperties properties) {
 
-		return new org.springframework.cloud.dataflow.composedtaskrunner.ComposedRunnerJobFactory(this.properties);
+		return new org.springframework.cloud.dataflow.composedtaskrunner.ComposedRunnerJobFactory(properties);
 	}
 
 	@Bean
-	public TaskExecutor taskExecutor() {
+	public TaskExecutor taskExecutor(ComposedTaskProperties properties) {
 		ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
 		taskExecutor.setCorePoolSize(properties.getSplitThreadCorePoolSize());
 		taskExecutor.setMaxPoolSize(properties.getSplitThreadMaxPoolSize());
@@ -75,7 +71,10 @@ public TaskExecutor taskExecutor() {
 	}
 
 	@Bean
-	public BatchConfigurer getComposedBatchConfigurer(BatchProperties properties, DataSource dataSource, TransactionManagerCustomizers transactionManagerCustomizers) {
-		return new org.springframework.cloud.dataflow.composedtaskrunner.ComposedBatchConfigurer(properties, dataSource, transactionManagerCustomizers);
+	public BatchConfigurer getComposedBatchConfigurer(BatchProperties properties,
+			DataSource dataSource, TransactionManagerCustomizers transactionManagerCustomizers,
+			ComposedTaskProperties composedTaskProperties) {
+		return new org.springframework.cloud.dataflow.composedtaskrunner.ComposedBatchConfigurer(properties,
+				dataSource, transactionManagerCustomizers, composedTaskProperties);
 	}
 }

spring-cloud-dataflow-composed-task-runner/src/main/java/org/springframework/cloud/dataflow/composedtaskrunner/ComposedTaskRunnerConfiguration.java

@@ -199,6 +199,12 @@ public class ComposedTaskProperties {
 	 */
 	private boolean skipTlsCertificateVerification = false;
 
+	/**
+	 * Establish the transaction isolation level for the Composed Task Runner.
+	 * Default is ISOLATION_REPEATABLE_READ.
+	 */
+	private String transactionIsolationLevel = "ISOLATION_REPEATABLE_READ";
+
 	public ComposedTaskProperties() {
 		try {
 			this.dataflowServerUri = new URI("http://localhost:9393");
@@ -407,4 +413,12 @@ public boolean isUuidInstanceEnabled() {
 	public void setUuidInstanceEnabled(boolean uuIdInstanceEnabled) {
 		this.uuidInstanceEnabled = uuIdInstanceEnabled;
 	}
+
+	public String getTransactionIsolationLevel() {
+		return transactionIsolationLevel;
+	}
+
+	public void setTransactionIsolationLevel(String transactionIsolationLevel) {
+		this.transactionIsolationLevel = transactionIsolationLevel;
+	}
 }

spring-cloud-dataflow-composed-task-runner/src/main/java/org/springframework/cloud/dataflow/composedtaskrunner/properties/ComposedTaskProperties.java

@@ -113,6 +113,12 @@
       "description": "The platform property that will be used for each task in the workflow when it is launched.",
       "sourceType": "org.springframework.cloud.dataflow.composedtaskrunner.properties.ComposedTaskProperties"
     },
+    {
+      "name": "transaction-isolation-level",
+      "type": "java.lang.String",
+      "description": "Establish the transaction isolation level for the Composed Task Runner.  Default is ISOLATION_REPEATABLE_READ",
+      "sourceType": "org.springframework.cloud.dataflow.composedtaskrunner.properties.ComposedTaskProperties"
+    },
     {
       "name": "skip-tls-certificate-verification",
       "type": "java.lang.Boolean",

spring-cloud-dataflow-composed-task-runner/src/main/resources/META-INF/spring-configuration-metadata.json

@@ -33,6 +33,7 @@
 import org.springframework.boot.autoconfigure.jdbc.EmbeddedDataSourceConfiguration;
 import org.springframework.cloud.common.security.CommonSecurityAutoConfiguration;
 import org.springframework.cloud.dataflow.composedtaskrunner.configuration.DataFlowTestConfiguration;
+import org.springframework.cloud.dataflow.composedtaskrunner.properties.ComposedTaskProperties;
 import org.springframework.cloud.dataflow.rest.client.TaskOperations;
 import org.springframework.context.ApplicationContext;
 import org.springframework.test.annotation.DirtiesContext;
@@ -42,6 +43,7 @@
 import org.springframework.test.util.ReflectionTestUtils;
 import org.springframework.util.Assert;
 
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 
@@ -66,6 +68,9 @@ public class ComposedTaskRunnerConfigurationNoPropertiesTests {
 	@Autowired
 	private ApplicationContext context;
 
+	@Autowired
+	private ComposedTaskProperties composedTaskProperties;
+
 	@Test
 	@DirtiesContext
 	public void testComposedConfiguration() throws Exception {
@@ -75,6 +80,7 @@ public void testComposedConfiguration() throws Exception {
 		TaskOperations taskOperations = mock(TaskOperations.class);
 		ReflectionTestUtils.setField(ctrStep.getTasklet(), "taskOperations", taskOperations);
 		job.execute(jobExecution);
+		assertThat(composedTaskProperties.getTransactionIsolationLevel()).isEqualTo("ISOLATION_REPEATABLE_READ");
 
 		Assert.notNull(job.getJobParametersIncrementer(), "JobParametersIncrementer must not be null.");
 		verify(taskOperations).launch("AAA", new HashMap<>(0), new ArrayList<>(0));

spring-cloud-dataflow-composed-task-runner/src/test/java/org/springframework/cloud/dataflow/composedtaskrunner/ComposedTaskRunnerConfigurationNoPropertiesTests.java

@@ -59,6 +59,7 @@
 @TestPropertySource(properties = {"graph=ComposedTest-AAA && ComposedTest-BBB && ComposedTest-CCC","max-wait-time=1010",
 		"composed-task-properties=" + ComposedTaskRunnerConfigurationWithPropertiesTests.COMPOSED_TASK_PROPS ,
 		"interval-time-between-checks=1100", "composed-task-arguments=--baz=boo --AAA.foo=bar BBB.que=qui",
+		"transaction-isolation-level=ISOLATION_READ_COMMITTED",
 		"dataflow-server-uri=https://bar", "spring.cloud.task.name=ComposedTest"})
 @EnableAutoConfiguration(exclude = { CommonSecurityAutoConfiguration.class})
 public class ComposedTaskRunnerConfigurationWithPropertiesTests {
@@ -99,6 +100,8 @@ public void testComposedConfiguration() throws Exception {
 		assertThat(composedTaskProperties.getMaxWaitTime()).isEqualTo(1010);
 		assertThat(composedTaskProperties.getIntervalTimeBetweenChecks()).isEqualTo(1100);
 		assertThat(composedTaskProperties.getDataflowServerUri().toASCIIString()).isEqualTo("https://bar");
+		assertThat(composedTaskProperties.getTransactionIsolationLevel()).isEqualTo("ISOLATION_READ_COMMITTED");
+
 		List<String> args = new ArrayList<>(1);
 		args.add("--baz=boo --foo=bar");
 		Assert.notNull(job.getJobParametersIncrementer(), "JobParametersIncrementer must not be null.");

spring-cloud-dataflow-composed-task-runner/src/test/java/org/springframework/cloud/dataflow/composedtaskrunner/ComposedTaskRunnerConfigurationWithPropertiesTests.java

@@ -578,6 +578,11 @@ Set this option to `true` when running multiple instances of the same composed t
 The amount of time, in milliseconds, that the `ComposedTaskRunner` waits between checks of the database to see if a task has completed. (Integer, default: `10000`).
 `ComposedTaskRunner` uses the datastore to determine the status of each child tasks. This interval indicates to `ComposedTaskRunner` how often it should check the status its child tasks.
 
+* `transaction-isolation-level`
+Establish the transaction isolation level for the Composed Task Runner.
+A list of available transaction isolation levels can be found  https://docs.spring.io/spring-framework/docs/current/javadoc-api/org/springframework/transaction/TransactionDefinition.html[here].
+Default is `ISOLATION_REPEATABLE_READ`.
+
 * `max-wait-time`
 The maximum amount of time, in milliseconds, that an individual step can run before the execution of the Composed task is failed (Integer, default: 0).
 Determines the maximum time each child task is allowed to run before the CTR ends with a failure. The default of `0` indicates no timeout.

spring-cloud-dataflow-docs/src/main/asciidoc/tasks.adoc

@@ -0,0 +1,3 @@
+task.ssbjp=maven://org.springframework.cloud:spring-cloud-dataflow-single-step-batch-job:2.9.0-SNAPSHOT
+task.ssbjp.metadata=maven://org.springframework.cloud:spring-cloud-dataflow-single-step-batch-job:jar:metadata:2.9.0-SNAPSHOT
+

ssbj.properties

@@ -0,0 +1,3 @@
+abc
+def
+ghi