diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/locks/RenewableLockRegistry.java b/spring-integration-core/src/main/java/org/springframework/integration/support/locks/RenewableLockRegistry.java new file mode 100644 index 0000000000..969f3c3ade --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/locks/RenewableLockRegistry.java @@ -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); + +} diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/DefaultLockRepository.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/DefaultLockRepository.java index 60a8be5139..0b8654b6bf 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/DefaultLockRepository.java +++ b/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-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 DEFAULT. + * 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; + } + } diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/JdbcLockRegistry.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/JdbcLockRegistry.java index 658ed302e5..9bede5f10c 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/JdbcLockRegistry.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/JdbcLockRegistry.java @@ -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); + } + } + } + } } diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/LockRepository.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/LockRepository.java index cbe2aed50a..efe26d09f8 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/LockRepository.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/LockRepository.java @@ -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(); diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/lock/JdbcLockRegistryDifferentClientTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/lock/JdbcLockRegistryDifferentClientTests.java index e37b7e7890..c5c5aa1a6a 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/lock/JdbcLockRegistryDifferentClientTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/lock/JdbcLockRegistryDifferentClientTests.java @@ -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 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 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); + } + } + } diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/lock/JdbcLockRegistryTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/lock/JdbcLockRegistryTests.java index 44d1fefd38..bb40297b79 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/lock/JdbcLockRegistryTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/lock/JdbcLockRegistryTests.java @@ -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")); } } diff --git a/src/reference/asciidoc/jdbc.adoc b/src/reference/asciidoc/jdbc.adoc index 30390a7563..332fcd0d7d 100644 --- a/src/reference/asciidoc/jdbc.adoc +++ b/src/reference/asciidoc/jdbc.adoc @@ -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 diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index be4ec644be..82e01e8f33 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -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