CURATOR-671: Make sure success LeaderSelector::requeue after observing hasLeadership and then !hasLeadership (#462)
`TestLeaderSelectorCluster.testLostRestart` depends on success
`LeaderSelector::requeue` after observing `hasLeadership` and then
`!hasLeadership`. #446(CURATOR-518) breaks this.
Instead of simply looping till success `LeaderSelector::requeue`, I
tend to revert to old behavior as it was that since its introduce in
eb4d2aaf7702e7e5ffbf1c1dd8544e88b1171045.
Though, we can't depend this in case of no leadership was ever granted.
In this case, when session expired, I guess looping on `requeue` may be
the only option. Anyway, `requeue` is somewhat hard to use correctly.
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/leader/LeaderSelector.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/leader/LeaderSelector.java
index 854385a..afc3a8c 100644
--- a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/leader/LeaderSelector.java
+++ b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/leader/LeaderSelector.java
@@ -245,15 +245,7 @@
@Override
public Void call() throws Exception
{
- try
- {
- taskStarted();
- doWorkLoop();
- }
- finally
- {
- taskDone();
- }
+ doWorkLoop();
return null;
}
});
@@ -377,12 +369,25 @@
ourThread = Thread.currentThread();
}
- private synchronized void taskDone() {
+ private synchronized boolean taskDone() {
ourTask = null;
ourThread = null;
+ // We are about to complete the very last steps in election task, there is
+ // no synchronization point after this method. Safety:
+ // * Next task will not run into election body before this method return, so no
+ // interference on hasLeadership.
+ // * mutex is bound to current thread(e.g. not JVM), so leadership flag reset
+ // and leadership release, which is a time-consuming task, are not necessary
+ // to be atomic. Also, it is safe to run mutex.release() in parallel with
+ // mutex.acquire() from next election task.
+ boolean leadership = hasLeadership;
+ if (leadership) {
+ hasLeadership = false;
+ }
if (autoRequeue.get()) {
internalRequeue();
}
+ return leadership;
}
/**
@@ -433,6 +438,7 @@
@VisibleForTesting
void doWork() throws Exception
{
+ taskStarted();
hasLeadership = false;
try
{
@@ -468,10 +474,9 @@
}
finally
{
- if ( hasLeadership )
+ if ( taskDone() )
{
- hasLeadership = false;
- boolean wasInterrupted = Thread.interrupted(); // clear any interrupted tatus so that mutex.release() works immediately
+ boolean wasInterrupted = Thread.interrupted(); // clear any interrupted status so that mutex.release() works immediately
try
{
mutex.release();
