GH-10471: Add the lockKey to the channel to control Redis cluster slot hashing

Fixes: #10471

As a consequence, change the `ChannelTopic` to a `PatternTopic` to keep receiving subscription events for Pub/Sub lock

* Pass `unLockChannelKey` as `KEYS[2]` instead of `ARGV[2]` as values used as keys should be passed as such

Signed-off-by: Severin Kistler <[email protected]>

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

(cherry picked from commit 00fa656)

Author: kistlers

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

@@ -23,6 +23,7 @@
 import java.util.ConcurrentModificationException;
 import java.util.Date;
 import java.util.LinkedHashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.UUID;
@@ -52,7 +53,7 @@
 import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.data.redis.core.script.DefaultRedisScript;
 import org.springframework.data.redis.core.script.RedisScript;
-import org.springframework.data.redis.listener.ChannelTopic;
+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.ExpirableLockRegistry;
@@ -94,6 +95,7 @@
  * @author Roman Zabaluev
  * @author Alex Peelman
  * @author Youbin Wu
+ * @author Severin Kistler
  *
  * @since 4.0
  *
@@ -199,7 +201,7 @@ private void setupUnlockMessageListener(RedisConnectionFactory connectionFactory
 				"'unlockNotifyMessageListener' must not have been re-initialized.");
 		RedisLockRegistry.this.redisMessageListenerContainer = new RedisMessageListenerContainer();
 		RedisLockRegistry.this.unlockNotifyMessageListener = new RedisPubSubLock.RedisUnLockNotifyMessageListener();
-		final Topic topic = new ChannelTopic(this.unLockChannelKey);
+		final Topic topic = new PatternTopic(this.unLockChannelKey + ":*");
 		this.redisMessageListenerContainer.setConnectionFactory(connectionFactory);
 		this.redisMessageListenerContainer.setTaskExecutor(this.executor);
 		this.redisMessageListenerContainer.setSubscriptionExecutor(this.executor);
@@ -665,7 +667,7 @@ private final class RedisPubSubLock extends RedisLock {
 		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])
+					redis.call('PUBLISH', KEYS[2], KEYS[1])
 					return true
 				end
 				return false
@@ -674,7 +676,7 @@ private final class RedisPubSubLock extends RedisLock {
 		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])
+					redis.call('PUBLISH', KEYS[2], KEYS[1])
 					return true
 				end
 				return false
@@ -706,9 +708,10 @@ protected boolean removeLockKeyInnerDelete() {
 		}
 
 		private boolean removeLockKeyWithScript(RedisScript<Boolean> redisScript) {
+			String unLockChannelKeyToUse = RedisLockRegistry.this.unLockChannelKey + ":" + this.lockKey;
 			return Boolean.TRUE.equals(RedisLockRegistry.this.redisTemplate.execute(
-					redisScript, Collections.singletonList(this.lockKey),
-					RedisLockRegistry.this.clientId, RedisLockRegistry.this.unLockChannelKey));
+					redisScript, List.of(this.lockKey, unLockChannelKeyToUse),
+					RedisLockRegistry.this.clientId));
 		}
 
 		private boolean subscribeLock(long time) throws ExecutionException, InterruptedException {