curator-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From randg...@apache.org
Subject git commit: CURATOR-73 - add additional clearIsQueued() for safety
Date Sun, 23 Feb 2014 10:06:23 GMT
Repository: curator
Updated Branches:
  refs/heads/master 9f108285d -> a26872775


CURATOR-73 - add additional clearIsQueued() for safety


Project: http://git-wip-us.apache.org/repos/asf/curator/repo
Commit: http://git-wip-us.apache.org/repos/asf/curator/commit/a2687277
Tree: http://git-wip-us.apache.org/repos/asf/curator/tree/a2687277
Diff: http://git-wip-us.apache.org/repos/asf/curator/diff/a2687277

Branch: refs/heads/master
Commit: a26872775f4f86d7c575b48623ca47e1115e57ea
Parents: 9f10828
Author: randgalt <randgalt@apache.org>
Authored: Sun Feb 23 15:35:50 2014 +0530
Committer: randgalt <randgalt@apache.org>
Committed: Sun Feb 23 15:35:50 2014 +0530

----------------------------------------------------------------------
 .../recipes/leader/LeaderSelector.java          | 33 +++++++-----
 .../recipes/leader/TestLeaderSelector.java      | 57 ++++++++++++++------
 2 files changed, 59 insertions(+), 31 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/curator/blob/a2687277/curator-recipes/src/main/java/org/apache/curator/framework/recipes/leader/LeaderSelector.java
----------------------------------------------------------------------
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 6a15f23..ac10733 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
@@ -117,20 +117,21 @@ public class LeaderSelector implements Closeable
     }
 
     /**
-     * @param client the client
-     * @param leaderPath the path for this leadership group
+     * @param client          the client
+     * @param leaderPath      the path for this leadership group
      * @param executorService thread pool to use
-     * @param listener listener
+     * @param listener        listener
      */
-    public LeaderSelector(CuratorFramework client, String leaderPath, ExecutorService executorService,
LeaderSelectorListener listener) {
+    public LeaderSelector(CuratorFramework client, String leaderPath, ExecutorService executorService,
LeaderSelectorListener listener)
+    {
         this(client, leaderPath, new CloseableExecutorService(executorService), listener);
     }
 
     /**
-     * @param client the client
-     * @param leaderPath the path for this leadership group
+     * @param client          the client
+     * @param leaderPath      the path for this leadership group
      * @param executorService thread pool to use
-     * @param listener listener
+     * @param listener        listener
      */
     public LeaderSelector(CuratorFramework client, String leaderPath, CloseableExecutorService
executorService, LeaderSelectorListener listener)
     {
@@ -230,18 +231,22 @@ public class LeaderSelector implements Closeable
         if ( !isQueued )
         {
             isQueued = true;
-            Future<Void> task = executorService.submit
-            (
-                new Callable<Void>()
+            Future<Void> task = executorService.submit(new Callable<Void>()
+            {
+                @Override
+                public Void call() throws Exception
                 {
-                    @Override
-                    public Void call() throws Exception
+                    try
                     {
                         doWorkLoop();
-                        return null;
                     }
+                    finally
+                    {
+                        clearIsQueued();
+                    }
+                    return null;
                 }
-            );
+            });
             ourTask.set(task);
 
             return true;

http://git-wip-us.apache.org/repos/asf/curator/blob/a2687277/curator-recipes/src/test/java/org/apache/curator/framework/recipes/leader/TestLeaderSelector.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/leader/TestLeaderSelector.java
b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/leader/TestLeaderSelector.java
index bfc8c97..1ae041b 100644
--- a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/leader/TestLeaderSelector.java
+++ b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/leader/TestLeaderSelector.java
@@ -20,7 +20,6 @@
 package org.apache.curator.framework.recipes.leader;
 
 import com.google.common.collect.Lists;
-import org.apache.curator.utils.CloseableUtils;
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.curator.framework.CuratorFrameworkFactory;
 import org.apache.curator.framework.recipes.BaseClassForTests;
@@ -29,6 +28,7 @@ import org.apache.curator.retry.RetryOneTime;
 import org.apache.curator.test.KillSession;
 import org.apache.curator.test.TestingServer;
 import org.apache.curator.test.Timing;
+import org.apache.curator.utils.CloseableUtils;
 import org.testng.Assert;
 import org.testng.annotations.Test;
 import org.testng.internal.annotations.Sets;
@@ -41,6 +41,8 @@ import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import static org.testng.Assert.fail;
+
 public class TestLeaderSelector extends BaseClassForTests
 {
     private static final String PATH_NAME = "/one/two/me";
@@ -275,7 +277,7 @@ public class TestLeaderSelector extends BaseClassForTests
     }
 
     @Test
-    public void testKillSession() throws Exception
+    public void testKillSessionThenCloseShouldElectNewLeader() throws Exception
     {
         final Timing timing = new Timing();
 
@@ -286,7 +288,7 @@ public class TestLeaderSelector extends BaseClassForTests
             final Semaphore semaphore = new Semaphore(0);
             final CountDownLatch interruptedLatch = new CountDownLatch(1);
             final AtomicInteger leaderCount = new AtomicInteger(0);
-            LeaderSelectorListener listener = new LeaderSelectorListener()
+            LeaderSelectorListener listener = new LeaderSelectorListenerAdapter()
             {
                 @Override
                 public void takeLeadership(CuratorFramework client) throws Exception
@@ -297,7 +299,7 @@ public class TestLeaderSelector extends BaseClassForTests
                         semaphore.release();
                         try
                         {
-                            Thread.sleep(1000000);
+                            Thread.currentThread().join();
                         }
                         catch ( InterruptedException e )
                         {
@@ -310,19 +312,13 @@ public class TestLeaderSelector extends BaseClassForTests
                         leaderCount.decrementAndGet();
                     }
                 }
-
-                @Override
-                public void stateChanged(CuratorFramework client, ConnectionState newState)
-                {
-                    if ( newState == ConnectionState.LOST )
-                    {
-                        throw new CancelLeadershipException();
-                    }
-                }
             };
             LeaderSelector leaderSelector1 = new LeaderSelector(client, PATH_NAME, listener);
             LeaderSelector leaderSelector2 = new LeaderSelector(client, PATH_NAME, listener);
 
+            boolean leaderSelector1Closed = false;
+            boolean leaderSelector2Closed = false;
+
             leaderSelector1.start();
             leaderSelector2.start();
 
@@ -333,14 +329,41 @@ public class TestLeaderSelector extends BaseClassForTests
             Assert.assertTrue(timing.awaitLatch(interruptedLatch));
             timing.sleepABit();
 
-            leaderSelector1.requeue();
-            leaderSelector2.requeue();
+            boolean requeued1 = leaderSelector1.requeue();
+            boolean requeued2 = leaderSelector2.requeue();
+            Assert.assertTrue(requeued1);
+            Assert.assertTrue(requeued2);
 
             Assert.assertTrue(timing.acquireSemaphore(semaphore, 1));
             Assert.assertEquals(leaderCount.get(), 1);
 
-            leaderSelector1.close();
-            leaderSelector2.close();
+            if ( leaderSelector1.hasLeadership() )
+            {
+                leaderSelector1.close();
+                leaderSelector1Closed = true;
+            }
+            else if ( leaderSelector2.hasLeadership() )
+            {
+                leaderSelector2.close();
+                leaderSelector2Closed = true;
+            }
+            else
+            {
+                fail("No leaderselector has leadership!");
+            }
+
+            // Verify that the other leader took over leadership.
+            Assert.assertTrue(timing.acquireSemaphore(semaphore, 1));
+            Assert.assertEquals(leaderCount.get(), 1);
+
+            if ( !leaderSelector1Closed )
+            {
+                leaderSelector1.close();
+            }
+            if ( !leaderSelector2Closed )
+            {
+                leaderSelector2.close();
+            }
         }
         finally
         {


Mime
View raw message