Skip to content

Support for PubSub Lock on AWS Elasticache Valkey #10473

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 2 commits into
base: main
Choose a base branch
from
+8 −5
Open

Support for PubSub Lock on AWS Elasticache Valkey #10473

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a578e0d
GH-10471: add the lockKey to the unlockChannelKey to control cluster …
kistlers Oct 9, 2025
6804c03
GH-10471: pass unLockChannelKey as `KEYS[2]` instead of `ARGV[2]` as …
kistlers Oct 9, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 8 additions & 5 deletions ...ion-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -53,7 +54,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;
Expand Down Expand Up @@ -97,6 +98,7 @@
* @author Alex Peelman
* @author Youbin Wu
* @author Michal Domagala
* @author Severin Kistler
*
* @since 4.0
*
Expand Down Expand Up @@ -213,7 +215,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);
Expand Down Expand Up @@ -667,7 +669,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
Expand All @@ -688,9 +690,10 @@ protected boolean tryRedisLockInner(long time, long expireAfter)

@Override
protected boolean removeLockKeyInnerUnlink() {
final String unLockChannelKey = RedisLockRegistry.this.unLockChannelKey + ":" + this.lockKey;
return Boolean.TRUE.equals(RedisLockRegistry.this.redisTemplate.execute(
UNLINK_UNLOCK_REDIS_SCRIPT, Collections.singletonList(this.lockKey),
RedisLockRegistry.this.clientId, RedisLockRegistry.this.unLockChannelKey));
UNLINK_UNLOCK_REDIS_SCRIPT, List.of(this.lockKey, unLockChannelKey),
RedisLockRegistry.this.clientId));
}

private boolean subscribeLock(long time, long expireAfter) throws ExecutionException, InterruptedException {
Expand Down