GH-3683: Always new TX in DefaultLockRepository

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

If a transaction is already active while `JdbcLockRegistry` uses
`DefaultLockRepository` to acquire/release a lock, the repository
must execute SQL queries in a separate transaction to prevent
problems with blocking, deadlocking etc.

It also allows to properly follow transaction isolation level that
is set on some methods of `DefaultLockRepository`.

Previously all methods of DefaultLockRepository supported the current
transaction if there are any. Now all methods will always create new
transaction on each call.

* Change tests to be more unit-ish
* Change `@since` for the `DefaultLockRepositoryTests` to `5.3.10`

**Cherry-pick to `5.4.x` & `5.3.x`**

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

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-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.
@@ -27,6 +27,7 @@ import org.springframework.dao.DuplicateKeyException;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.stereotype.Repository;
 import org.springframework.transaction.annotation.Isolation;
+import org.springframework.transaction.annotation.Propagation;
 import org.springframework.transaction.annotation.Transactional;
 import org.springframework.util.Assert;
 
@@ -43,11 +44,11 @@ import org.springframework.util.Assert;
  * @author Artem Bilan
  * @author Glenn Renfro
  * @author Gary Russell
+ * @author Ruslan Stelmachenko
  *
  * @since 4.3
  */
 @Repository
-@Transactional
 public class DefaultLockRepository implements LockRepository, InitializingBean {
 
 	/**
@@ -142,17 +143,19 @@ public class DefaultLockRepository implements LockRepository, InitializingBean {
 		this.countQuery = String.format(this.countQuery, this.prefix);
 	}
 
+	@Transactional(propagation = Propagation.REQUIRES_NEW)
 	@Override
 	public void close() {
 		this.template.update(this.deleteAllQuery, this.region, this.id);
 	}
 
+	@Transactional(propagation = Propagation.REQUIRES_NEW)
 	@Override
 	public void delete(String lock) {
 		this.template.update(this.deleteQuery, this.region, lock, this.id);
 	}
 
-	@Transactional(isolation = Isolation.SERIALIZABLE, timeout = 1)
+	@Transactional(propagation = Propagation.REQUIRES_NEW, isolation = Isolation.SERIALIZABLE, timeout = 1)
 	@Override
 	public boolean acquire(String lock) {
 		deleteExpired(lock);
@@ -167,6 +170,7 @@ public class DefaultLockRepository implements LockRepository, InitializingBean {
 		}
 	}
 
+	@Transactional(propagation = Propagation.REQUIRES_NEW, readOnly = true)
 	@Override
 	public boolean isAcquired(String lock) {
 		deleteExpired(lock);

spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/lock/DefaultLockRepositoryTests.java

@@ -0,0 +1,90 @@
+/*
+ * Copyright 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.
+ * 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.lock;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+import java.sql.Connection;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+import org.springframework.transaction.annotation.Isolation;
+import org.springframework.transaction.annotation.Transactional;
+import org.springframework.transaction.support.TransactionSynchronization;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
+
+/**
+ * @author Ruslan Stelmachenko
+ *
+ * @since 5.3.10
+ */
+@SpringJUnitConfig(locations = "JdbcLockRegistryTests-context.xml")
+@DirtiesContext
+public class DefaultLockRepositoryTests {
+
+	@Autowired
+	private LockRepository client;
+
+	@BeforeEach
+	public void clear() {
+		this.client.close();
+	}
+
+	@Transactional
+	@Test
+	public void testNewTransactionIsStartedWhenTransactionIsAlreadyActive() {
+		// Make sure a transaction is active
+		assertThat(TransactionSynchronizationManager.isActualTransactionActive()).isTrue();
+
+		TransactionSynchronization transactionSynchronization = spy(TransactionSynchronization.class);
+		TransactionSynchronizationManager.registerSynchronization(transactionSynchronization);
+
+		this.client.acquire("foo"); // 1
+		this.client.delete("foo"); // 2
+		this.client.isAcquired("foo"); // 3
+		this.client.close(); // 4
+
+		// Make sure a transaction is still active
+		assertThat(TransactionSynchronizationManager.isActualTransactionActive()).isTrue();
+		// And was suspended for each invocation of @Transactional methods of DefaultLockRepository,
+		// that confirms that these methods were called in a separate transaction each.
+		verify(transactionSynchronization, times(4)).suspend();
+	}
+
+	@Transactional(isolation = Isolation.REPEATABLE_READ)
+	@Test
+	public void testIsAcquiredFromRepeatableReadTransaction() {
+		// Make sure a transaction with REPEATABLE_READ isolation level is active
+		assertThat(TransactionSynchronizationManager.isActualTransactionActive()).isTrue();
+		assertThat(TransactionSynchronizationManager.getCurrentTransactionIsolationLevel())
+				.isEqualTo(Connection.TRANSACTION_REPEATABLE_READ);
+
+		this.client.acquire("foo");
+		assertThat(this.client.isAcquired("foo")).isTrue();
+
+		this.client.delete("foo");
+		assertThat(this.client.isAcquired("foo")).isFalse();
+	}
+
+}