INT-4396: Add retrying lock in case of exception
JIRA: https://jira.spring.io/browse/INT-4396 When target distributed `Lock` implementation throws an exception, e.g. in case of no connection to the service, the `LockRegistryLeaderInitiator` exists the loop and can come back to the elections only after restart * Catch all the exception on `this.lock.tryLock()` and resubmit `LeaderSelector` for a new locking cycle if `LockRegistryLeaderInitiator.isRunning()` and `InterruptedException` * Remove diagnostics from the `JdbcLockRegistryLeaderInitiatorTests` since this fix confirms that we just didn't have a reconnect logic before when this test failed sporadically **Cherry-pick to 4.3.x**
This commit is contained in:
committed by
Gary Russell
parent
e53e8941a1
commit
09aeaac4da
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2016-2017 the original author or authors.
|
||||
* Copyright 2016-2018 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
@@ -326,7 +326,7 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe
|
||||
@Override
|
||||
public Void call() throws Exception {
|
||||
try {
|
||||
while (LockRegistryLeaderInitiator.this.running) {
|
||||
while (isRunning()) {
|
||||
try {
|
||||
// We always try to acquire the lock, in case it expired
|
||||
boolean acquired = this.lock.tryLock(LockRegistryLeaderInitiator.this.heartBeatMillis,
|
||||
@@ -345,26 +345,40 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe
|
||||
// If we were able to acquire it but we were already locked we
|
||||
// should release it
|
||||
this.lock.unlock();
|
||||
// Give it a chance to expire.
|
||||
Thread.sleep(LockRegistryLeaderInitiator.this.heartBeatMillis);
|
||||
if (isRunning()) {
|
||||
// Give it a chance to expire.
|
||||
Thread.sleep(LockRegistryLeaderInitiator.this.heartBeatMillis);
|
||||
}
|
||||
}
|
||||
else {
|
||||
this.locked = false;
|
||||
// We were not able to acquire it, therefore not leading any more
|
||||
handleRevoked();
|
||||
// Try again quickly in case the lock holder dropped it
|
||||
Thread.sleep(LockRegistryLeaderInitiator.this.busyWaitMillis);
|
||||
if (isRunning()) {
|
||||
// Try again quickly in case the lock holder dropped it
|
||||
Thread.sleep(LockRegistryLeaderInitiator.this.busyWaitMillis);
|
||||
}
|
||||
}
|
||||
}
|
||||
catch (InterruptedException e) {
|
||||
catch (Exception e) {
|
||||
if (this.locked) {
|
||||
this.lock.unlock();
|
||||
this.locked = false;
|
||||
// The lock was broken and we are no longer leader
|
||||
handleRevoked();
|
||||
// Give it a chance to elect some other leader.
|
||||
Thread.sleep(LockRegistryLeaderInitiator.this.busyWaitMillis);
|
||||
if (isRunning()) {
|
||||
// Give it a chance to elect some other leader.
|
||||
Thread.sleep(LockRegistryLeaderInitiator.this.busyWaitMillis);
|
||||
}
|
||||
}
|
||||
|
||||
if (e instanceof InterruptedException) {
|
||||
Thread.currentThread().interrupt();
|
||||
if (isRunning()) {
|
||||
logger.warn("Restarting LeaderSelector because of error.", e);
|
||||
LockRegistryLeaderInitiator.this.future =
|
||||
LockRegistryLeaderInitiator.this.executorService.submit(this);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
}
|
||||
@@ -446,11 +460,6 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe
|
||||
public void yield() {
|
||||
if (LockRegistryLeaderInitiator.this.future != null) {
|
||||
LockRegistryLeaderInitiator.this.future.cancel(true);
|
||||
if (isRunning()) {
|
||||
LockRegistryLeaderInitiator.this.future =
|
||||
LockRegistryLeaderInitiator.this.executorService
|
||||
.submit(LockRegistryLeaderInitiator.this.leaderSelector);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2012-2017 the original author or authors.
|
||||
* Copyright 2012-2018 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
@@ -265,6 +265,38 @@ public class LockRegistryLeaderInitiatorTests {
|
||||
assertNull(throwable);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testExceptionFromLock() throws Exception {
|
||||
Lock mockLock = mock(Lock.class);
|
||||
|
||||
AtomicBoolean exceptionThrown = new AtomicBoolean();
|
||||
|
||||
willAnswer(invocation -> {
|
||||
if (!exceptionThrown.getAndSet(true)) {
|
||||
throw new RuntimeException("lock is broken");
|
||||
}
|
||||
else {
|
||||
return true;
|
||||
}
|
||||
}).given(mockLock).tryLock(anyLong(), any(TimeUnit.class));
|
||||
|
||||
LockRegistry registry = lockKey -> mockLock;
|
||||
|
||||
CountDownLatch onGranted = new CountDownLatch(1);
|
||||
|
||||
LockRegistryLeaderInitiator initiator = new LockRegistryLeaderInitiator(registry);
|
||||
|
||||
initiator.setLeaderEventPublisher(new CountingPublisher(onGranted));
|
||||
|
||||
initiator.start();
|
||||
|
||||
assertTrue(onGranted.await(10, TimeUnit.SECONDS));
|
||||
assertTrue(initiator.getContext().isLeader());
|
||||
assertTrue(exceptionThrown.get());
|
||||
|
||||
initiator.stop();
|
||||
}
|
||||
|
||||
private static class CountingPublisher implements LeaderEventPublisher {
|
||||
|
||||
private final CountDownLatch granted;
|
||||
|
||||
@@ -27,7 +27,6 @@ import java.util.concurrent.TimeUnit;
|
||||
|
||||
import org.junit.AfterClass;
|
||||
import org.junit.BeforeClass;
|
||||
import org.junit.Rule;
|
||||
import org.junit.Test;
|
||||
|
||||
import org.springframework.integration.jdbc.lock.DefaultLockRepository;
|
||||
@@ -36,7 +35,6 @@ import org.springframework.integration.leader.Context;
|
||||
import org.springframework.integration.leader.DefaultCandidate;
|
||||
import org.springframework.integration.leader.event.LeaderEventPublisher;
|
||||
import org.springframework.integration.support.leader.LockRegistryLeaderInitiator;
|
||||
import org.springframework.integration.test.rule.Log4j2LevelAdjuster;
|
||||
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase;
|
||||
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
|
||||
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType;
|
||||
@@ -52,14 +50,6 @@ public class JdbcLockRegistryLeaderInitiatorTests {
|
||||
|
||||
public static EmbeddedDatabase dataSource;
|
||||
|
||||
@Rule
|
||||
public Log4j2LevelAdjuster adjuster =
|
||||
Log4j2LevelAdjuster.trace()
|
||||
.categories("org.springframework.integration",
|
||||
"org.springframework.integration.jdbc",
|
||||
"org.springframework.jdbc",
|
||||
"org.apache.derby");
|
||||
|
||||
@BeforeClass
|
||||
public static void init() {
|
||||
dataSource = new EmbeddedDatabaseBuilder()
|
||||
|
||||
Reference in New Issue
Block a user