Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-3733 Configure TxManager for DefLockRepository (#3782)

Browse Source 
* GH-3733 Configure TxManager for DefLockRepository

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

The `@Transactional` resolves a primary `TransactionManager` bean
from the application context which might not be sufficient for all
the use-case.

To make it work with the custom (or specific) `TransactionManager`
we have to extend a `DefaultLockRepository` and override all those
`@Transactional` method

* Change the logic of the `DefaultLockRepository` from `@Transactional`
to the `TransactionTemplate` and use provided `TransactionManager`
or resolve one from the application context
* Adjust tests to use explicit `TransactionManager` and call
`afterSingletonsInstantiated()` to initialize a default `TransactionTemplate`
* Mention the change in the docs

* * Extracted all the `TransactionTemplate`s to the properties for caching
* Add `BeanInitializationException` for no-unique `PlatformTransactionManager`
bean in the `afterSingletonsInstantiated()`

* Fix language in the exception message

Co-authored-by: Gary Russell <grussell@vmware.com>

Co-authored-by: Gary Russell <grussell@vmware.com>
This commit is contained in:
Artem Bilan
2022-04-12 12:13:15 -04:00
committed by GitHub
parent cd84f1699a
commit 4b57363a05
6 changed files with 139 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2021 the original author or authors.
* Copyright 2016-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.
@@ -21,14 +21,18 @@ import java.util.UUID;
import javax.sql.DataSource;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanInitializationException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
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.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.support.DefaultTransactionDefinition;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.Assert;
/**
@@ -36,7 +40,7 @@ import org.springframework.util.Assert;
* table from the script presented in the {@code org/springframework/integration/jdbc/schema-*.sql}.
* <p>
* This repository can't be shared between different {@link JdbcLockRegistry} instances.
* Otherwise it opens a possibility to break {@link java.util.concurrent.locks.Lock} contract,
* Otherwise, it opens a possibility to break {@link java.util.concurrent.locks.Lock} contract,
* where {@link JdbcLockRegistry} uses non-shared {@link java.util.concurrent.locks.ReentrantLock}s
* for local synchronizations.
*
@@ -49,8 +53,8 @@ import org.springframework.util.Assert;
*
* @since 4.3
*/
@Repository
public class DefaultLockRepository implements LockRepository, InitializingBean {
public class DefaultLockRepository
implements LockRepository, InitializingBean, ApplicationContextAware, SmartInitializingSingleton {
/**
* Default value for the table prefix property.
@@ -89,12 +93,21 @@ public class DefaultLockRepository implements LockRepository, InitializingBean {
private String renewQuery = "UPDATE %sLOCK SET CREATED_DATE=? WHERE REGION=? AND LOCK_KEY=? AND CLIENT_ID=?";
private ApplicationContext applicationContext;
private PlatformTransactionManager transactionManager;
private TransactionTemplate defaultTransactionTemplate;
private TransactionTemplate readOnlyTransactionTemplate;
private TransactionTemplate serializableTransactionTemplate;
/**
* Constructor that initializes the client id that will be associated for
* all the locks persisted by the store instance to a random {@link UUID}.
* @param dataSource the {@link DataSource} used to maintain the lock repository.
*/
@Autowired
public DefaultLockRepository(DataSource dataSource) {
this(dataSource, UUID.randomUUID().toString());
}
@@ -124,21 +137,37 @@ public class DefaultLockRepository implements LockRepository, InitializingBean {
}
/**
* Specify the prefix for target data base table used from queries.
* @param prefix the prefix to set (default INT_).
* Specify the prefix for target database table used from queries.
* @param prefix the prefix to set (default {@code INT_}).
*/
public void setPrefix(String prefix) {
this.prefix = prefix;
}
/**
* Specify the time (in milliseconds) to expire dead locks.
* @param timeToLive the time to expire dead locks.
* Specify the time (in milliseconds) to expire dead-locks.
* @param timeToLive the time to expire dead-locks.
*/
public void setTimeToLive(int timeToLive) {
this.ttl = timeToLive;
}
/**
* Set a {@link PlatformTransactionManager} for operations.
* Otherwise, a primary {@link PlatformTransactionManager} bean is obtained
* from the application context.
* @param transactionManager the {@link PlatformTransactionManager} to use.
* @since 6.0
*/
public void setTransactionManager(PlatformTransactionManager transactionManager) {
this.transactionManager = transactionManager;
}
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
this.applicationContext = applicationContext;
}
@Override
public void afterPropertiesSet() {
this.deleteQuery = String.format(this.deleteQuery, this.prefix);
@@ -150,50 +179,84 @@ public class DefaultLockRepository implements LockRepository, InitializingBean {
this.renewQuery = String.format(this.renewQuery, this.prefix);
}
@Transactional(propagation = Propagation.REQUIRES_NEW)
@Override
public void afterSingletonsInstantiated() {
if (this.transactionManager == null) {
try {
this.transactionManager = this.applicationContext.getBean(PlatformTransactionManager.class);
}
catch (BeansException ex) {
throw new BeanInitializationException(
"A unique or primary 'PlatformTransactionManager' bean " +
"must be present in the application context.", ex);
}
}
DefaultTransactionDefinition transactionDefinition =
new DefaultTransactionDefinition(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
this.defaultTransactionTemplate =
new TransactionTemplate(this.transactionManager, transactionDefinition);
// It is safe to reuse the transactionDefinition - the TransactionTemplate makes copy of its properties.
transactionDefinition.setReadOnly(true);
this.readOnlyTransactionTemplate = new TransactionTemplate(this.transactionManager, transactionDefinition);
transactionDefinition.setReadOnly(false);
transactionDefinition.setIsolationLevel(TransactionDefinition.ISOLATION_SERIALIZABLE);
this.serializableTransactionTemplate = new TransactionTemplate(this.transactionManager, transactionDefinition);
}
@Override
public void close() {
this.template.update(this.deleteAllQuery, this.region, this.id);
this.defaultTransactionTemplate.executeWithoutResult(
transactionStatus -> 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);
this.defaultTransactionTemplate.executeWithoutResult(
transactionStatus -> this.template.update(this.deleteQuery, this.region, lock, this.id));
}
@Transactional(propagation = Propagation.REQUIRES_NEW, isolation = Isolation.SERIALIZABLE)
@Override
public boolean acquire(String lock) {
if (this.template.update(this.updateQuery, this.id, new Date(), this.region, lock, this.id,
new Date(System.currentTimeMillis() - this.ttl)) > 0) {
return true;
}
try {
return this.template.update(this.insertQuery, this.region, lock, this.id, new Date()) > 0;
}
catch (DuplicateKeyException e) {
return false;
}
return this.serializableTransactionTemplate.execute(transactionStatus -> {
if (this.template.update(this.updateQuery, this.id, new Date(), this.region, lock, this.id,
new Date(System.currentTimeMillis() - this.ttl)) > 0) {
return true;
}
try {
return this.template.update(this.insertQuery, this.region, lock, this.id, new Date()) > 0;
}
catch (DuplicateKeyException e) {
return false;
}
});
}
@Transactional(propagation = Propagation.REQUIRES_NEW, readOnly = true)
@Override
public boolean isAcquired(String lock) {
return this.template.queryForObject(this.countQuery, Integer.class, // NOSONAR query never returns null
this.region, lock, this.id, new Date(System.currentTimeMillis() - this.ttl)) == 1;
return this.readOnlyTransactionTemplate.execute(transactionStatus ->
this.template.queryForObject(this.countQuery, // NOSONAR query never returns null
Integer.class, this.region, lock, this.id, new Date(System.currentTimeMillis() - this.ttl))
== 1);
}
@Transactional(propagation = Propagation.REQUIRES_NEW)
@Override
public void deleteExpired() {
this.template.update(this.deleteExpiredQuery, this.region, new Date(System.currentTimeMillis() - this.ttl));
this.defaultTransactionTemplate.executeWithoutResult(
transactionStatus ->
this.template.update(this.deleteExpiredQuery, this.region,
new Date(System.currentTimeMillis() - this.ttl)));
}
@Transactional(propagation = Propagation.REQUIRES_NEW)
@Override
public boolean renew(String lock) {
return this.template.update(this.renewQuery, new Date(), this.region, lock, this.id) > 0;
return this.defaultTransactionTemplate.execute(
transactionStatus -> this.template.update(this.renewQuery, new Date(), this.region, lock, this.id) > 0);
}
}

19
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/leader/JdbcLockRegistryLeaderInitiatorTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2019 the original author or authors.
* Copyright 2016-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.
@@ -23,10 +23,10 @@ import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.springframework.integration.jdbc.lock.DefaultLockRepository;
import org.springframework.integration.jdbc.lock.JdbcLockRegistry;
@@ -34,6 +34,7 @@ import org.springframework.integration.leader.Context;
import org.springframework.integration.leader.DefaultCandidate;
import org.springframework.integration.leader.event.LeaderEventPublisher;
import org.springframework.integration.support.leader.LockRegistryLeaderInitiator;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase;
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType;
@@ -49,7 +50,7 @@ public class JdbcLockRegistryLeaderInitiatorTests {
public static EmbeddedDatabase dataSource;
@BeforeClass
@BeforeAll
public static void init() {
dataSource = new EmbeddedDatabaseBuilder()
.setType(EmbeddedDatabaseType.H2)
@@ -58,7 +59,7 @@ public class JdbcLockRegistryLeaderInitiatorTests {
.build();
}
@AfterClass
@AfterAll
public static void destroy() {
dataSource.shutdown();
}
@@ -70,7 +71,9 @@ public class JdbcLockRegistryLeaderInitiatorTests {
List<LockRegistryLeaderInitiator> initiators = new ArrayList<>();
for (int i = 0; i < 2; i++) {
DefaultLockRepository lockRepository = new DefaultLockRepository(dataSource);
lockRepository.setTransactionManager(new DataSourceTransactionManager(dataSource));
lockRepository.afterPropertiesSet();
lockRepository.afterSingletonsInstantiated();
LockRegistryLeaderInitiator initiator = new LockRegistryLeaderInitiator(
new JdbcLockRegistry(lockRepository),
new DefaultCandidate("foo", "bar"));
@@ -170,7 +173,7 @@ public class JdbcLockRegistryLeaderInitiatorTests {
}
@Test
@Ignore("Looks like an embedded DBd is not fully cleared if we don't close application context")
@Disabled("Looks like an embedded DBd is not fully cleared if we don't close application context")
public void testLostConnection() throws InterruptedException {
CountDownLatch granted = new CountDownLatch(1);
CountingPublisher countingPublisher = new CountingPublisher(granted);

18
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/lock/JdbcLockRegistryDifferentClientTests.java
View File

@@ -79,7 +79,7 @@ public class JdbcLockRegistryDifferentClientTests {
this.registry.expireUnusedOlderThan(0);
this.client.close();
this.child = new AnnotationConfigApplicationContext();
this.child.register(DefaultLockRepository.class);
this.child.registerBean("childLockRepository", DefaultLockRepository.class, this.dataSource);
this.child.setParent(this.context);
this.child.refresh();
}
@@ -195,7 +195,9 @@ public class JdbcLockRegistryDifferentClientTests {
for (int j = 0; j < 20; j++) {
Callable<Boolean> task = () -> {
DefaultLockRepository client = new DefaultLockRepository(this.dataSource);
client.setApplicationContext(this.context);
client.afterPropertiesSet();
client.afterSingletonsInstantiated();
Lock lock = new JdbcLockRegistry(client).obtain("foo");
try {
if (locked.isEmpty() && lock.tryLock(Long.MAX_VALUE, TimeUnit.MILLISECONDS)) {
@@ -231,9 +233,13 @@ public class JdbcLockRegistryDifferentClientTests {
@Test
public void testExclusiveAccess() throws Exception {
DefaultLockRepository client1 = new DefaultLockRepository(dataSource);
client1.setApplicationContext(this.context);
client1.afterPropertiesSet();
final DefaultLockRepository client2 = new DefaultLockRepository(dataSource);
client1.afterSingletonsInstantiated();
DefaultLockRepository client2 = new DefaultLockRepository(dataSource);
client2.setApplicationContext(this.context);
client2.afterPropertiesSet();
client2.afterSingletonsInstantiated();
Lock lock1 = new JdbcLockRegistry(client1).obtain("foo");
final BlockingQueue<Integer> data = new LinkedBlockingQueue<>();
final CountDownLatch latch1 = new CountDownLatch(1);
@@ -278,10 +284,14 @@ public class JdbcLockRegistryDifferentClientTests {
public void testOutOfDateLockTaken() throws Exception {
DefaultLockRepository client1 = new DefaultLockRepository(dataSource);
client1.setTimeToLive(100);
client1.setApplicationContext(this.context);
client1.afterPropertiesSet();
client1.afterSingletonsInstantiated();
DefaultLockRepository client2 = new DefaultLockRepository(dataSource);
client2.setTimeToLive(100);
client2.setApplicationContext(this.context);
client2.afterPropertiesSet();
client2.afterSingletonsInstantiated();
Lock lock1 = new JdbcLockRegistry(client1).obtain("foo");
final BlockingQueue<Integer> data = new LinkedBlockingQueue<>();
final CountDownLatch latch = new CountDownLatch(1);
@@ -316,10 +326,14 @@ public class JdbcLockRegistryDifferentClientTests {
public void testRenewLock() throws Exception {
DefaultLockRepository client1 = new DefaultLockRepository(dataSource);
client1.setTimeToLive(500);
client1.setApplicationContext(this.context);
client1.afterPropertiesSet();
client1.afterSingletonsInstantiated();
DefaultLockRepository client2 = new DefaultLockRepository(dataSource);
client2.setTimeToLive(500);
client2.setApplicationContext(this.context);
client2.afterPropertiesSet();
client2.afterSingletonsInstantiated();
JdbcLockRegistry registry = new JdbcLockRegistry(client1);
Lock lock1 = registry.obtain("foo");
final BlockingQueue<Integer> data = new LinkedBlockingQueue<>();

8
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/lock/JdbcLockRegistryTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2021 the original author or authors.
* Copyright 2016-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.
@@ -36,6 +36,7 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.core.task.AsyncTaskExecutor;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.integration.test.util.TestUtils;
@@ -67,6 +68,9 @@ public class JdbcLockRegistryTests {
@Autowired
private DataSource dataSource;
@Autowired
private ApplicationContext context;
@BeforeEach
public void clear() {
this.registry.expireUnusedOlderThan(0);
@@ -143,7 +147,9 @@ public class JdbcLockRegistryTests {
public void testReentrantLockAfterExpiration() throws Exception {
DefaultLockRepository client = new DefaultLockRepository(dataSource);
client.setTimeToLive(1);
client.setApplicationContext(this.context);
client.afterPropertiesSet();
client.afterSingletonsInstantiated();
JdbcLockRegistry registry = new JdbcLockRegistry(client);
Lock lock1 = registry.obtain("foo");
assertThat(lock1.tryLock()).isTrue();

2
src/reference/asciidoc/jdbc.adoc
View File

@@ -1092,6 +1092,8 @@ NOTE: The lock renewal can be done only if the lock is held by the current threa
String with version 5.5.6, the `JdbcLockRegistry` is support automatically clean up cache for JdbcLock in `JdbcLockRegistry.locks` via `JdbcLockRegistry.setCacheCapacity()`.
See its JavaDocs for more information.
String with version 6.0, the `DefaultLockRepository` can be supplied with a `PlatformTransactionManager` instead of relying on the primary bean from the application context.
[[jdbc-metadata-store]]
=== JDBC Metadata Store

4
src/reference/asciidoc/whats-new.adoc
View File

@@ -48,3 +48,7 @@ When providing a `RetryTemplate` on the inbound gateway or message-driven channe
In addition, the new `KafkaErrorMessageSendingRecoverer` is provided; this can be used with a `DefaultErrorHandler` to avoid issues with long aggregated retry delays causing partitions rebalances.
See <<./kafka.adoc#kafka,Spring for Apache Kafka Support>> for more information.
=== JDBC Changes
The `DefaultLockRepository` can now be supplied with a `PlatformTransactionManager` instead of relying on the primary bean from the application context.