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

GH-3272: Add lease renewal for distributed locks

Browse Source 
Fixes https://github.com/spring-projects/spring-integration/issues/3272

* Introduce a `RenewableLockRegistry` since not all `LockRegistry` implementations
provide a way to renew the lease for the lock
* Implement `RenewableLockRegistry` in the `JdbcLockRegistry`
* Test and document the feature
This commit is contained in:
Alexandre Strubel
2020-07-01 15:17:37 +02:00
committed by Artem Bilan
parent 381a071287
commit e2c6e77f2f
8 changed files with 250 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
spring-integration-core/src/main/java/org/springframework/integration/support/locks/RenewableLockRegistry.java Normal file
View File

@@ -0,0 +1,37 @@
/*
* Copyright 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.integration.support.locks;
/**
* A {@link LockRegistry} implementing this interface supports the renewal
* of the time to live of a lock.
*
* @author Alexandre Strubel
* @author Artem Bilan
*
* @since 5.4
*/
public interface RenewableLockRegistry extends LockRegistry {
/**
* Renew the time to live of the lock is associated with the parameter object.
* The lock must be held by the current thread
* @param lockKey The object with which the lock is associated.
*/
void renewLock(Object lockKey);
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2019 the original author or authors.
* Copyright 2016-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.
@@ -43,6 +43,7 @@ import org.springframework.util.Assert;
* @author Artem Bilan
* @author Glenn Renfro
* @author Gary Russell
* @author Alexandre Strubel
*
* @since 4.3
*/
@@ -80,7 +81,8 @@ public class DefaultLockRepository implements LockRepository, InitializingBean {
private String insertQuery = "INSERT INTO %sLOCK (REGION, LOCK_KEY, CLIENT_ID, CREATED_DATE) VALUES (?, ?, ?, ?)";
private String countQuery = "SELECT COUNT(REGION) FROM %sLOCK WHERE REGION=? AND LOCK_KEY=? AND CLIENT_ID=? AND CREATED_DATE>=?";
private String countQuery =
"SELECT COUNT(REGION) FROM %sLOCK WHERE REGION=? AND LOCK_KEY=? AND CLIENT_ID=? AND CREATED_DATE>=?";
/**
* Constructor that initializes the client id that will be associated for
@@ -108,7 +110,7 @@ public class DefaultLockRepository implements LockRepository, InitializingBean {
/**
* A unique grouping identifier for all locks persisted with this store. Using
* multiple regions allows the store to be partitioned (if necessary) for different
* purposes. Defaults to <code>DEFAULT</code>.
* purposes. Defaults to {@code DEFAULT}.
* @param region the region name to set
*/
public void setRegion(String region) {
@@ -179,4 +181,9 @@ public class DefaultLockRepository implements LockRepository, InitializingBean {
new Date(System.currentTimeMillis() - this.ttl));
}
@Override
public boolean renew(String lock) {
return this.template.update(this.updateQuery, new Date(), this.region, lock, this.id) > 0;
}
}

42
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/JdbcLockRegistry.java
View File

@@ -30,6 +30,7 @@ import org.springframework.dao.CannotAcquireLockException;
import org.springframework.dao.DataAccessResourceFailureException;
import org.springframework.dao.TransientDataAccessException;
import org.springframework.integration.support.locks.ExpirableLockRegistry;
import org.springframework.integration.support.locks.RenewableLockRegistry;
import org.springframework.integration.util.UUIDConverter;
import org.springframework.transaction.TransactionTimedOutException;
import org.springframework.util.Assert;
@@ -49,10 +50,11 @@ import org.springframework.util.Assert;
* @author Bartosz Rempuszewski
* @author Gary Russell
* @author Alexandre Strubel
* @author Stefan Vassilev
*
* @since 4.3
*/
public class JdbcLockRegistry implements ExpirableLockRegistry {
public class JdbcLockRegistry implements ExpirableLockRegistry, RenewableLockRegistry {
private static final int DEFAULT_IDLE = 100;
@@ -101,6 +103,19 @@ public class JdbcLockRegistry implements ExpirableLockRegistry {
}
}
@Override
public void renewLock(Object lockKey) {
Assert.isInstanceOf(String.class, lockKey);
String path = pathFor((String) lockKey);
JdbcLock jdbcLock = this.locks.get(path);
if (jdbcLock == null) {
throw new IllegalStateException("Could not found mutex at " + path);
}
if (!jdbcLock.renew()) {
throw new IllegalStateException("Could not renew mutex at " + path);
}
}
private static final class JdbcLock implements Lock {
private final LockRepository mutex;
@@ -232,7 +247,7 @@ public class JdbcLockRegistry implements ExpirableLockRegistry {
@Override
public void unlock() {
if (!this.delegate.isHeldByCurrentThread()) {
throw new IllegalMonitorStateException("You do not own mutex at " + this.path);
throw new IllegalMonitorStateException("The current thread doesn't own mutex at " + this.path);
}
if (this.delegate.getHoldCount() > 1) {
this.delegate.unlock();
@@ -243,7 +258,7 @@ public class JdbcLockRegistry implements ExpirableLockRegistry {
this.mutex.delete(this.path);
return;
}
catch (TransientDataAccessException e) {
catch (TransientDataAccessException | TransactionTimedOutException e) {
// try again
}
catch (Exception e) {
@@ -264,6 +279,27 @@ public class JdbcLockRegistry implements ExpirableLockRegistry {
return this.mutex.isAcquired(this.path);
}
public boolean renew() {
if (!this.delegate.isHeldByCurrentThread()) {
throw new IllegalMonitorStateException("The current thread doesn't own mutex at " + this.path);
}
while (true) {
try {
boolean renewed = this.mutex.renew(this.path);
if (renewed) {
this.lastUsed = System.currentTimeMillis();
}
return renewed;
}
catch (TransientDataAccessException | TransactionTimedOutException e) {
// try again
}
catch (Exception e) {
throw new DataAccessResourceFailureException("Failed to renew mutex at " + this.path, e);
}
}
}
}
}

6
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/LockRepository.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2019 the original author or authors.
* Copyright 2016-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.
@@ -24,6 +24,8 @@ import java.io.Closeable;
* has to be declared as a bean.
*
* @author Dave Syer
* @author Alexandre Strubel
*
* @since 4.3
*/
public interface LockRepository extends Closeable {
@@ -34,6 +36,8 @@ public interface LockRepository extends Closeable {
boolean acquire(String lock);
boolean renew(String lock);
@Override
void close();

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2019 the original author or authors.
* Copyright 2016-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.
@@ -34,29 +34,27 @@ import javax.sql.DataSource;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import org.springframework.util.StopWatch;
/**
* @author Dave Syer
* @author Artem Bilan
* @author Glenn Renfro
* @author Alexandre Strubel
*
* @since 4.3
*/
@ContextConfiguration("JdbcLockRegistryTests-context.xml")
@RunWith(SpringJUnit4ClassRunner.class)
@SpringJUnitConfig(locations = "JdbcLockRegistryTests-context.xml")
@DirtiesContext
public class JdbcLockRegistryDifferentClientTests {
@@ -76,7 +74,7 @@ public class JdbcLockRegistryDifferentClientTests {
@Autowired
private DataSource dataSource;
@Before
@BeforeEach
public void clear() {
this.registry.expireUnusedOlderThan(0);
this.client.close();
@@ -86,7 +84,7 @@ public class JdbcLockRegistryDifferentClientTests {
this.child.refresh();
}
@After
@AfterEach
public void close() {
if (this.child != null) {
this.child.close();
@@ -276,4 +274,111 @@ public class JdbcLockRegistryDifferentClientTests {
}
}
@Test
public void testOutOfDateLockTaken() throws Exception {
DefaultLockRepository client1 = new DefaultLockRepository(dataSource);
client1.setTimeToLive(500);
client1.afterPropertiesSet();
final DefaultLockRepository client2 = new DefaultLockRepository(dataSource);
client2.afterPropertiesSet();
Lock lock1 = new JdbcLockRegistry(client1).obtain("foo");
final BlockingQueue<Integer> data = new LinkedBlockingQueue<>();
final CountDownLatch latch1 = new CountDownLatch(1);
final CountDownLatch latch2 = new CountDownLatch(1);
lock1.lockInterruptibly();
Thread.sleep(500);
new SimpleAsyncTaskExecutor()
.execute(() -> {
Lock lock2 = new JdbcLockRegistry(client2).obtain("foo");
try {
latch1.countDown();
StopWatch stopWatch = new StopWatch();
stopWatch.start();
lock2.lockInterruptibly();
stopWatch.stop();
data.add(1);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
finally {
lock2.unlock();
}
latch2.countDown();
});
assertThat(latch1.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(latch2.await(10, TimeUnit.SECONDS)).isTrue();
data.add(2);
lock1.unlock();
for (int i = 0; i < 2; i++) {
Integer integer = data.poll(10, TimeUnit.SECONDS);
assertThat(integer).isNotNull();
assertThat(integer.intValue()).isEqualTo(i + 1);
}
}
@Test
public void testRenewLock() throws Exception {
DefaultLockRepository client1 = new DefaultLockRepository(dataSource);
client1.setTimeToLive(500);
client1.afterPropertiesSet();
final DefaultLockRepository client2 = new DefaultLockRepository(dataSource);
client2.afterPropertiesSet();
JdbcLockRegistry registry = new JdbcLockRegistry(client1);
Lock lock1 = registry.obtain("foo");
final BlockingQueue<Integer> data = new LinkedBlockingQueue<>();
final CountDownLatch latch1 = new CountDownLatch(2);
final CountDownLatch latch2 = new CountDownLatch(1);
lock1.lockInterruptibly();
new SimpleAsyncTaskExecutor()
.execute(() -> {
Lock lock2 = new JdbcLockRegistry(client2).obtain("foo");
try {
latch1.countDown();
StopWatch stopWatch = new StopWatch();
stopWatch.start();
lock2.lockInterruptibly();
stopWatch.stop();
data.add(4);
Thread.sleep(10);
data.add(5);
Thread.sleep(10);
data.add(6);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
finally {
lock2.unlock();
}
});
new SimpleAsyncTaskExecutor()
.execute(() -> {
try {
latch1.countDown();
Thread.sleep(1000);
data.add(1);
Thread.sleep(100);
data.add(2);
Thread.sleep(100);
data.add(3);
latch2.countDown();
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
assertThat(latch1.await(10, TimeUnit.SECONDS)).isTrue();
while (latch2.getCount() > 0) {
Thread.sleep(100);
registry.renewLock("foo");
}
lock1.unlock();
for (int i = 0; i < 6; i++) {
Integer integer = data.poll(10, TimeUnit.SECONDS);
assertThat(integer).isNotNull();
assertThat(integer.intValue()).isEqualTo(i + 1);
}
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2019 the original author or authors.
* Copyright 2016-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.
@@ -17,6 +17,7 @@
package org.springframework.integration.jdbc.lock;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
@@ -25,26 +26,24 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Lock;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.task.AsyncTaskExecutor;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
/**
* @author Dave Syer
* @author Artem Bilan
* @author Stefan Vassilev
*
* @since 4.3
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@SpringJUnitConfig
@DirtiesContext
public class JdbcLockRegistryTests {
@@ -56,7 +55,7 @@ public class JdbcLockRegistryTests {
@Autowired
private LockRepository client;
@Before
@BeforeEach
public void clear() {
this.registry.expireUnusedOlderThan(0);
this.client.close();
@@ -95,7 +94,7 @@ public class JdbcLockRegistryTests {
}
@Test
public void testReentrantLock() throws Exception {
public void testReentrantLock() {
for (int i = 0; i < 10; i++) {
Lock lock1 = this.registry.obtain("foo");
lock1.lock();
@@ -168,7 +167,7 @@ public class JdbcLockRegistryTests {
lock1.unlock();
Object ise = result.get(10, TimeUnit.SECONDS);
assertThat(ise).isInstanceOf(IllegalMonitorStateException.class);
assertThat(((Exception) ise).getMessage()).contains("You do not own");
assertThat(((Exception) ise).getMessage()).contains("own");
}
@Test
@@ -263,7 +262,28 @@ public class JdbcLockRegistryTests {
lock.unlock();
Object imse = result.get(10, TimeUnit.SECONDS);
assertThat(imse).isInstanceOf(IllegalMonitorStateException.class);
assertThat(((Exception) imse).getMessage()).contains("You do not own");
assertThat(((Exception) imse).getMessage()).contains("own");
}
@Test
public void testLockRenew() {
final Lock lock = this.registry.obtain("foo");
assertThat(lock.tryLock()).isTrue();
try {
registry.renewLock("foo");
}
finally {
lock.unlock();
}
}
@Test
public void testLockRenewLockNotOwned() {
this.registry.obtain("foo");
assertThatExceptionOfType(IllegalMonitorStateException.class)
.isThrownBy(() -> registry.renewLock("foo"));
}
}

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

@@ -1085,6 +1085,12 @@ If so, you can specify the `id` to be associated with the `DefaultLockRepository
Starting with version 5.1.8, the `JdbcLockRegistry` can be configured with the `idleBetweenTries` - a `Duration` to sleep between lock record insert/update executions.
By default it is `100` milliseconds and in some environments non-leaders pollute connections with data source too often.
Starting with version 5.4, the `RenewableLockRegistry` interface has been introduced and added to `JdbcLockRegistry`.
The `renewLock()` method must be called during locked process in case of the locked process would be longer than time to live of the lock.
So the time to live can be highly reduce and deployments can retake a lost lock quickly.
NOTE: The lock renewal can be done only if the lock is held by the current thread.
[[jdbc-metadata-store]]
=== JDBC Metadata Store

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

@@ -30,6 +30,11 @@ See <<./r2dbc.adoc#r2dbc,R2DBC Support>> for more information.
The Channel Adapters for Redis Stream support have been introduced.
See <<./redis.adoc#redis-stream-outbound,Redis Stream Outbound Channel Adapter>> for more information.
==== Renewable Lock Registry
A Renewable lock registry has been introduced to allow renew lease of a distributed lock.
See <<./jdbc.adoc#jdbc-lock-registry,JDBC implementation>> for more information.
[[x5.4-general]]
=== General Changes