You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@curator.apache.org by ra...@apache.org on 2014/02/23 11:06:07 UTC
git commit: CURATOR-73 - add additional clearIsQueued() for safety
Repository: curator
Updated Branches:
refs/heads/CURATOR-73 [created] 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/CURATOR-73
Commit: a26872775f4f86d7c575b48623ca47e1115e57ea
Parents: 9f10828
Author: randgalt <ra...@apache.org>
Authored: Sun Feb 23 15:35:50 2014 +0530
Committer: randgalt <ra...@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
{