Adjust cache capacity in the JdbcLockRegistry (#11172)

Fixes: #11172

When `cacheCapacity` has shrunk, and we obtain already held lock again,
it is evicted due to the current `if (this.locks.size() > this.cacheCapacity)` logic.
This leads to several side effects for already locked instance.

* Implement the logic for shrinking `cacheCapacity`:
evict unsed lock when cache size is higher than new `cacheCapacity`.
* Fix `obtain()` logic to skip eviction attempt if obtained lock is held

**Auto-cherry-pick to `7.0.x`**
# Conflicts:
#	spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/lock/JdbcLockRegistryTests.java

Author: artembilan

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

@@ -139,10 +139,17 @@ public void setIdleBetweenTries(Duration idleBetweenTries) {
 
 	/**
 	 * Set the capacity of cached 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;
 	}
 
@@ -154,7 +161,8 @@ public DistributedLock obtain(Object lockKey) {
 		try {
 			JdbcLock jdbcLock =
 					this.locks.computeIfAbsent(lockKeyToUse, key -> new JdbcLock(lockKeyToUse));
-			if (this.locks.size() > this.cacheCapacity) {
+
+			if (!jdbcLock.isAcquiredInThisProcess() && this.locks.size() > this.cacheCapacity) {
 				long now = System.currentTimeMillis();
 				this.locks.entrySet()
 						.removeIf(entry -> {
@@ -168,6 +176,7 @@ public DistributedLock obtain(Object lockKey) {
 							"from " + this + ". Cannot obtain more at the moment.");
 				}
 			}
+
 			return jdbcLock;
 		}
 		finally {
@@ -187,7 +196,7 @@ public void expireUnusedOlderThan(long age) {
 			this.locks.entrySet()
 					.removeIf(entry -> {
 						JdbcLock lock = entry.getValue();
-						return now - lock.getLastUsed() > age && !lock.isAcquiredInThisProcess();
+						return now - lock.getLastUsed() >= age && !lock.isAcquiredInThisProcess();
 					});
 		}
 		finally {

spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/lock/JdbcLockRegistryTests.java

@@ -727,6 +727,72 @@ void unusedLockIsStale() {
 				.withMessageStartingWith("The lock 'lock1' was evicted from the exhausted cache due to its unused period for");
 	}
 
+	@Test
+	void shrinkingCacheCapacityEvictsUnusedEntries() {
+		DefaultLockRepository client = new DefaultLockRepository(dataSource);
+		client.setApplicationContext(this.context);
+		client.afterPropertiesSet();
+		client.afterSingletonsInstantiated();
+		JdbcLockRegistry registry = new JdbcLockRegistry(client);
+		registry.setCacheCapacity(2);
+
+		DistributedLock lock1 = registry.obtain("lock1");
+		lock1.lock();
+		try {
+			DistributedLock lock2 = registry.obtain("lock2");
+			registry.setCacheCapacity(1);
+			assertThatExceptionOfType(CannotAcquireLockException.class)
+					.isThrownBy(lock2::lock)
+					.havingCause()
+					.isInstanceOf(IllegalStateException.class)
+					.withMessageStartingWith(
+							"The lock 'lock2' was evicted from the exhausted cache due to its unused period for");
+		}
+		finally {
+			lock1.unlock();
+		}
+	}
+
+	@Test
+	void obtainDoesNotEvictCurrentlyHeldLockOfRequestedKey() {
+		DefaultLockRepository client = new DefaultLockRepository(dataSource);
+		client.setApplicationContext(this.context);
+		client.afterPropertiesSet();
+		client.afterSingletonsInstantiated();
+		JdbcLockRegistry registry = new JdbcLockRegistry(client);
+		registry.setCacheCapacity(2);
+
+		DistributedLock lock1 = registry.obtain("lock1");
+		lock1.lock();
+		try {
+			DistributedLock lock2 = registry.obtain("lock2");
+			lock2.lock();
+			try {
+				// Shrinking the capacity below the number of currently held locks forces
+				// the next `obtain` for an already-cached, held key to see `size() > cacheCapacity`.
+				registry.setCacheCapacity(1);
+
+				Map<String, Lock> registryLocks = getRegistryLocks(registry);
+
+				assertThat(registryLocks).containsKeys("lock1", "lock2");
+
+				assertThat(registry.obtain("lock1")).isSameAs(lock1);
+
+				assertThat(registryLocks).containsKeys("lock1", "lock2");
+
+				assertThatNoException().isThrownBy(lock1::lock);
+
+				lock1.unlock();
+			}
+			finally {
+				lock2.unlock();
+			}
+		}
+		finally {
+			lock1.unlock();
+		}
+	}
+
 	@SuppressWarnings("unchecked")
 	private static Map<String, Lock> getRegistryLocks(JdbcLockRegistry registry) {
 		return TestUtils.getPropertyValue(registry, "locks", Map.class);