add the lockKey to the unlockChannelKey to control cluster slot hashing

As a consequence, change the ChannelTopic to a PatternTopic to keep receiving subscription events for PubSub lock

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

Author: kistlers

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

@@ -53,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.DistributedLock;
@@ -97,6 +97,7 @@
  * @author Alex Peelman
  * @author Youbin Wu
  * @author Michal Domagala
+ * @author Severin Kistler
  *
  * @since 4.0
  *
@@ -213,7 +214,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 + ":*");
 		RedisMessageListenerContainer container = RedisLockRegistry.this.redisMessageListenerContainer;
 		RedisPubSubLock.RedisUnLockNotifyMessageListener listener = RedisLockRegistry.this.unlockNotifyMessageListener;
 		container.setConnectionFactory(connectionFactory);
@@ -690,7 +691,7 @@ protected boolean tryRedisLockInner(long time, long expireAfter)
 		protected boolean removeLockKeyInnerUnlink() {
 			return Boolean.TRUE.equals(RedisLockRegistry.this.redisTemplate.execute(
 					UNLINK_UNLOCK_REDIS_SCRIPT, Collections.singletonList(this.lockKey),
-					RedisLockRegistry.this.clientId, RedisLockRegistry.this.unLockChannelKey));
+					RedisLockRegistry.this.clientId, RedisLockRegistry.this.unLockChannelKey + ":" + this.lockKey));
 		}
 
 		private boolean subscribeLock(long time, long expireAfter) throws ExecutionException, InterruptedException {