GH-3272: Add lease renewal for distributed locks

Fixes #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

Author: Alexandre Strubel

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);
+
+}

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 @@
  * @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 DefaultLockRepository(DataSource dataSource, String id) {
 	/**
 	 * 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 @@ private void deleteExpired(String lock) {
 				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;
+	}
+
 }

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

@@ -30,6 +30,7 @@
 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 @@
  * @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 void expireUnusedOlderThan(long age) {
 		}
 	}
 
+	@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 @@ private boolean doLock() {
 		@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 void unlock() {
 					this.mutex.delete(this.path);
 					return;
 				}
-				catch (TransientDataAccessException e) {
+				catch (TransientDataAccessException | TransactionTimedOutException e) {
 					// try again
 				}
 				catch (Exception e) {
@@ -264,6 +279,27 @@ public boolean isAcquiredInThisProcess() {
 			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);
+				}
+			}
+		}
+
 	}
 
 }

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 @@
  * 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();
 

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 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 void clear() {
 		this.child.refresh();
 	}
 
-	@After
+	@AfterEach
 	public void close() {
 		if (this.child != null) {
 			this.child.close();
@@ -276,4 +274,111 @@ public void testExclusiveAccess() throws Exception {
 		}
 	}
 
+	@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);
+		}
+	}
+
 }