GH-8699: Atomic Redis script for unlock()

Expected Behavior

Using a single Lua script to verify ownership of the lock and remove it.

Current Behavior

`unlock()` method of `RedisLock` uses two separate Redis operations:

    * `isAcquiredInThisProcess()`` method executes a `GET` operation to verify if the lock is owned by the process.
    * `removeLockKey()`` method executes `UNLINK/DEL` operation to remove the lock.

* The `removeLockKeyInnerUnlink()`, and `removeLockKeyInnerDelete()`
methods will execute a script both verify ownership of the lock and remove it.

**Cherry-pick to `6.1.x` & `6.0.x`**

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

Author: EddieChoCho

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

@@ -82,6 +82,7 @@
  * @author Vedran Pavic
  * @author Unseok Kim
  * @author Anton Gabov
+ * @author Eddie Cho
  *
  * @since 4.0
  *
@@ -95,7 +96,7 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl
 	private static final int DEFAULT_CAPACITY = 100_000;
 
 	private final Map<String, RedisLock> locks =
-			new LinkedHashMap<String, RedisLock>(16, 0.75F, true) {
+			new LinkedHashMap<>(16, 0.75F, true) {
 
 				@Override
 				protected boolean removeEldestEntry(Entry<String, RedisLock> eldest) {
@@ -331,12 +332,12 @@ public long getLockedAt() {
 		/**
 		 * Unlock the lock using the unlink method in redis.
 		 */
-		protected abstract void removeLockKeyInnerUnlink();
+		protected abstract boolean removeLockKeyInnerUnlink();
 
 		/**
 		 * Unlock the lock using the delete method in redis.
 		 */
-		protected abstract void removeLockKeyInnerDelete();
+		protected abstract boolean removeLockKeyInnerDelete();
 
 		@Override
 		public final void lock() {
@@ -442,11 +443,6 @@ public final void unlock() {
 				return;
 			}
 			try {
-				if (!isAcquiredInThisProcess()) {
-					throw new IllegalStateException("Lock was released in the store due to expiration. " +
-							"The integrity of data protected by this lock may have been compromised.");
-				}
-
 				if (Thread.currentThread().isInterrupted()) {
 					RedisLockRegistry.this.executor.execute(this::removeLockKey);
 				}
@@ -469,7 +465,11 @@ public final void unlock() {
 		private void removeLockKey() {
 			if (RedisLockRegistry.this.unlinkAvailable) {
 				try {
-					removeLockKeyInnerUnlink();
+					boolean unlinkResult = removeLockKeyInnerUnlink();
+					if (!unlinkResult) {
+						throw new IllegalStateException("Lock was released in the store due to expiration. " +
+								"The integrity of data protected by this lock may have been compromised.");
+					}
 					return;
 				}
 				catch (Exception ex) {
@@ -484,7 +484,10 @@ private void removeLockKey() {
 					}
 				}
 			}
-			removeLockKeyInnerDelete();
+			if (!removeLockKeyInnerDelete()) {
+				throw new IllegalStateException("Lock was released in the store due to expiration. " +
+						"The integrity of data protected by this lock may have been compromised.");
+			}
 		}
 
 		@Override
@@ -547,19 +550,23 @@ private RedisLockRegistry getOuterType() {
 
 	private final class RedisPubSubLock extends RedisLock {
 
-		private static final String UNLINK_UNLOCK_SCRIPT =
-				"if (redis.call('unlink', KEYS[1]) == 1) then " +
-						"redis.call('publish', ARGV[1], KEYS[1]) " +
-						"return true " +
-						"end " +
-						"return false";
-
-		private static final String DELETE_UNLOCK_SCRIPT =
-				"if (redis.call('del', KEYS[1]) == 1) then " +
-						"redis.call('publish', ARGV[1], KEYS[1]) " +
-						"return true " +
-						"end " +
-						"return false";
+		private static final String UNLINK_UNLOCK_SCRIPT = """
+				local lockClientId = redis.call('GET', KEYS[1])
+				if (lockClientId == ARGV[1] and redis.call('UNLINK', KEYS[1]) == 1) then
+					redis.call('PUBLISH', ARGV[2], KEYS[1])
+					return true
+				end
+				return false
+				""";
+
+		private static final String DELETE_UNLOCK_SCRIPT = """
+				local lockClientId = redis.call('GET', KEYS[1])
+				if (lockClientId == ARGV[1] and redis.call('DEL', KEYS[1]) == 1) then
+					redis.call('PUBLISH', ARGV[2], KEYS[1])
+					return true
+				end
+				return false
+				""";
 
 		private static final RedisScript<Boolean>
 				UNLINK_UNLOCK_REDIS_SCRIPT = new DefaultRedisScript<>(UNLINK_UNLOCK_SCRIPT, Boolean.class);
@@ -577,18 +584,19 @@ protected boolean tryRedisLockInner(long time) throws ExecutionException, Interr
 		}
 
 		@Override
-		protected void removeLockKeyInnerUnlink() {
-			RedisLockRegistry.this.redisTemplate.execute(
-					UNLINK_UNLOCK_REDIS_SCRIPT, Collections.singletonList(this.lockKey),
-					RedisLockRegistry.this.unLockChannelKey);
+		protected boolean removeLockKeyInnerUnlink() {
+			return removeLockKeyWithScript(UNLINK_UNLOCK_REDIS_SCRIPT);
 		}
 
 		@Override
-		protected void removeLockKeyInnerDelete() {
-			RedisLockRegistry.this.redisTemplate.execute(
-					DELETE_UNLOCK_REDIS_SCRIPT, Collections.singletonList(this.lockKey),
-					RedisLockRegistry.this.unLockChannelKey);
+		protected boolean removeLockKeyInnerDelete() {
+			return removeLockKeyWithScript(DELETE_UNLOCK_REDIS_SCRIPT);
+		}
 
+		private boolean removeLockKeyWithScript(RedisScript<Boolean> redisScript) {
+			return Boolean.TRUE.equals(RedisLockRegistry.this.redisTemplate.execute(
+					redisScript, Collections.singletonList(this.lockKey),
+					RedisLockRegistry.this.clientId, RedisLockRegistry.this.unLockChannelKey));
 		}
 
 		private boolean subscribeLock(long time) throws ExecutionException, InterruptedException {
@@ -678,6 +686,30 @@ private void unlockNotify(String lockKey) {
 
 	private final class RedisSpinLock extends RedisLock {
 
+		private static final String UNLINK_UNLOCK_SCRIPT = """
+				local lockClientId = redis.call('GET', KEYS[1])
+				if lockClientId == ARGV[1] then
+					redis.call('UNLINK', KEYS[1])
+					return true
+				end
+				return false
+				""";
+
+		private static final String DELETE_UNLOCK_SCRIPT = """
+				local lockClientId = redis.call('GET', KEYS[1])
+				if lockClientId == ARGV[1] then
+					redis.call('DEL', KEYS[1])
+					return true
+				end
+				return false
+				""";
+
+		private static final RedisScript<Boolean>
+				UNLINK_UNLOCK_REDIS_SCRIPT = new DefaultRedisScript<>(UNLINK_UNLOCK_SCRIPT, Boolean.class);
+
+		private static final RedisScript<Boolean>
+				DELETE_UNLOCK_REDIS_SCRIPT = new DefaultRedisScript<>(DELETE_UNLOCK_SCRIPT, Boolean.class);
+
 		private RedisSpinLock(String path) {
 			super(path);
 		}
@@ -702,13 +734,19 @@ protected boolean tryRedisLockInner(long time) throws InterruptedException {
 		}
 
 		@Override
-		protected void removeLockKeyInnerUnlink() {
-			RedisLockRegistry.this.redisTemplate.unlink(this.lockKey);
+		protected boolean removeLockKeyInnerUnlink() {
+			return removeLockKeyWithScript(UNLINK_UNLOCK_REDIS_SCRIPT);
 		}
 
 		@Override
-		protected void removeLockKeyInnerDelete() {
-			RedisLockRegistry.this.redisTemplate.delete(this.lockKey);
+		protected boolean removeLockKeyInnerDelete() {
+			return removeLockKeyWithScript(DELETE_UNLOCK_REDIS_SCRIPT);
+		}
+
+		private boolean removeLockKeyWithScript(RedisScript<Boolean> redisScript) {
+			return Boolean.TRUE.equals(RedisLockRegistry.this.redisTemplate.execute(
+					redisScript, Collections.singletonList(this.lockKey),
+					RedisLockRegistry.this.clientId));
 		}
 
 	}