GH-11180: Fix caching logic in the RedisLockRegistry (#11181)

Fixes: #11180

The previous solution, based on the shared pool of `ReentrantLock`
from the `DefaultLockRegistry`, is proven to be inconvenient with keys
collision: different keys might stumble on the same shared
`ReentrantLock` making the application being blocked for nothing.

* Remove `DefaultLockRegistry` logic from the `RedisLockRegistry`
* Remove local `RedisLock.holdCount` as we rely on the `localLock` again
* Revert back default `cacheCapacity` to `100_000` as we don't use a shared pool of locks anymore
* Do not evict from the cache still in-held locks
* In the `obtain()`, try to remove unused lock when `cacheCapacity` is reached.
If it cannot clean up the room, throw a `CannotAcquireLockException`.
This is to prevent an out-of-memory error
* Also, throw a `CannotAcquireLockException` on lock calls for an orphaned lock.
This is to prevent a race condition when two distinct local locks are locked in different threads against the same key
* Add human-readable `toString()` into the `RedisLockRegistry`

**Auto-cherry-pick to `7.0.x`**

Author: artembilan

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

@@ -59,7 +59,6 @@
 import org.springframework.data.redis.listener.PatternTopic;
 import org.springframework.data.redis.listener.RedisMessageListenerContainer;
 import org.springframework.data.redis.listener.Topic;
-import org.springframework.integration.support.locks.DefaultLockRegistry;
 import org.springframework.integration.support.locks.DistributedLock;
 import org.springframework.integration.support.locks.ExpirableLockRegistry;
 import org.springframework.integration.support.locks.RenewableLockRegistry;
@@ -79,14 +78,21 @@
  * Locks are reentrant.
  * <p>
  * <b>However, locks are scoped by the registry; a lock from a different registry with the
- * same key (even if the registry uses the same 'registryKey') are different
+ * same key (even if the registry uses the same 'registryKey') is different
  * locks, and the second cannot be acquired by the same thread while the first is
  * locked.</b>
  * <p>
- * <b>Note: This is not intended for low latency applications.</b> It is intended
+ * <b>Note: This is not intended for low-latency applications.</b> It is intended
  * for resource locking across multiple JVMs.
  * <p>
  * {@link Condition}s are not supported.
+ * <p>
+ * The lock instances are cached by their key for future reuse.
+ * If all the locks in the cache are acquired,
+ * the {@link #obtain(Object)} method throws {@link CannotAcquireLockException}.
+ * Otherwise, the old, unused locks are evicted from the cache.
+ * An attempt to lock such an orphaned {@link RedisLock} throws another {@link CannotAcquireLockException}
+ * to avoid double locking from different threads on the same key.
  *
  * @author Gary Russell
  * @author Konstantin Yakimov
@@ -114,16 +120,14 @@ public final class RedisLockRegistry
 
 	private static final long DEFAULT_EXPIRE_AFTER = 60000L;
 
-	private static final int DEFAULT_CAPACITY = 256;
+	private static final int DEFAULT_CAPACITY = 100_000;
 
 	private static final int DEFAULT_IDLE = 100;
 
 	private final Lock lock = new ReentrantLock();
 
 	private Duration idleBetweenTries = Duration.ofMillis(DEFAULT_IDLE);
 
-	private DefaultLockRegistry defaultLockRegistry = new DefaultLockRegistry();
-
 	private final Map<String, RedisLock> locks =
 			new LinkedHashMap<>(16, 0.75F, true) {
 
@@ -132,7 +136,8 @@ public final class RedisLockRegistry
 
 				@Override
 				protected boolean removeEldestEntry(Entry<String, RedisLock> eldest) {
-					return size() > RedisLockRegistry.this.cacheCapacity;
+					return size() > RedisLockRegistry.this.cacheCapacity
+							&& !eldest.getValue().isAcquiredInThisProcess();
 				}
 
 			};
@@ -249,14 +254,18 @@ public void setRenewalTaskScheduler(TaskScheduler renewalTaskScheduler) {
 
 	/**
 	 * Set the capacity of cached locks.
-	 * @param cacheCapacity The capacity of cached lock, (default 256 locks).
+	 * If the number of currently cached locks exceeds the new capacity,
+	 * this method tries to evict unused locks.
+	 * @param cacheCapacity The capacity of cached lock, (default 100_000).
 	 * @since 5.5.6
+	 * @see #expireUnusedOlderThan(long)
 	 */
 	public void setCacheCapacity(int cacheCapacity) {
+		Assert.isTrue(cacheCapacity > 0, "'cacheCapacity' must be greater than 0");
+		if (this.locks.size() > cacheCapacity) {
+			expireUnusedOlderThan(0);
+		}
 		this.cacheCapacity = cacheCapacity;
-		// Find the highest power of 2 for (n + 1), then subtract 1 to get the mask
-		int mask = Integer.highestOneBit(cacheCapacity + 1) - 1;
-		this.defaultLockRegistry = new DefaultLockRegistry(mask);
 	}
 
 	/**
@@ -290,7 +299,22 @@ public DistributedLock obtain(Object lockKey) {
 		String path = (String) lockKey;
 		this.lock.lock();
 		try {
-			return this.locks.computeIfAbsent(path, getRedisLockConstructor(this.redisLockType));
+			RedisLock redisLock = this.locks.computeIfAbsent(path, getRedisLockConstructor(this.redisLockType));
+			if (!redisLock.isAcquiredInThisProcess() && this.locks.size() > this.cacheCapacity) {
+				this.locks.entrySet()
+						.removeIf(entry -> {
+							RedisLock lock = entry.getValue();
+							return lock != redisLock && !lock.isAcquiredInThisProcess();
+						});
+
+				if (this.locks.size() > this.cacheCapacity) {
+					this.locks.remove(path);
+					throw new CannotAcquireLockException("There are already " +
+							this.cacheCapacity + " unique Redis locks acquired in the application " +
+							"from " + this + ". Cannot obtain more at the moment.");
+				}
+			}
+			return redisLock;
 		}
 		finally {
 			this.lock.unlock();
@@ -305,11 +329,7 @@ public void expireUnusedOlderThan(long age) {
 			this.locks.entrySet()
 					.removeIf(entry -> {
 						RedisLock lock = entry.getValue();
-						long lockedAt = lock.getLockedAt();
-						return now - lockedAt > age
-								// 'lockedAt = 0' means that the lock is still not acquired!
-								&& lockedAt > 0
-								&& !lock.isAcquiredInThisProcess();
+						return now - lock.getLockedAt() >= age && !lock.isAcquiredInThisProcess();
 					});
 		}
 		finally {
@@ -359,6 +379,13 @@ public void renewLock(Object lockKey, Duration ttl) {
 		}
 	}
 
+	@Override
+	public String toString() {
+		return "RedisLockRegistry{" +
+				"registryKey=" + this.registryKey +
+				'}';
+	}
+
 	/**
 	 * The mode in which this registry is going to work with locks.
 	 */
@@ -410,19 +437,19 @@ private abstract class RedisLock implements DistributedLock {
 		public static final RedisScript<Boolean> RENEW_REDIS_SCRIPT =
 				new DefaultRedisScript<>(RENEW_SCRIPT, Boolean.class);
 
-		protected final String lockKey;
+		private final String key;
 
-		private final ReentrantLock localLock;
+		protected final String lockKey;
 
-		private int holdCount;
+		private final ReentrantLock localLock = new ReentrantLock();
 
-		private volatile long lockedAt;
+		private volatile long lockedAt = System.currentTimeMillis();
 
 		private volatile @Nullable ScheduledFuture<?> renewFuture;
 
 		private RedisLock(String path) {
+			this.key = path;
 			this.lockKey = constructLockKey(path);
-			this.localLock = (ReentrantLock) RedisLockRegistry.this.defaultLockRegistry.obtain(this.lockKey);
 		}
 
 		private String constructLockKey(String path) {
@@ -460,7 +487,6 @@ public void lock(Duration ttl) {
 			while (true) {
 				try {
 					if (tryRedisLock(-1L, ttl.toMillis())) {
-						this.holdCount++;
 						return;
 					}
 				}
@@ -488,7 +514,6 @@ public final void lockInterruptibly() throws InterruptedException {
 			while (true) {
 				try {
 					if (tryRedisLock(-1L, RedisLockRegistry.this.expireAfter.toMillis())) {
-						this.holdCount++;
 						return;
 					}
 				}
@@ -530,9 +555,6 @@ public boolean tryLock(Duration waitTime, Duration ttl) throws InterruptedExcept
 				if (!acquired) {
 					this.localLock.unlock();
 				}
-				else {
-					this.holdCount++;
-				}
 				return acquired;
 			}
 			catch (Exception e) {
@@ -543,6 +565,10 @@ public boolean tryLock(Duration waitTime, Duration ttl) throws InterruptedExcept
 		}
 
 		private boolean tryRedisLock(long time, long expireAfter) throws ExecutionException, InterruptedException {
+			if (!RedisLockRegistry.this.locks.containsKey(this.key)) {
+				throw new IllegalStateException("The lock '" + this.key + "' was evicted from the exhausted cache. " +
+						"Obtain a fresh one from " + RedisLockRegistry.this);
+			}
 			final boolean acquired = tryRedisLockInner(time, expireAfter);
 			if (acquired) {
 				if (LOGGER.isDebugEnabled()) {
@@ -571,8 +597,7 @@ public final void unlock() {
 			if (!this.localLock.isHeldByCurrentThread()) {
 				throw new IllegalStateException("You do not own lock at " + this.lockKey);
 			}
-			if (this.holdCount > 1) {
-				this.holdCount--;
+			if (this.localLock.getHoldCount() > 1) {
 				this.localLock.unlock();
 				return;
 			}
@@ -617,7 +642,6 @@ public final void unlock() {
 				ReflectionUtils.rethrowRuntimeException(e);
 			}
 			finally {
-				this.holdCount--;
 				this.localLock.unlock();
 			}
 		}
@@ -667,7 +691,7 @@ public String toString() {
 			SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd@HH:mm:ss.SSS");
 			return "RedisLock [lockKey=" + this.lockKey
 					+ ",lockedAt=" + dateFormat.format(new Date(this.lockedAt))
-					+ ", clientId=" + RedisLockRegistry.this.clientId
+					+ ", clientId=" + RedisLockRegistry.this
 					+ "]";
 		}
 
@@ -676,7 +700,7 @@ public int hashCode() {
 			final int prime = 31;
 			int result = 1;
 			result = prime * result + getOuterType().hashCode();
-			result = prime * result + ((this.lockKey == null) ? 0 : this.lockKey.hashCode());
+			result = prime * result + this.lockKey.hashCode();
 			result = prime * result + Long.hashCode(this.lockedAt);
 			result = prime * result + RedisLockRegistry.this.clientId.hashCode();
 			return result;
@@ -687,9 +711,6 @@ public boolean equals(Object obj) {
 			if (this == obj) {
 				return true;
 			}
-			if (obj == null) {
-				return false;
-			}
 			if (getClass() != obj.getClass()) {
 				return false;
 			}
@@ -763,7 +784,7 @@ private boolean subscribeLock(long time, long expireAfter) throws ExecutionExcep
 						return true;
 					}
 					try {
-						//if short expireAfter key expire for ttl, no receive unlock msg
+						//if short expireAfter key expire for ttl, no received unlock msg
 						long waitTime = time >= 0 ? time : RedisLockRegistry.this.expireAfter.toMillis();
 						future.get(waitTime, TimeUnit.MILLISECONDS);
 					}

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

@@ -47,6 +47,7 @@
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.EnumSource;
 
+import org.springframework.dao.CannotAcquireLockException;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.integration.redis.RedisContainerTest;
@@ -1019,7 +1020,7 @@ void testInitialiseWithCustomExecutor() {
 	}
 
 	@Test
-	void noSecondLockOnEviction() throws InterruptedException {
+	void notUsedLockIsEvictedOnCacheLimit() throws InterruptedException {
 		RedisLockRegistry registry = new RedisLockRegistry(redisConnectionFactory, this.registryKey);
 		registry.setRedisLockType(RedisLockType.SPIN_LOCK);
 		registry.setCacheCapacity(2);
@@ -1044,11 +1045,11 @@ void noSecondLockOnEviction() throws InterruptedException {
 				});
 
 		assertThat(furtherLocksLatch.await(10, TimeUnit.SECONDS)).isTrue();
-		// Two new locks to trigger cache eviction for the 'lock1'
+		// Two new locks to overcharge the cache.
+		// The 'lock2' is evicted by 'lock3' because it is not locked.
 		registry.obtain("lock2");
 		registry.obtain("lock3");
 
-		// Request 'lock1' again: will trigger new RedisLock instance
 		DistributedLock lock1 = registry.obtain("lock1");
 
 		// Cannot lock because 'lock1' is still locked by another thread
@@ -1062,6 +1063,52 @@ void noSecondLockOnEviction() throws InterruptedException {
 		});
 	}
 
+	@ParameterizedTest
+	@EnumSource(RedisLockType.class)
+	void cannotAcquireLockExceptionOnCacheLimit(RedisLockType redisLockType) {
+		RedisLockRegistry registry = new RedisLockRegistry(redisConnectionFactory, this.registryKey);
+		registry.setRedisLockType(redisLockType);
+		registry.setCacheCapacity(2);
+
+		DistributedLock lock1 = registry.obtain("lock1");
+		lock1.lock();
+		try {
+			DistributedLock lock2 = registry.obtain("lock2");
+			lock2.lock();
+			try {
+				assertThatExceptionOfType(CannotAcquireLockException.class)
+						.isThrownBy(() -> registry.obtain("lock3"))
+						.withMessageStartingWith("There are already 2 unique Redis locks acquired in the application")
+						.withMessageEndingWith("Cannot obtain more at the moment.");
+			}
+			finally {
+				lock2.unlock();
+			}
+		}
+		finally {
+			lock1.unlock();
+		}
+		registry.destroy();
+	}
+
+	@ParameterizedTest
+	@EnumSource(RedisLockType.class)
+	void unusedLockIsStale(RedisLockType redisLockType) {
+		RedisLockRegistry registry = new RedisLockRegistry(redisConnectionFactory, this.registryKey);
+		registry.setRedisLockType(redisLockType);
+		registry.setCacheCapacity(1);
+
+		DistributedLock lock1 = registry.obtain("lock1");
+		registry.obtain("lock2");
+
+		assertThatExceptionOfType(CannotAcquireLockException.class)
+				.isThrownBy(lock1::lock)
+				.havingCause()
+				.isInstanceOf(IllegalStateException.class)
+				.withMessageStartingWith("The lock 'lock1' was evicted from the exhausted cache.");
+		registry.destroy();
+	}
+
 	@Test
 	void testUnlockWhenInterruptedBlocksUntilKeyRemoved() throws Exception {
 		RedisLockRegistry registry = new RedisLockRegistry(redisConnectionFactory, this.registryKey);
@@ -1133,55 +1180,6 @@ private void waitForExpire(String key) throws Exception {
 		assertThat(n < 100).as(key + " key did not expire").isTrue();
 	}
 
-	@Test
-	void sharedDelegateIsDeletedFromRedisOnUnlock() throws Exception {
-		RedisLockRegistry registry1 = new RedisLockRegistry(redisConnectionFactory, this.registryKey);
-		// setCacheCapacity(2) gives mask=1 in DefaultLockRegistry → only 2 slots, guaranteed collision in ≤3 keys
-		registry1.setCacheCapacity(2);
-		RedisLockRegistry registry2 = new RedisLockRegistry(redisConnectionFactory, this.registryKey);
-
-		// Find two distinct lock keys that share the same ReentrantLock (same DefaultLockRegistry slot)
-		String keyB = null;
-		DistributedLock lockA = null;
-		DistributedLock lockB = null;
-		outer:
-		for (int i = 0; i < 10; i++) {
-			for (int j = i + 1; j < 10; j++) {
-				DistributedLock la = registry1.obtain("key-" + i);
-				DistributedLock lb = registry1.obtain("key-" + j);
-				if (TestUtils.getPropertyValue(la, "localLock") == TestUtils.getPropertyValue(lb, "localLock")) {
-					keyB = "key-" + j;
-					lockA = la;
-					lockB = lb;
-					break outer;
-				}
-			}
-		}
-		assertThat(lockA)
-				.as("Could not find two lock keys mapping to the same DefaultLockRegistry slot")
-				.isNotNull();
-
-		lockA.lock();
-		// Shared localLock: localLock.holdCount becomes 2 after this call
-		lockB.lock();
-
-		// Unlock B first — before the fix, localLock.getHoldCount()>1 short-circuited and skipped the Redis delete
-		lockB.unlock();
-
-		// Another process must be able to acquire keyB immediately (its Redis key must be deleted)
-		DistributedLock lockBOtherProcess = registry2.obtain(keyB);
-		assertThat(lockBOtherProcess.tryLock(500, TimeUnit.MILLISECONDS))
-				.as("Redis key for '" + keyB + "' was not deleted on unlock — orphaned lock detected")
-				.isTrue();
-		lockBOtherProcess.unlock();
-
-		assertThatNoException().isThrownBy(lockA::unlock);
-
-		registry1.destroy();
-		registry2.destroy();
-	}
-
-	@SuppressWarnings("unchecked")
 	private static Map<String, Lock> getRedisLockRegistryLocks(RedisLockRegistry registry) {
 		return TestUtils.getPropertyValue(registry, "locks", Map.class);
 	}