GH-11044: Revise cache in RedisLockRegistry

- Use a DefaultLockRegistry to manage local lock instances.
- Document the internal registry changes

Fixes: #11044

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

Author: cppwfs

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

@@ -57,6 +57,7 @@
 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;
@@ -99,6 +100,7 @@
  * @author Youbin Wu
  * @author Michal Domagala
  * @author Severin Kistler
+ * @author Glenn Renfro
  *
  * @since 4.0
  *
@@ -110,14 +112,16 @@ public final class RedisLockRegistry
 
 	private static final long DEFAULT_EXPIRE_AFTER = 60000L;
 
-	private static final int DEFAULT_CAPACITY = 100_000;
+	private static final int DEFAULT_CAPACITY = 256;
 
 	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) {
 
@@ -243,11 +247,14 @@ public void setRenewalTaskScheduler(TaskScheduler renewalTaskScheduler) {
 
 	/**
 	 * Set the capacity of cached locks.
-	 * @param cacheCapacity The capacity of cached lock, (default 100_000).
+	 * @param cacheCapacity The capacity of cached lock, (default 256 locks).
 	 * @since 5.5.6
 	 */
 	public void setCacheCapacity(int cacheCapacity) {
 		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);
 	}
 
 	/**
@@ -403,14 +410,15 @@ private abstract class RedisLock implements DistributedLock {
 
 		protected final String lockKey;
 
-		private final ReentrantLock localLock = new ReentrantLock();
+		private final ReentrantLock localLock;
 
 		private volatile long lockedAt;
 
 		private volatile @Nullable ScheduledFuture<?> renewFuture;
 
 		private RedisLock(String path) {
 			this.lockKey = constructLockKey(path);
+			this.localLock = (ReentrantLock) RedisLockRegistry.this.defaultLockRegistry.obtain(this.lockKey);
 		}
 
 		private String constructLockKey(String path) {

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

@@ -1018,6 +1018,50 @@ void testInitialiseWithCustomExecutor() {
 		assertThatNoException().isThrownBy(() -> redisLockRegistry.setExecutor(mock()));
 	}
 
+	@Test
+	void noSecondLockOnEviction() throws InterruptedException {
+		RedisLockRegistry registry = new RedisLockRegistry(redisConnectionFactory, this.registryKey);
+		registry.setRedisLockType(testRedisLockType);
+		registry.setCacheCapacity(2);
+
+		CountDownLatch lock1Latch = new CountDownLatch(1);
+		CountDownLatch furtherLocksLatch = new CountDownLatch(1);
+
+		Executors.newSingleThreadExecutor()
+				.execute(() -> {
+					DistributedLock lock1 = registry.obtain("lock1");
+					lock1.lock();
+					try {
+						furtherLocksLatch.countDown();
+						lock1Latch.await(10, TimeUnit.SECONDS);
+					}
+					catch (InterruptedException e) {
+						Thread.currentThread().interrupt();
+					}
+					finally {
+						lock1.unlock();
+					}
+				});
+
+		assertThat(furtherLocksLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		// Two new locks to trigger cache eviction for the 'lock1'
+		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
+		assertThat(lock1.tryLock(1, TimeUnit.SECONDS)).isFalse();
+
+		lock1Latch.countDown();
+
+		assertThatNoException().isThrownBy(() -> {
+			lock1.lock();
+			lock1.unlock();
+		});
+	}
+
 	private Long getExpire(RedisLockRegistry registry, String lockKey) {
 		StringRedisTemplate template = createTemplate();
 		String registryKey = TestUtils.getPropertyValue(registry, "registryKey", String.class);

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

@@ -846,7 +846,11 @@ The expiry should be set at a large enough value to prevent this condition, but
 Starting with version 5.0, the `RedisLockRegistry` implements `ExpirableLockRegistry`, which removes locks last acquired more than `age` ago and that are not currently locked.
 
 Starting with version 5.5.6, the `RedisLockRegistry` is support automatically clean up cache for redisLocks in `RedisLockRegistry.locks` via `RedisLockRegistry.setCacheCapacity()`.
+The default value is `256`.
 See its JavaDocs for more information.
+This capacity number is also propagated into the `DefaultLockRegistry` used internally in the `RedisLockRegistry` for a pool of shared `ReentrantLock` instances.
+So, if higher concurrent access (the number of unique lock keys used in parallel) is expected (or allowed) for the application, the `cacheCapacity` should be increased respectively, with the closest power of 2 in mind (essentially `Integer.highestOneBit(cacheCapacity + 1)`).
+
 
 Starting with version 5.5.13, the `RedisLockRegistry` exposes a `setRedisLockType(RedisLockType)` option to determine in which mode a Redis lock acquisition should happen:
 
@@ -887,4 +891,4 @@ RedisLockRegistry lockRegistry = new RedisLockRegistry("my-lock-key{choose_your_
 lockRegistry.lock();
 # critical section
 lockRegistry.unlock();
-----
+----