Skip to content

RedisLock: Throw exception from unlock on expire #2661

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

Merged
merged 2 commits into from
Dec 14, 2018
Merged

RedisLock: Throw exception from unlock on expire #2661

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a2c9abf
RedisLock: Throw exception from unlock on expire
artembilan Dec 13, 2018
3f3aee5
* Create a new test for exception
artembilan Dec 14, 2018
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
48 changes: 27 additions & 21 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 @@ -92,6 +92,18 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl
"return false";


private final Map<String, RedisLock> locks = new ConcurrentHashMap<>();

private final String clientId = UUID.randomUUID().toString();

private final String registryKey;

private final StringRedisTemplate redisTemplate;

private final RedisScript<Boolean> obtainLockScript;

private final long expireAfter;

/**
* An {@link ExecutorService} to call {@link StringRedisTemplate#delete(Object)} in
* the separate thread when the current one is interrupted.
Expand All @@ -105,18 +117,6 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl
*/
private boolean executorExplicitlySet;

private final Map<String, RedisLock> locks = new ConcurrentHashMap<>();

private final String clientId = UUID.randomUUID().toString();

private final String registryKey;

private final StringRedisTemplate redisTemplate;

private final RedisScript<Boolean> obtainLockScript;

private final long expireAfter;

/**
* Constructs a lock registry with the default (60 second) lock expiration.
* @param connectionFactory The connection factory.
Expand Down Expand Up @@ -282,13 +282,17 @@ public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
}

private boolean obtainLock() {
boolean success = RedisLockRegistry.this.redisTemplate.execute(RedisLockRegistry.this.obtainLockScript,
Collections.singletonList(this.lockKey), RedisLockRegistry.this.clientId,
String.valueOf(RedisLockRegistry.this.expireAfter));
if (success) {
Boolean success =
RedisLockRegistry.this.redisTemplate.execute(RedisLockRegistry.this.obtainLockScript,
Collections.singletonList(this.lockKey), RedisLockRegistry.this.clientId,
String.valueOf(RedisLockRegistry.this.expireAfter));

boolean result = Boolean.TRUE.equals(success);

if (result) {
this.lockedAt = System.currentTimeMillis();
}
return success;
return result;
}

@Override
Expand All @@ -301,6 +305,11 @@ public 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);
}
Expand Down Expand Up @@ -377,10 +386,7 @@ public boolean equals(Object obj) {
if (!this.lockKey.equals(other.lockKey)) {
return false;
}
if (this.lockedAt != other.lockedAt) {
return false;
}
return true;
return this.lockedAt == other.lockedAt;
}

private RedisLockRegistry getOuterType() {
Expand Down
81 changes: 47 additions & 34 deletions ...edis/src/test/java/org/springframework/integration/redis/util/RedisLockRegistryTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,13 +78,13 @@ public void setupShutDown() {
}

private StringRedisTemplate createTemplate() {
return new StringRedisTemplate(this.getConnectionFactoryForTest());
return new StringRedisTemplate(getConnectionFactoryForTest());
}

@Test
@RedisAvailable
public void testLock() throws Exception {
RedisLockRegistry registry = new RedisLockRegistry(this.getConnectionFactoryForTest(), this.registryKey);
public void testLock() {
RedisLockRegistry registry = new RedisLockRegistry(getConnectionFactoryForTest(), this.registryKey);
for (int i = 0; i < 10; i++) {
Lock lock = registry.obtain("foo");
lock.lock();
Expand All @@ -102,7 +102,7 @@ public void testLock() throws Exception {
@Test
@RedisAvailable
public void testLockInterruptibly() throws Exception {
RedisLockRegistry registry = new RedisLockRegistry(this.getConnectionFactoryForTest(), this.registryKey);
RedisLockRegistry registry = new RedisLockRegistry(getConnectionFactoryForTest(), this.registryKey);
for (int i = 0; i < 10; i++) {
Lock lock = registry.obtain("foo");
lock.lockInterruptibly();
Expand All @@ -119,8 +119,8 @@ public void testLockInterruptibly() throws Exception {

@Test
@RedisAvailable
public void testReentrantLock() throws Exception {
RedisLockRegistry registry = new RedisLockRegistry(this.getConnectionFactoryForTest(), this.registryKey);
public void testReentrantLock() {
RedisLockRegistry registry = new RedisLockRegistry(getConnectionFactoryForTest(), this.registryKey);
for (int i = 0; i < 10; i++) {
Lock lock1 = registry.obtain("foo");
lock1.lock();
Expand All @@ -146,7 +146,7 @@ public void testReentrantLock() throws Exception {
@Test
@RedisAvailable
public void testReentrantLockInterruptibly() throws Exception {
RedisLockRegistry registry = new RedisLockRegistry(this.getConnectionFactoryForTest(), this.registryKey);
RedisLockRegistry registry = new RedisLockRegistry(getConnectionFactoryForTest(), this.registryKey);
for (int i = 0; i < 10; i++) {
Lock lock1 = registry.obtain("foo");
lock1.lockInterruptibly();
Expand All @@ -172,7 +172,7 @@ public void testReentrantLockInterruptibly() throws Exception {
@Test
@RedisAvailable
public void testTwoLocks() throws Exception {
RedisLockRegistry registry = new RedisLockRegistry(this.getConnectionFactoryForTest(), this.registryKey);
RedisLockRegistry registry = new RedisLockRegistry(getConnectionFactoryForTest(), this.registryKey);
for (int i = 0; i < 10; i++) {
Lock lock1 = registry.obtain("foo");
lock1.lockInterruptibly();
Expand All @@ -198,7 +198,7 @@ public void testTwoLocks() throws Exception {
@Test
@RedisAvailable
public void testTwoThreadsSecondFailsToGetLock() throws Exception {
final RedisLockRegistry registry = new RedisLockRegistry(this.getConnectionFactoryForTest(), this.registryKey);
final RedisLockRegistry registry = new RedisLockRegistry(getConnectionFactoryForTest(), this.registryKey);
final Lock lock1 = registry.obtain("foo");
lock1.lockInterruptibly();
final AtomicBoolean locked = new AtomicBoolean();
Expand Down Expand Up @@ -228,12 +228,12 @@ public void testTwoThreadsSecondFailsToGetLock() throws Exception {
@Test
@RedisAvailable
public void testTwoThreads() throws Exception {
final RedisLockRegistry registry = new RedisLockRegistry(this.getConnectionFactoryForTest(), this.registryKey);
final Lock lock1 = registry.obtain("foo");
final AtomicBoolean locked = new AtomicBoolean();
final CountDownLatch latch1 = new CountDownLatch(1);
final CountDownLatch latch2 = new CountDownLatch(1);
final CountDownLatch latch3 = new CountDownLatch(1);
RedisLockRegistry registry = new RedisLockRegistry(getConnectionFactoryForTest(), this.registryKey);
Lock lock1 = registry.obtain("foo");
AtomicBoolean locked = new AtomicBoolean();
CountDownLatch latch1 = new CountDownLatch(1);
CountDownLatch latch2 = new CountDownLatch(1);
CountDownLatch latch3 = new CountDownLatch(1);
lock1.lockInterruptibly();
assertEquals(1, TestUtils.getPropertyValue(registry, "locks", Map.class).size());
Executors.newSingleThreadExecutor().execute(() -> {
Expand Down Expand Up @@ -266,13 +266,13 @@ public void testTwoThreads() throws Exception {
@Test
@RedisAvailable
public void testTwoThreadsDifferentRegistries() throws Exception {
final RedisLockRegistry registry1 = new RedisLockRegistry(this.getConnectionFactoryForTest(), this.registryKey);
final RedisLockRegistry registry2 = new RedisLockRegistry(this.getConnectionFactoryForTest(), this.registryKey);
final Lock lock1 = registry1.obtain("foo");
final AtomicBoolean locked = new AtomicBoolean();
final CountDownLatch latch1 = new CountDownLatch(1);
final CountDownLatch latch2 = new CountDownLatch(1);
final CountDownLatch latch3 = new CountDownLatch(1);
RedisLockRegistry registry1 = new RedisLockRegistry(getConnectionFactoryForTest(), this.registryKey);
RedisLockRegistry registry2 = new RedisLockRegistry(getConnectionFactoryForTest(), this.registryKey);
Lock lock1 = registry1.obtain("foo");
AtomicBoolean locked = new AtomicBoolean();
CountDownLatch latch1 = new CountDownLatch(1);
CountDownLatch latch2 = new CountDownLatch(1);
CountDownLatch latch3 = new CountDownLatch(1);
lock1.lockInterruptibly();
assertEquals(1, TestUtils.getPropertyValue(registry1, "locks", Map.class).size());
Executors.newSingleThreadExecutor().execute(() -> {
Expand Down Expand Up @@ -313,11 +313,11 @@ public void testTwoThreadsDifferentRegistries() throws Exception {
@Test
@RedisAvailable
public void testTwoThreadsWrongOneUnlocks() throws Exception {
final RedisLockRegistry registry = new RedisLockRegistry(this.getConnectionFactoryForTest(), this.registryKey);
final Lock lock = registry.obtain("foo");
RedisLockRegistry registry = new RedisLockRegistry(getConnectionFactoryForTest(), this.registryKey);
Lock lock = registry.obtain("foo");
lock.lockInterruptibly();
final AtomicBoolean locked = new AtomicBoolean();
final CountDownLatch latch = new CountDownLatch(1);
AtomicBoolean locked = new AtomicBoolean();
CountDownLatch latch = new CountDownLatch(1);
Future<Object> result = Executors.newSingleThreadExecutor().submit(() -> {
try {
lock.unlock();
Expand All @@ -341,8 +341,8 @@ public void testTwoThreadsWrongOneUnlocks() throws Exception {
@Test
@RedisAvailable
public void testExpireTwoRegistries() throws Exception {
RedisLockRegistry registry1 = new RedisLockRegistry(this.getConnectionFactoryForTest(), this.registryKey, 100);
RedisLockRegistry registry2 = new RedisLockRegistry(this.getConnectionFactoryForTest(), this.registryKey, 100);
RedisLockRegistry registry1 = new RedisLockRegistry(getConnectionFactoryForTest(), this.registryKey, 1);
RedisLockRegistry registry2 = new RedisLockRegistry(getConnectionFactoryForTest(), this.registryKey, 1);
Lock lock1 = registry1.obtain("foo");
Lock lock2 = registry2.obtain("foo");
assertTrue(lock1.tryLock());
Expand All @@ -354,8 +354,21 @@ public void testExpireTwoRegistries() throws Exception {

@Test
@RedisAvailable
public void testEquals() throws Exception {
RedisConnectionFactory connectionFactory = this.getConnectionFactoryForTest();
public void testExceptionOnExpire() throws Exception {
RedisLockRegistry registry = new RedisLockRegistry(getConnectionFactoryForTest(), this.registryKey, 1);
Lock lock1 = registry.obtain("foo");
assertTrue(lock1.tryLock());
this.thrown.expect(IllegalStateException.class);
this.thrown.expectMessage("Lock was released in the store due to expiration.");
waitForExpire("foo");
lock1.unlock();
}


@Test
@RedisAvailable
public void testEquals() {
RedisConnectionFactory connectionFactory = getConnectionFactoryForTest();
RedisLockRegistry registry1 = new RedisLockRegistry(connectionFactory, this.registryKey);
RedisLockRegistry registry2 = new RedisLockRegistry(connectionFactory, this.registryKey);
RedisLockRegistry registry3 = new RedisLockRegistry(connectionFactory, this.registryKey2);
Expand Down Expand Up @@ -388,7 +401,7 @@ public void testEquals() throws Exception {
@Test
@RedisAvailable
public void testThreadLocalListLeaks() {
RedisLockRegistry registry = new RedisLockRegistry(this.getConnectionFactoryForTest(), this.registryKey, 100);
RedisLockRegistry registry = new RedisLockRegistry(getConnectionFactoryForTest(), this.registryKey, 100);

for (int i = 0; i < 10; i++) {
registry.obtain("foo" + i);
Expand All @@ -411,7 +424,7 @@ public void testThreadLocalListLeaks() {
@Test
@RedisAvailable
public void testExpireNotChanged() throws Exception {
RedisConnectionFactory connectionFactory = this.getConnectionFactoryForTest();
RedisConnectionFactory connectionFactory = getConnectionFactoryForTest();
final RedisLockRegistry registry = new RedisLockRegistry(connectionFactory, this.registryKey, 10000);
Lock lock = registry.obtain("foo");
lock.lock();
Expand All @@ -429,13 +442,13 @@ public void testExpireNotChanged() throws Exception {
}

private Long getExpire(RedisLockRegistry registry, String lockKey) {
StringRedisTemplate template = this.createTemplate();
StringRedisTemplate template = createTemplate();
String registryKey = TestUtils.getPropertyValue(registry, "registryKey", String.class);
return template.getExpire(registryKey + ":" + lockKey);
}

private void waitForExpire(String key) throws Exception {
StringRedisTemplate template = this.createTemplate();
StringRedisTemplate template = createTemplate();
int n = 0;
while (n++ < 100 && template.keys(this.registryKey + ":" + key).size() > 0) {
Thread.sleep(100);
Expand Down