GH-3272: Support lease renewal for distributed locks

Author: Alexandre Strubel

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
  */
@@ -179,4 +180,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

@@ -49,10 +49,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 +102,20 @@ public void expireUnusedOlderThan(long age) {
 		}
 	}
 
+	@Override
+	public void renewLock(Lock lock) {
+		Assert.isInstanceOf(JdbcLock.class, lock);
+		JdbcLock jdbcLock = (JdbcLock) lock;
+
+		if (!jdbcLock.isAcquiredInThisProcess()) {
+			throw new IllegalMonitorStateException("You do not own mutex at " + jdbcLock.path);
+		}
+
+		if (!jdbcLock.renew()) {
+			throw new IllegalStateException("Could not renew mutex at " + jdbcLock.path);
+		}
+	}
+
 	private static final class JdbcLock implements Lock {
 
 		private final LockRepository mutex;
@@ -264,6 +279,21 @@ public boolean isAcquiredInThisProcess() {
 			return this.mutex.isAcquired(this.path);
 		}
 
+		public boolean renew() {
+			if (!this.delegate.isHeldByCurrentThread()) {
+				throw new IllegalMonitorStateException("You do not own mutex at " + this.path);
+			}
+			while (true) {
+				try {
+					return this.mutex.renew(this.path);
+				} catch (TransientDataAccessException 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/main/java/org/springframework/integration/jdbc/lock/RenewableLockRegistry.java

@@ -0,0 +1,38 @@
+/*
+ * 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.jdbc.lock;
+
+import org.springframework.integration.support.locks.LockRegistry;
+
+import java.util.concurrent.locks.Lock;
+
+/**
+ * A {@link LockRegistry} implementing this interface supports the renewal of the time to live of a lock
+ *
+ * @author Alexandre Strubel
+ *
+ * @since 4.3
+ */
+public interface RenewableLockRegistry extends LockRegistry {
+
+	/**
+	 * Renew the time to live of a 'lock' held by the current thread
+	 * @param lock The lock to renew its time to live
+	 */
+	void renewLock(Lock lock);
+
+}

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.
@@ -52,6 +52,7 @@
  * @author Dave Syer
  * @author Artem Bilan
  * @author Glenn Renfro
+ * @author Alexandre Strubel
  *
  * @since 4.3
  */
@@ -276,4 +277,110 @@ 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(lock1);
+		}
+		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);
+		}
+	}
 }

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.
@@ -27,6 +27,7 @@
 
 import org.junit.Before;
 import org.junit.Test;
+import org.junit.jupiter.api.Assertions;
 import org.junit.runner.RunWith;
 
 import org.springframework.beans.factory.annotation.Autowired;
@@ -40,6 +41,7 @@
 /**
  * @author Dave Syer
  * @author Artem Bilan
+ * @author Stefan Vassilev
  *
  * @since 4.3
  */
@@ -266,4 +268,18 @@ public void testTwoThreadsWrongOneUnlocks() throws Exception {
 		assertThat(((Exception) imse).getMessage()).contains("You do not own");
 	}
 
+	@Test
+	public void testLockRenew() {
+		final Lock lock = this.registry.obtain("foo");
+
+		assertThat(lock.tryLock()).isTrue();
+		registry.renewLock(lock);
+	}
+
+	@Test
+	public void testLockRenewLockNotOwned() {
+		final Lock lock = this.registry.obtain("foo");
+
+		Assertions.assertThrows(IllegalMonitorStateException.class, () -> registry.renewLock(lock));
+	}
 }