Issue 257: Fixed a race condition in LeaderLatch that cause the recipe to create
two nodes in some edge cases.
diff --git a/CHANGES.txt b/CHANGES.txt
index b3386cc..de9f3a4 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,3 +1,8 @@
+1.3.4 - xxxxxxxxxxxxx
+=====================
+* Issue 257: Fixed a race condition in LeaderLatch that cause the recipe to create
+two nodes in some edge cases.
+
1.3.3 - March 6, 2013
=====================
* Issue 250: Restore support for passing null to usingWatcher().
diff --git a/curator-recipes/src/main/java/com/netflix/curator/framework/recipes/leader/LeaderLatch.java b/curator-recipes/src/main/java/com/netflix/curator/framework/recipes/leader/LeaderLatch.java
index ae0f464..13b8851 100644
--- a/curator-recipes/src/main/java/com/netflix/curator/framework/recipes/leader/LeaderLatch.java
+++ b/curator-recipes/src/main/java/com/netflix/curator/framework/recipes/leader/LeaderLatch.java
@@ -16,6 +16,7 @@
package com.netflix.curator.framework.recipes.leader;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.netflix.curator.framework.CuratorFramework;
import com.netflix.curator.framework.api.BackgroundCallback;
@@ -37,6 +38,7 @@
import java.io.IOException;
import java.util.Collection;
import java.util.List;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
@@ -57,6 +59,7 @@
private final String id;
private final AtomicReference<State> state = new AtomicReference<State>(State.LATENT);
private final AtomicBoolean hasLeadership = new AtomicBoolean(false);
+ private final AtomicReference<String> ourPath = new AtomicReference<String>();
private final ConnectionStateListener listener = new ConnectionStateListener()
{
@@ -67,8 +70,6 @@
}
};
- private volatile String ourPath = null;
-
private static final String LOCK_NAME = "latch-";
private static final LockInternalsSorter sorter = new LockInternalsSorter()
@@ -135,7 +136,7 @@
try
{
- deleteNode();
+ setNode(null);
}
catch ( Exception e )
{
@@ -308,19 +309,29 @@
return (state.get() == State.STARTED) && hasLeadership.get();
}
- private void reset() throws Exception
+ @VisibleForTesting
+ volatile CountDownLatch debugResetWaitLatch = null;
+
+ @VisibleForTesting
+ void reset() throws Exception
{
setLeadership(false);
- deleteNode();
+ setNode(null);
BackgroundCallback callback = new BackgroundCallback()
{
@Override
public void processResult(CuratorFramework client, CuratorEvent event) throws Exception
{
+ if ( debugResetWaitLatch != null )
+ {
+ debugResetWaitLatch.await();
+ debugResetWaitLatch = null;
+ }
+
if ( event.getResultCode() == KeeperException.Code.OK.intValue() )
{
- ourPath = event.getName();
+ setNode(event.getName());
getChildren();
}
else
@@ -334,8 +345,9 @@
private void checkLeadership(List<String> children) throws Exception
{
+ final String localOurPath = ourPath.get();
List<String> sortedChildren = LockInternals.getSortedChildren(LOCK_NAME, sorter, children);
- int ourIndex = (ourPath != null) ? sortedChildren.indexOf(ZKPaths.getNodeFromPath(ourPath)) : -1;
+ int ourIndex = (localOurPath != null) ? sortedChildren.indexOf(ZKPaths.getNodeFromPath(localOurPath)) : -1;
if ( ourIndex < 0 )
{
log.error("Can't find our node. Resetting. Index: " + ourIndex);
@@ -353,7 +365,7 @@
@Override
public void process(WatchedEvent event)
{
- if ( (state.get() == State.STARTED) && (event.getType() == Event.EventType.NodeDeleted) && (ourPath != null) )
+ if ( (state.get() == State.STARTED) && (event.getType() == Event.EventType.NodeDeleted) && (localOurPath != null) )
{
try
{
@@ -438,19 +450,12 @@
notifyAll();
}
- private void deleteNode() throws Exception
+ private void setNode(String newValue) throws Exception
{
- String localPath = ourPath;
- if ( localPath != null )
+ String oldPath = ourPath.getAndSet(newValue);
+ if ( oldPath != null )
{
- try
- {
- client.delete().guaranteed().inBackground().forPath(localPath);
- }
- finally
- {
- ourPath = null;
- }
+ client.delete().guaranteed().inBackground().forPath(oldPath);
}
}
}
diff --git a/curator-recipes/src/test/java/com/netflix/curator/framework/recipes/leader/TestLeaderLatch.java b/curator-recipes/src/test/java/com/netflix/curator/framework/recipes/leader/TestLeaderLatch.java
index 90f2837..b292948 100644
--- a/curator-recipes/src/test/java/com/netflix/curator/framework/recipes/leader/TestLeaderLatch.java
+++ b/curator-recipes/src/test/java/com/netflix/curator/framework/recipes/leader/TestLeaderLatch.java
@@ -43,6 +43,36 @@
private static final int MAX_LOOPS = 5;
@Test
+ public void testResetRace() throws Exception
+ {
+ Timing timing = new Timing();
+ LeaderLatch latch = null;
+ CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), timing.session(), timing.connection(), new RetryOneTime(1));
+ try
+ {
+ client.start();
+ latch = new LeaderLatch(client, PATH_NAME);
+
+ latch.debugResetWaitLatch = new CountDownLatch(1);
+ latch.start(); // will call reset()
+ latch.reset(); // should not result in two nodes
+
+ timing.sleepABit();
+
+ latch.debugResetWaitLatch.countDown();
+
+ timing.sleepABit();
+
+ Assert.assertEquals(client.getChildren().forPath(PATH_NAME).size(), 1);
+ }
+ finally
+ {
+ Closeables.closeQuietly(latch);
+ Closeables.closeQuietly(client);
+ }
+ }
+
+ @Test
public void testLostConnection() throws Exception
{
final int PARTICIPANT_QTY = 10;
