Merge branch 'master' into CURATOR-3.0
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/nodes/PersistentEphemeralNode.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/nodes/PersistentEphemeralNode.java
index 1606c36..fe42738 100644
--- a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/nodes/PersistentEphemeralNode.java
+++ b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/nodes/PersistentEphemeralNode.java
@@ -21,23 +21,22 @@
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
-
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.WatcherRemoveCuratorFramework;
import org.apache.curator.framework.api.ACLBackgroundPathAndBytesable;
import org.apache.curator.framework.api.BackgroundCallback;
import org.apache.curator.framework.api.CreateModable;
import org.apache.curator.framework.api.CuratorEvent;
+import org.apache.curator.framework.api.CuratorWatcher;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
+import org.apache.curator.utils.PathUtils;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
-import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-
import java.io.Closeable;
import java.io.IOException;
import java.util.Arrays;
@@ -46,8 +45,6 @@
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
-import org.apache.curator.utils.PathUtils;
-
/**
* <p>
* A persistent ephemeral node is an ephemeral node that attempts to stay present in
@@ -70,15 +67,19 @@
private final AtomicReference<State> state = new AtomicReference<State>(State.LATENT);
private final AtomicBoolean authFailure = new AtomicBoolean(false);
private final BackgroundCallback backgroundCallback;
- private final Watcher watcher = new Watcher()
+ private final CuratorWatcher watcher = new CuratorWatcher()
{
@Override
- public void process(WatchedEvent event)
+ public void process(WatchedEvent event) throws Exception
{
- if ( event.getType() == EventType.NodeDeleted)
- {
- createNode();
- }
+ if ( event.getType() == EventType.NodeDeleted )
+ {
+ createNode();
+ }
+ else if ( event.getType() == EventType.NodeDataChanged )
+ {
+ watchNode();
+ }
}
};
private final BackgroundCallback checkExistsCallback = new BackgroundCallback()
@@ -92,21 +93,23 @@
}
}
};
- private final BackgroundCallback setDataCallback = new BackgroundCallback() {
-
- @Override
- public void processResult(CuratorFramework client, CuratorEvent event)
- throws Exception {
- //If the result is ok then initialisation is complete (if we're still initialising)
- //Don't retry on other errors as the only recoverable cases will be connection loss
- //and the node not existing, both of which are already handled by other watches.
- if ( event.getResultCode() == KeeperException.Code.OK.intValue() )
- {
- //Update is ok, mark initialisation as complete if required.
- initialisationComplete();
- }
- }
- };
+ private final BackgroundCallback setDataCallback = new BackgroundCallback()
+ {
+
+ @Override
+ public void processResult(CuratorFramework client, CuratorEvent event)
+ throws Exception
+ {
+ //If the result is ok then initialisation is complete (if we're still initialising)
+ //Don't retry on other errors as the only recoverable cases will be connection loss
+ //and the node not existing, both of which are already handled by other watches.
+ if ( event.getResultCode() == KeeperException.Code.OK.intValue() )
+ {
+ //Update is ok, mark initialisation as complete if required.
+ initialisationComplete();
+ }
+ }
+ };
private final ConnectionStateListener connectionStateListener = new ConnectionStateListener()
{
@Override
@@ -215,7 +218,7 @@
* @param client client instance
* @param mode creation/protection mode
* @param basePath the base path for the node
- * @param data data for the node
+ * @param initData data for the node
*/
public PersistentEphemeralNode(CuratorFramework client, Mode mode, String basePath, byte[] initData)
{
@@ -232,9 +235,9 @@
String path = null;
boolean nodeExists = false;
if ( event.getResultCode() == KeeperException.Code.NODEEXISTS.intValue() )
- {
- path = event.getPath();
- nodeExists = true;
+ {
+ path = event.getPath();
+ nodeExists = true;
}
else if ( event.getResultCode() == KeeperException.Code.OK.intValue() )
{
@@ -242,23 +245,23 @@
}
else if ( event.getResultCode() == KeeperException.Code.NOAUTH.intValue() )
{
- log.warn("Client does not have authorisation to write ephemeral node at path {}", path);
- authFailure.set(true);
- return;
+ log.warn("Client does not have authorisation to write ephemeral node at path {}", event.getPath());
+ authFailure.set(true);
+ return;
}
if ( path != null )
{
- authFailure.set(false);
+ authFailure.set(false);
nodePath.set(path);
watchNode();
- if(nodeExists)
+ if ( nodeExists )
{
- client.setData().inBackground(setDataCallback).forPath(getActualPath(), getData());
+ client.setData().inBackground(setDataCallback).forPath(getActualPath(), getData());
}
else
{
- initialisationComplete();
+ initialisationComplete();
}
}
else
@@ -271,7 +274,7 @@
createMethod = mode.isProtected() ? client.create().creatingParentContainersIfNeeded().withProtection() : client.create().creatingParentContainersIfNeeded();
this.data.set(Arrays.copyOf(data, data.length));
}
-
+
private void initialisationComplete()
{
CountDownLatch localLatch = initialCreateLatch.getAndSet(null);
@@ -358,7 +361,8 @@
}
}
- private byte[] getData() {
+ private byte[] getData()
+ {
return this.data.get();
}
@@ -375,10 +379,6 @@
{
// ignore
}
- catch ( Exception e )
- {
- throw e;
- }
}
}
@@ -424,14 +424,7 @@
String localNodePath = nodePath.get();
if ( localNodePath != null )
{
- try
- {
- client.checkExists().usingWatcher(watcher).inBackground(checkExistsCallback).forPath(localNodePath);
- }
- catch ( Exception e )
- {
- throw e;
- }
+ client.checkExists().usingWatcher(watcher).inBackground(checkExistsCallback).forPath(localNodePath);
}
}
@@ -439,10 +432,10 @@
{
return (state.get() == State.STARTED);
}
-
+
@VisibleForTesting
boolean isAuthFailure()
{
- return authFailure.get();
+ return authFailure.get();
}
-}
\ No newline at end of file
+}
diff --git a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/nodes/TestPersistentEphemeralNode.java b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/nodes/TestPersistentEphemeralNode.java
index 20cc028..7ee768b 100644
--- a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/nodes/TestPersistentEphemeralNode.java
+++ b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/nodes/TestPersistentEphemeralNode.java
@@ -409,6 +409,46 @@
}
@Test
+ public void testRecreatesNodeWhenEphemeralOwnerSessionExpires() throws Exception
+ {
+ CuratorFramework curator = newCurator();
+ CuratorFramework nodeCreator = newCurator();
+ CuratorFramework observer = newCurator();
+
+ nodeCreator.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(PATH, new byte[0]);
+
+ Trigger dataChangedTrigger = Trigger.dataChanged();
+ observer.getData().usingWatcher(dataChangedTrigger).forPath(PATH);
+
+ PersistentEphemeralNode node = new PersistentEphemeralNode(curator, PersistentEphemeralNode.Mode.EPHEMERAL, PATH, new byte[0]);
+ node.start();
+ try
+ {
+ node.waitForInitialCreate(5, TimeUnit.SECONDS);
+ assertNodeExists(observer, node.getActualPath());
+
+ assertTrue(dataChangedTrigger.firedWithin(timing.forWaiting().seconds(), TimeUnit.SECONDS));
+
+ Trigger deletedTrigger = Trigger.deleted();
+ observer.checkExists().usingWatcher(deletedTrigger).forPath(node.getActualPath());
+
+ killSession(nodeCreator);
+
+ // Make sure the node got deleted...
+ assertTrue(deletedTrigger.firedWithin(timing.forWaiting().seconds(), TimeUnit.SECONDS));
+
+ // Check for it to be recreated...
+ Trigger createdTrigger = Trigger.created();
+ Stat stat = observer.checkExists().usingWatcher(createdTrigger).forPath(node.getActualPath());
+ assertTrue(stat != null || createdTrigger.firedWithin(timing.forWaiting().seconds(), TimeUnit.SECONDS));
+ }
+ finally
+ {
+ node.close();
+ }
+ }
+
+ @Test
public void testRecreatesNodeWhenItGetsDeleted() throws Exception
{
CuratorFramework curator = newCurator();
@@ -437,6 +477,43 @@
}
@Test
+ public void testRecreatesNodeWhenItGetsDeletedAfterSetData() throws Exception
+ {
+ CuratorFramework curator = newCurator();
+
+ PersistentEphemeralNode node = new PersistentEphemeralNode(curator, PersistentEphemeralNode.Mode.EPHEMERAL, PATH, new byte[0]);
+ node.start();
+ try
+ {
+ node.waitForInitialCreate(timing.forWaiting().seconds(), TimeUnit.SECONDS);
+ String originalNode = node.getActualPath();
+ assertNodeExists(curator, originalNode);
+
+ Trigger dataChangedTrigger = Trigger.dataChanged();
+ curator.getData().usingWatcher(dataChangedTrigger).forPath(originalNode);
+
+ // update the data of the node
+ node.setData(new byte[0]);
+
+ // wait for the data to be updated:
+ assertTrue(dataChangedTrigger.firedWithin(timing.forWaiting().seconds(), TimeUnit.SECONDS));
+
+ // Delete the original node...
+ curator.delete().forPath(originalNode);
+
+ // Since we're using an ephemeral node, and the original session hasn't been interrupted the name of the new
+ // node that gets created is going to be exactly the same as the original.
+ Trigger createdWatchTrigger = Trigger.created();
+ Stat stat = curator.checkExists().usingWatcher(createdWatchTrigger).forPath(originalNode);
+ assertTrue(stat != null || createdWatchTrigger.firedWithin(timing.forWaiting().seconds(), TimeUnit.SECONDS));
+ }
+ finally
+ {
+ node.close();
+ }
+ }
+
+ @Test
public void testNodesCreateUniquePaths() throws Exception
{
CuratorFramework curator = newCurator();
@@ -737,5 +814,10 @@
{
return new Trigger(Event.EventType.NodeDeleted, EventType.NodeDataChanged);
}
+
+ private static Trigger dataChanged()
+ {
+ return new Trigger(EventType.NodeDataChanged);
+ }
}
}
