[curator] branch master updated: CURATOR-671: Make sure success LeaderSelector::requeue after observing hasLeadership and then !hasLeadership (#462)

[ke...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

kezhuw pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/curator.git


The following commit(s) were added to refs/heads/master by this push:
     new 328f9853 CURATOR-671: Make sure success LeaderSelector::requeue after observing hasLeadership and then !hasLeadership (#462)
328f9853 is described below

commit 328f985304b07cec046e4b9e4244a9c962fc91f5
Author: Kezhu Wang <ke...@apache.org>
AuthorDate: Wed May 31 18:16:59 2023 +0800

    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.
---
 .../framework/recipes/leader/LeaderSelector.java   | 31 +++++++++++++---------
 1 file changed, 18 insertions(+), 13 deletions(-)

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 854385aa..afc3a8c7 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 @@ public class LeaderSelector implements Closeable
                 @Override
                 public Void call() throws Exception
                 {
-                    try
-                    {
-                        taskStarted();
-                        doWorkLoop();
-                    }
-                    finally
-                    {
-                        taskDone();
-                    }
+                    doWorkLoop();
                     return null;
                 }
             });
@@ -377,12 +369,25 @@ public class LeaderSelector implements Closeable
         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 @@ public class LeaderSelector implements Closeable
     @VisibleForTesting
     void doWork() throws Exception
     {
+        taskStarted();
         hasLeadership = false;
         try
         {
@@ -468,10 +474,9 @@ public class LeaderSelector implements Closeable
         }
         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();