Spring/spring-integration-extensions
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-183: LeaderInitiator: Catch unlock errors

Browse Source 
Fixes https://github.com/spring-projects/spring-integration-extensions/issues/183

When we get an exception during `lock.unlock()`, we don't revoke leadership.
If Hazelcast is down, this may cause a race condition when the second
candidate is selected as leader when connection comes back

* Catch `lock.unlock()` exceptions and log them under DEBUG.
This way we proceed to the `handleRevoked()` logic
This commit is contained in:
Artem Bilan
2018-07-13 13:17:27 -04:00
parent 2e567f24ba
commit 2486fdaa41
1 changed files with 46 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/leader/LeaderInitiator.java
View File

@@ -304,34 +304,70 @@ public class LeaderInitiator implements SmartLifecycle, DisposableBean, Applicat
}
catch (Exception e) {
if (this.locked) {
LeaderInitiator.this.lock.unlock();
this.locked = false;
try {
LeaderInitiator.this.lock.unlock();
}
catch (Exception e1) {
logger.debug("Could not unlock - treat as broken " + this.context +
". Revoking " + (isRunning() ? " and retrying..." : "..."), e1);
}
// The lock was broken and we are no longer leader
handleRevoked();
if (isRunning()) {
// Give it a chance to elect some other leader.
Thread.sleep(LeaderInitiator.this.busyWaitMillis);
}
}
if (e instanceof InterruptedException) {
if (e instanceof InterruptedException || Thread.currentThread().isInterrupted()) {
Thread.currentThread().interrupt();
if (isRunning()) {
logger.warn("Restarting LeaderSelector because of error.", e);
LeaderInitiator.this.future = LeaderInitiator.this.executorService.submit(this);
logger.warn("Restarting LeaderSelector for " + this.context + " because of error.", e);
LeaderInitiator.this.future =
LeaderInitiator.this.executorService.submit(
new Callable<Void>() {
@Override
public Void call() throws Exception {
// Give it a chance to elect some other leader.
Thread.sleep(LeaderInitiator.this.busyWaitMillis);
return LeaderSelector.this.call();
}
});
}
return null;
}
else {
if (isRunning()) {
// Give it a chance to elect some other leader.
try {
Thread.sleep(LeaderInitiator.this.busyWaitMillis);
}
catch (InterruptedException e1) {
// Ignore interruption and let it to be caught on the next cycle.
Thread.currentThread().interrupt();
}
}
if (logger.isDebugEnabled()) {
logger.debug("Error acquiring the lock for " + this.context +
". " + (isRunning() ? "Retrying..." : ""), e);
}
}
}
}
}
finally {
if (this.locked) {
LeaderInitiator.this.lock.unlock();
this.locked = false;
try {
LeaderInitiator.this.lock.unlock();
}
catch (Exception e) {
logger.debug("Could not unlock during stop for " + this.context
+ " - treat as broken. Revoking...", e);
}
// We are stopping, therefore not leading any more
handleRevoked();
}
this.locked = false;
}
return null;
}