Add Redis lock periodic renewal

Although `RenewableLockRegistry` provides a renew interface, it is inconvenient for users.
Developers hope to have a lock that can be automatically renewed.
On the one hand, it can avoid subsequent failures caused by locks that will not expire when abnormal exits,
and on the other hand, it can avoid unlock failures caused by lock expired.

* Add `RenewableLockRegistry.setRenewalTaskScheduler()` and when it is set, schedule a `renew()` script periodically
when lock is acquired  from Redis with `1/3` of `expireAfter`
* Test and document the feature

spring-integration-core/src/main/java/org/springframework/integration/support/locks/RenewableLockRegistry.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020 the original author or authors.
+ * Copyright 2020-2024 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.
@@ -16,12 +16,15 @@
 
 package org.springframework.integration.support.locks;
 
+import org.springframework.scheduling.TaskScheduler;
+
 /**
  * A {@link LockRegistry} implementing this interface supports the renewal
  * of the time to live of a lock.
  *
  * @author Alexandre Strubel
  * @author Artem Bilan
+ * @author Youbin Wu
  *
  * @since 5.4
  */
@@ -34,4 +37,14 @@ public interface RenewableLockRegistry extends LockRegistry {
 	 */
 	void renewLock(Object lockKey);
 
+	/**
+	 * Set the {@link TaskScheduler} to use for the renewal task.
+	 * When renewalTaskScheduler is set, it will be used to periodically renew the lock to ensure that
+	 * the lock does not expire while the thread is working.
+	 * @param renewalTaskScheduler renew task scheduler
+	 * @since 6.4
+	 */
+	default void setRenewalTaskScheduler(TaskScheduler renewalTaskScheduler) {
+	}
+
 }

spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java

@@ -16,6 +16,7 @@
 
 package org.springframework.integration.redis.util;
 
+import java.io.Serial;
 import java.text.SimpleDateFormat;
 import java.time.Duration;
 import java.util.Collections;
@@ -32,6 +33,7 @@ import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.locks.Condition;
@@ -54,6 +56,8 @@ import org.springframework.data.redis.listener.ChannelTopic;
 import org.springframework.data.redis.listener.RedisMessageListenerContainer;
 import org.springframework.data.redis.listener.Topic;
 import org.springframework.integration.support.locks.ExpirableLockRegistry;
+import org.springframework.integration.support.locks.RenewableLockRegistry;
+import org.springframework.scheduling.TaskScheduler;
 import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
 import org.springframework.util.Assert;
 import org.springframework.util.ReflectionUtils;
@@ -89,11 +93,12 @@ import org.springframework.util.ReflectionUtils;
  * @author Myeonghyeon Lee
  * @author Roman Zabaluev
  * @author Alex Peelman
+ * @author Youbin Wu
  *
  * @since 4.0
  *
  */
-public final class RedisLockRegistry implements ExpirableLockRegistry, DisposableBean {
+public final class RedisLockRegistry implements ExpirableLockRegistry, DisposableBean, RenewableLockRegistry {
 
 	private static final Log LOGGER = LogFactory.getLog(RedisLockRegistry.class);
 
@@ -110,6 +115,9 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl
 	private final Map<String, RedisLock> locks =
 			new LinkedHashMap<>(16, 0.75F, true) {
 
+				@Serial
+				private static final long serialVersionUID = 7419938441348450459L;
+
 				@Override
 				protected boolean removeEldestEntry(Entry<String, RedisLock> eldest) {
 					return size() > RedisLockRegistry.this.cacheCapacity;
@@ -138,6 +146,8 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl
 	private Executor executor =
 			Executors.newCachedThreadPool(new CustomizableThreadFactory("redis-lock-registry-"));
 
+	private TaskScheduler renewalTaskScheduler;
+
 	/**
 	 * Flag to denote whether the {@link ExecutorService} was provided via the setter and
 	 * thus should not be shutdown when {@link #destroy()} is called
@@ -207,6 +217,12 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl
 		this.executorExplicitlySet = true;
 	}
 
+	@Override
+	public void setRenewalTaskScheduler(TaskScheduler renewalTaskScheduler) {
+		Assert.notNull(renewalTaskScheduler, "'renewalTaskScheduler' must not be null");
+		this.renewalTaskScheduler = renewalTaskScheduler;
+	}
+
 	/**
 	 * Set the capacity of cached locks.
 	 * @param cacheCapacity The capacity of cached lock, (default 100_000).
@@ -291,6 +307,26 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl
 		}
 	}
 
+	@Override
+	public void renewLock(Object lockKey) {
+		String path = (String) lockKey;
+		RedisLock redisLock;
+		this.lock.lock();
+		try {
+			redisLock = this.locks.computeIfAbsent(path, getRedisLockConstructor(this.redisLockType));
+		}
+		finally {
+			this.lock.unlock();
+		}
+		if (redisLock == null) {
+			throw new IllegalStateException("Could not renew mutex at " + path);
+		}
+
+		if (!redisLock.renew()) {
+			throw new IllegalStateException("Could not renew mutex at " + path);
+		}
+	}
+
 	/**
 	 * The mode in which this registry is going to work with locks.
 	 */
@@ -328,8 +364,19 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl
 				return false
 				""";
 
-		protected static final RedisScript<Boolean>
-				OBTAIN_LOCK_REDIS_SCRIPT = new DefaultRedisScript<>(OBTAIN_LOCK_SCRIPT, Boolean.class);
+		private static final String RENEW_SCRIPT = """
+				if (redis.call('GET', KEYS[1]) == ARGV[1]) then
+					redis.call('PEXPIRE', KEYS[1], ARGV[2])
+					return true
+				end
+				return false
+				""";
+
+		protected static final RedisScript<Boolean> OBTAIN_LOCK_REDIS_SCRIPT =
+				new DefaultRedisScript<>(OBTAIN_LOCK_SCRIPT, Boolean.class);
+
+		public static final RedisScript<Boolean> RENEW_REDIS_SCRIPT =
+				new DefaultRedisScript<>(RENEW_SCRIPT, Boolean.class);
 
 		protected final String lockKey;
 
@@ -337,6 +384,8 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl
 
 		private volatile long lockedAt;
 
+		private volatile ScheduledFuture<?> renewFuture;
+
 		private RedisLock(String path) {
 			this.lockKey = constructLockKey(path);
 		}
@@ -454,6 +503,11 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl
 					LOGGER.debug("Acquired lock; " + this);
 				}
 				this.lockedAt = System.currentTimeMillis();
+				if (RedisLockRegistry.this.renewalTaskScheduler != null) {
+					Duration delay = Duration.ofMillis(RedisLockRegistry.this.expireAfter / 3);
+					this.renewFuture =
+							RedisLockRegistry.this.renewalTaskScheduler.scheduleWithFixedDelay(this::renew, delay);
+				}
 			}
 			return acquired;
 		}
@@ -515,6 +569,7 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl
 
 				if (Boolean.TRUE.equals(unlinkResult)) {
 					// Lock key successfully unlinked
+					stopRenew();
 					return;
 				}
 				else if (Boolean.FALSE.equals(unlinkResult)) {
@@ -526,6 +581,26 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl
 				throw new ConcurrentModificationException("Lock was released in the store due to expiration. " +
 						"The integrity of data protected by this lock may have been compromised.");
 			}
+			else {
+				stopRenew();
+			}
+		}
+
+		protected final boolean renew() {
+			boolean res = Boolean.TRUE.equals(RedisLockRegistry.this.redisTemplate.execute(
+					RENEW_REDIS_SCRIPT, Collections.singletonList(this.lockKey),
+					RedisLockRegistry.this.clientId, String.valueOf(RedisLockRegistry.this.expireAfter)));
+			if (!res) {
+				stopRenew();
+			}
+			return res;
+		}
+
+		protected final void stopRenew() {
+			if (this.renewFuture != null) {
+				this.renewFuture.cancel(true);
+				this.renewFuture = null;
+			}
 		}
 
 		@Override
@@ -553,7 +628,7 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl
 			int result = 1;
 			result = prime * result + getOuterType().hashCode();
 			result = prime * result + ((this.lockKey == null) ? 0 : this.lockKey.hashCode());
-			result = prime * result + (int) (this.lockedAt ^ (this.lockedAt >>> 32)); // NOSONAR magic number
+			result = prime * result + Long.hashCode(this.lockedAt);
 			result = prime * result + RedisLockRegistry.this.clientId.hashCode();
 			return result;
 		}

spring-integration-redis/src/test/java/org/springframework/integration/redis/util/RedisLockRegistryTests.java

@@ -51,8 +51,10 @@ import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.integration.redis.RedisContainerTest;
 import org.springframework.integration.redis.util.RedisLockRegistry.RedisLockType;
 import org.springframework.integration.test.util.TestUtils;
+import org.springframework.scheduling.concurrent.SimpleAsyncTaskScheduler;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 import static org.assertj.core.api.Assertions.assertThatNoException;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.mockito.Mockito.mock;
@@ -66,6 +68,7 @@ import static org.mockito.Mockito.mock;
  * @author Artem Vozhdayenko
  * @author Anton Gabov
  * @author Eddie Cho
+ * @author Youbin Wu
  *
  * @since 4.0
  *
@@ -427,6 +430,20 @@ class RedisLockRegistryTests implements RedisContainerTest {
 		registry.destroy();
 	}
 
+	@ParameterizedTest
+	@EnumSource(RedisLockType.class)
+	void testRenewalOnExpire(RedisLockType redisLockType) throws Exception {
+		long expireAfter = 300L;
+		RedisLockRegistry registry = new RedisLockRegistry(redisConnectionFactory, this.registryKey, expireAfter);
+		registry.setRenewalTaskScheduler(new SimpleAsyncTaskScheduler());
+		registry.setRedisLockType(redisLockType);
+		Lock lock1 = registry.obtain("foo");
+		assertThat(lock1.tryLock()).isTrue();
+		Thread.sleep(expireAfter * 2);
+		lock1.unlock();
+		registry.destroy();
+	}
+
 	@ParameterizedTest
 	@EnumSource(RedisLockType.class)
 	void testEquals(RedisLockType testRedisLockType) {
@@ -900,6 +917,33 @@ class RedisLockRegistryTests implements RedisContainerTest {
 		registry.destroy();
 	}
 
+	@ParameterizedTest
+	@EnumSource(RedisLockType.class)
+	void testLockRenew(RedisLockType redisLockType) {
+		final RedisLockRegistry registry = new RedisLockRegistry(redisConnectionFactory, this.registryKey);
+		registry.setRedisLockType(redisLockType);
+		final Lock lock = registry.obtain("foo");
+
+		assertThat(lock.tryLock()).isTrue();
+		try {
+			registry.renewLock("foo");
+		}
+		finally {
+			lock.unlock();
+		}
+	}
+
+	@ParameterizedTest
+	@EnumSource(RedisLockType.class)
+	void testLockRenewLockNotOwned(RedisLockType redisLockType) {
+		final RedisLockRegistry registry = new RedisLockRegistry(redisConnectionFactory, this.registryKey);
+		registry.setRedisLockType(redisLockType);
+		registry.obtain("foo");
+
+		assertThatExceptionOfType(IllegalStateException.class)
+				.isThrownBy(() -> registry.renewLock("foo"));
+	}
+
 	@Test
 	void testInitialiseWithCustomExecutor() {
 		RedisLockRegistry redisLockRegistry = new RedisLockRegistry(redisConnectionFactory, "registryKey");

src/reference/antora/modules/ROOT/pages/redis.adoc

@@ -856,4 +856,7 @@ Default.
 The pub-sub is preferred mode - less network chatter between client Redis server, and more performant - the lock is acquired immediately when subscription is notified about unlocking in the other process.
 However, the Redis does not support pub-sub in the Master/Replica connections (for example in AWS ElastiCache environment), therefore a busy-spin mode is chosen as a default to make the registry working in any environment.
 
-Starting with version 6.4, instead of throwing `IllegalStateException`, the `RedisLockRegistry.RedisLock.unlock()` method throws `ConcurrentModificationException` if the ownership of the lock is expired.
+Starting with version 6.4, instead of throwing `IllegalStateException`, the `RedisLockRegistry.RedisLock.unlock()` method throws `ConcurrentModificationException` if the ownership of the lock is expired.
+
+Starting with version 6.4, a `RedisLockRegistry.setRenewalTaskScheduler()` is added to configure the scheduler for periodic renewal of locks.
+When it is set, the lock will be automatically renewed every `1/3` of the expiration time after the lock is successfully acquired, until unlocked or the redis key is removed.

src/reference/antora/modules/ROOT/pages/whats-new.adoc

@@ -59,6 +59,7 @@ See xref:zeromq.adoc[ZeroMQ Support] for more information.
 === Redis Changes
 
 Instead of throwing `IllegalStateException`, the `RedisLockRegistry.RedisLock.unlock()` method throws `ConcurrentModificationException` if the ownership of the lock is expired.
+Add a `RedisLockRegistry.setRenewalTaskScheduler()` to periodic lock renewal.
 See xref:redis.adoc[Redis Support] for more information.
 
 [[x6.4-groovy-changes]]