Skip to content

Some LockRegistryLeaderInitiator improvements #8570

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
Mar 15, 2023
Merged

Some LockRegistryLeaderInitiator improvements #8570

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
7e98b3a
Some `LockRegistryLeaderInitiator` improvements
artembilan Mar 14, 2023
4102aec
* Introduce a `LeaderSelector.yielding` flag
artembilan Mar 15, 2023
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
36 changes: 23 additions & 13 deletions ...main/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
/*
* Copyright 2016-2021 the original author or authors.
* Copyright 2016-2023 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.
Expand Down Expand Up @@ -344,6 +344,8 @@ protected class LeaderSelector implements Callable<Void> {

private volatile boolean locked = false;

private volatile boolean yielding = false;

LeaderSelector(String lockKey) {
this.lock = LockRegistryLeaderInitiator.this.locks.obtain(lockKey);
this.lockKey = lockKey;
Expand All @@ -353,11 +355,22 @@ protected class LeaderSelector implements Callable<Void> {
public Void call() {
try {
while (isRunning()) {
if (Thread.currentThread().isInterrupted()) {
// No need to try to lock in the interrupted thread, and we might not be able to unlock
restartSelectorBecauseOfError(new InterruptedException());
return null;
}
if (this.yielding) {
this.yielding = false;
// When yielding, we have to unlock and continue after busyWaitMillis to elect
unlockAndHandleException(null);
continue;
}
try {
tryAcquireLock();
}
catch (Exception e) {
if (handleLockException(e)) {
if (unlockAndHandleException(e)) {
return null;
}
}
Expand All @@ -373,7 +386,7 @@ public Void call() {
LOGGER.debug("Could not unlock during stop for " + this.context
+ " - treat as broken. Revoking...", e);
}
// We are stopping, therefore not leading any more
// We are stopping, therefore not leading anymore
handleRevoked();
}
}
Expand All @@ -385,8 +398,8 @@ private void tryAcquireLock() throws InterruptedException {
LOGGER.debug("Acquiring the lock for " + this.context);
}
// We always try to acquire the lock, in case it expired
boolean acquired = this.lock.tryLock(LockRegistryLeaderInitiator.this.heartBeatMillis,
TimeUnit.MILLISECONDS);
boolean acquired =
this.lock.tryLock(LockRegistryLeaderInitiator.this.heartBeatMillis, TimeUnit.MILLISECONDS);
if (!this.locked) {
if (acquired) {
// Success: we are now leader
Expand All @@ -398,8 +411,7 @@ else if (isPublishFailedEvents()) {
}
}
else if (acquired) {
// If we were able to acquire it but we were already locked we
// should release it
// If we were able to acquire it, but we were already locked, we should release it
this.lock.unlock();
if (isRunning()) {
// Give it a chance to expire.
Expand All @@ -408,7 +420,7 @@ else if (acquired) {
}
else {
this.locked = false;
// We were not able to acquire it, therefore not leading any more
// We were not able to acquire it, therefore not leading anymore
handleRevoked();
if (isRunning()) {
// Try again quickly in case the lock holder dropped it
Expand All @@ -417,7 +429,7 @@ else if (acquired) {
}
}

private boolean handleLockException(Exception ex) { // NOSONAR
private boolean unlockAndHandleException(Exception ex) { // NOSONAR
if (this.locked) {
this.locked = false;
try {
Expand Down Expand Up @@ -446,7 +458,7 @@ private boolean handleLockException(Exception ex) { // NOSONAR
Thread.sleep(LockRegistryLeaderInitiator.this.busyWaitMillis);
}
catch (InterruptedException e1) {
// Ignore interruption and let it to be caught on the next cycle.
// Ignore interruption and let it be caught on the next cycle.
Thread.currentThread().interrupt();
}
}
Expand Down Expand Up @@ -534,9 +546,7 @@ public void yield() {
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("Yielding leadership from " + this);
}
if (LockRegistryLeaderInitiator.this.future != null) {
LockRegistryLeaderInitiator.this.future.cancel(true);
}
LockRegistryLeaderInitiator.this.leaderSelector.yielding = true;
}

@Override
Expand Down