GH-8748: JDBC locks: use READ_COMMITTED isolation (#8749)

Fixes https://github.com/spring-projects/spring-integration/issues/8748

The Oracle DB throws `ORA-08177: can't serialize access for this transaction`
when other transaction on the row has begun

* Change the isolation for `DefaultLockRepository.acquire()` transaction
to the `READ_COMMITTED` for what database automatically and silently
restarts the entire SQL statement, and no error occurs.
* Add `oracle` dependencies to JDBC module
* Introduce `OracleContainerTest` and implement it for `OracleLockRegistryTests`

**Cherry-pick to `6.1.x` & `6.0.x`**

build.gradle

@@ -91,6 +91,7 @@ ext {
     mockitoVersion = '5.3.1'
     mongoDriverVersion = '4.9.1'
     mysqlVersion = '8.0.33'
+    oracleVersion = '23.3.0.23.09'
     pahoMqttClientVersion = '1.2.5'
     postgresVersion = '42.6.0'
     protobufVersion = '3.23.4'
@@ -724,8 +725,10 @@ project('spring-integration-jdbc') {
         testImplementation "org.apache.commons:commons-dbcp2:$commonsDbcp2Version"
         testImplementation 'org.testcontainers:mysql'
         testImplementation 'org.testcontainers:postgresql'
+        testImplementation 'org.testcontainers:oracle-xe'
 
         testRuntimeOnly 'com.fasterxml.jackson.core:jackson-databind'
+        testRuntimeOnly "com.oracle.database.jdbc:ojdbc11:$oracleVersion"
     }
 }
 

spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/DefaultLockRepository.java

@@ -124,7 +124,7 @@ public class DefaultLockRepository
 
 	private TransactionTemplate readOnlyTransactionTemplate;
 
-	private TransactionTemplate serializableTransactionTemplate;
+	private TransactionTemplate readCommittedTransactionTemplate;
 
 	/**
 	 * Constructor that initializes the client id that will be associated for
@@ -320,9 +320,9 @@ public class DefaultLockRepository
 		this.readOnlyTransactionTemplate = new TransactionTemplate(this.transactionManager, transactionDefinition);
 
 		transactionDefinition.setReadOnly(false);
-		transactionDefinition.setIsolationLevel(TransactionDefinition.ISOLATION_SERIALIZABLE);
+		transactionDefinition.setIsolationLevel(TransactionDefinition.ISOLATION_READ_COMMITTED);
 
-		this.serializableTransactionTemplate = new TransactionTemplate(this.transactionManager, transactionDefinition);
+		this.readCommittedTransactionTemplate = new TransactionTemplate(this.transactionManager, transactionDefinition);
 	}
 
 	@Override
@@ -340,7 +340,7 @@ public class DefaultLockRepository
 	@Override
 	public boolean acquire(String lock) {
 		Boolean result =
-				this.serializableTransactionTemplate.execute(
+				this.readCommittedTransactionTemplate.execute(
 						transactionStatus -> {
 							if (this.template.update(this.updateQuery, this.id, epochMillis(),
 									this.region, lock, this.id, ttlEpochMillis()) > 0) {

spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/oracle/OracleContainerTest.java

@@ -0,0 +1,59 @@
+/*
+ * Copyright 2023 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.integration.jdbc.oracle;
+
+import javax.sql.DataSource;
+
+import org.apache.commons.dbcp2.BasicDataSource;
+import org.junit.jupiter.api.BeforeAll;
+import org.testcontainers.containers.OracleContainer;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.utility.DockerImageName;
+
+/**
+ * The base contract for JUnit tests based on the container for Oracle.
+ * The Testcontainers 'reuse' option must be disabled,so, Ryuk container is started
+ * and will clean all the containers up from this test suite after JVM exit.
+ * Since the Oracle container instance is shared via static property, it is going to be
+ * started only once per JVM, therefore the target Docker container is reused automatically.
+ *
+ * @author Artem Bilan
+ *
+ * @since 6.0.8
+ */
+@Testcontainers(disabledWithoutDocker = true)
+public interface OracleContainerTest {
+
+	OracleContainer ORACLE_CONTAINER =
+			new OracleContainer(DockerImageName.parse("gvenzl/oracle-xe:21-slim-faststart"))
+					.withInitScript("org/springframework/integration/jdbc/schema-oracle.sql");
+
+	@BeforeAll
+	static void startContainer() {
+		ORACLE_CONTAINER.start();
+	}
+
+	static DataSource dataSource() {
+		BasicDataSource dataSource = new BasicDataSource();
+		dataSource.setDriverClassName(ORACLE_CONTAINER.getDriverClassName());
+		dataSource.setUrl(ORACLE_CONTAINER.getJdbcUrl());
+		dataSource.setUsername(ORACLE_CONTAINER.getUsername());
+		dataSource.setPassword(ORACLE_CONTAINER.getPassword());
+		return dataSource;
+	}
+
+}

spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/oracle/OracleLockRegistryTests.java

@@ -0,0 +1,163 @@
+/*
+ * Copyright 2023 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.integration.jdbc.oracle;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.Lock;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.task.AsyncTaskExecutor;
+import org.springframework.core.task.SimpleAsyncTaskExecutor;
+import org.springframework.integration.jdbc.lock.DefaultLockRepository;
+import org.springframework.integration.jdbc.lock.JdbcLockRegistry;
+import org.springframework.integration.jdbc.lock.LockRepository;
+import org.springframework.jdbc.support.JdbcTransactionManager;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+import org.springframework.transaction.PlatformTransactionManager;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+import static org.assertj.core.api.Assertions.assertThatNoException;
+
+/**
+ * @author Artem Bilan
+ *
+ * @since 6.0.8
+ */
+@SpringJUnitConfig
+@DirtiesContext
+public class OracleLockRegistryTests implements OracleContainerTest {
+
+	@Autowired
+	AsyncTaskExecutor taskExecutor;
+
+	@Autowired
+	JdbcLockRegistry registry;
+
+	@Test
+	public void twoThreadsSameLock() throws Exception {
+		final Lock lock1 = this.registry.obtain("foo");
+		final AtomicBoolean locked = new AtomicBoolean();
+		final CountDownLatch latch1 = new CountDownLatch(1);
+		final CountDownLatch latch2 = new CountDownLatch(1);
+		final CountDownLatch latch3 = new CountDownLatch(1);
+		lock1.lockInterruptibly();
+		this.taskExecutor.execute(() -> {
+			Lock lock2 = this.registry.obtain("foo");
+			try {
+				latch1.countDown();
+				lock2.lockInterruptibly();
+				latch2.await(10, TimeUnit.SECONDS);
+				locked.set(true);
+			}
+			catch (InterruptedException e) {
+				Thread.currentThread().interrupt();
+			}
+			finally {
+				lock2.unlock();
+				latch3.countDown();
+			}
+		});
+		assertThat(latch1.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(locked.get()).isFalse();
+		lock1.unlock();
+		latch2.countDown();
+		assertThat(latch3.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(locked.get()).isTrue();
+	}
+
+	@Test
+	public void twoThreadsSecondFailsToGetLock() throws Exception {
+		final Lock lock1 = this.registry.obtain("foo");
+		lock1.lockInterruptibly();
+		final AtomicBoolean locked = new AtomicBoolean();
+		final CountDownLatch latch = new CountDownLatch(1);
+		Future<Object> result = taskExecutor.submit(() -> {
+			Lock lock2 = this.registry.obtain("foo");
+			locked.set(lock2.tryLock(200, TimeUnit.MILLISECONDS));
+			latch.countDown();
+			try {
+				lock2.unlock();
+			}
+			catch (Exception e) {
+				return e;
+			}
+			return null;
+		});
+		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(locked.get()).isFalse();
+		lock1.unlock();
+		Object ise = result.get(10, TimeUnit.SECONDS);
+		assertThat(ise).isInstanceOf(IllegalMonitorStateException.class);
+		assertThat(((Exception) ise).getMessage()).contains("own");
+	}
+
+	@Test
+	public void lockRenewed() {
+		Lock lock = this.registry.obtain("foo");
+
+		assertThat(lock.tryLock()).isTrue();
+
+		assertThatNoException()
+				.isThrownBy(() -> this.registry.renewLock("foo"));
+
+		lock.unlock();
+	}
+
+	@Test
+	public void lockRenewExceptionNotOwned() {
+		this.registry.obtain("foo");
+
+		assertThatExceptionOfType(IllegalMonitorStateException.class)
+				.isThrownBy(() -> this.registry.renewLock("foo"));
+	}
+
+	@Configuration
+	public static class Config {
+
+		@Bean
+		AsyncTaskExecutor taskExecutor() {
+			return new SimpleAsyncTaskExecutor();
+		}
+
+		@Bean
+		public PlatformTransactionManager transactionManager() {
+			return new JdbcTransactionManager(OracleContainerTest.dataSource());
+		}
+
+		@Bean
+		public DefaultLockRepository defaultLockRepository() {
+			return new DefaultLockRepository(OracleContainerTest.dataSource());
+		}
+
+		@Bean
+		public JdbcLockRegistry jdbcLockRegistry(LockRepository lockRepository) {
+			return new JdbcLockRegistry(lockRepository);
+		}
+
+	}
+
+}
+