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 2015/10/11 22:51:00 UTC
[1/3] curator git commit: continue watching the node until it is
deleted or the recipe is stopped
Repository: curator
Updated Branches:
refs/heads/master f4f220824 -> 8b08778bc
continue watching the node until it is deleted or the recipe is stopped
Project: http://git-wip-us.apache.org/repos/asf/curator/repo
Commit: http://git-wip-us.apache.org/repos/asf/curator/commit/cdb3a803
Tree: http://git-wip-us.apache.org/repos/asf/curator/tree/cdb3a803
Diff: http://git-wip-us.apache.org/repos/asf/curator/diff/cdb3a803
Branch: refs/heads/master
Commit: cdb3a8035a551e3fb23e972a19cbeae47b14c27f
Parents: 970bd39
Author: Njal Karevoll <nj...@karevoll.no>
Authored: Wed Sep 2 20:03:28 2015 +0200
Committer: Njal Karevoll <nj...@karevoll.no>
Committed: Wed Sep 2 20:03:28 2015 +0200
----------------------------------------------------------------------
.../recipes/nodes/PersistentEphemeralNode.java | 18 +++--
.../nodes/TestPersistentEphemeralNode.java | 82 ++++++++++++++++++++
2 files changed, 95 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/curator/blob/cdb3a803/curator-recipes/src/main/java/org/apache/curator/framework/recipes/nodes/PersistentEphemeralNode.java
----------------------------------------------------------------------
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 7e00e10..4868c7a 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
@@ -74,10 +74,18 @@ public class PersistentEphemeralNode implements Closeable
@Override
public void process(WatchedEvent event)
{
- if ( event.getType() == EventType.NodeDeleted)
- {
- createNode();
- }
+ if ( event.getType() == EventType.NodeDeleted)
+ {
+ createNode();
+ }
+ else if ( event.getType() == EventType.NodeDataChanged)
+ {
+ try {
+ watchNode();
+ } catch (Exception e) {
+ log.error(String.format("Unexpected error during watching of path: %s", basePath), e);
+ }
+ }
}
};
private final BackgroundCallback checkExistsCallback = new BackgroundCallback()
@@ -426,4 +434,4 @@ public class PersistentEphemeralNode implements Closeable
{
return authFailure.get();
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/curator/blob/cdb3a803/curator-recipes/src/test/java/org/apache/curator/framework/recipes/nodes/TestPersistentEphemeralNode.java
----------------------------------------------------------------------
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 8d5d05f..84eaa52 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
@@ -391,6 +391,46 @@ public class TestPersistentEphemeralNode extends BaseClassForTests
}
@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();
@@ -419,6 +459,43 @@ public class TestPersistentEphemeralNode extends BaseClassForTests
}
@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();
@@ -721,5 +798,10 @@ public class TestPersistentEphemeralNode extends BaseClassForTests
{
return new Trigger(Event.EventType.NodeDeleted);
}
+
+ private static Trigger dataChanged()
+ {
+ return new Trigger(EventType.NodeDataChanged);
+ }
}
}
[3/3] curator git commit: reformatting - also use CuratorWatcher
Posted by ra...@apache.org.
reformatting - also use CuratorWatcher
Project: http://git-wip-us.apache.org/repos/asf/curator/repo
Commit: http://git-wip-us.apache.org/repos/asf/curator/commit/8b08778b
Tree: http://git-wip-us.apache.org/repos/asf/curator/tree/8b08778b
Diff: http://git-wip-us.apache.org/repos/asf/curator/diff/8b08778b
Branch: refs/heads/master
Commit: 8b08778bc6c6b25145188e8df72a01ffa2f9924c
Parents: 02a03f9
Author: randgalt <ra...@apache.org>
Authored: Sun Oct 11 15:50:40 2015 -0500
Committer: randgalt <ra...@apache.org>
Committed: Sun Oct 11 15:50:40 2015 -0500
----------------------------------------------------------------------
.../recipes/nodes/PersistentEphemeralNode.java | 97 +++++++++-----------
1 file changed, 41 insertions(+), 56 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/curator/blob/8b08778b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/nodes/PersistentEphemeralNode.java
----------------------------------------------------------------------
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 4868c7a..a3298ab 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,22 +21,21 @@ package org.apache.curator.framework.recipes.nodes;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
-
import org.apache.curator.framework.CuratorFramework;
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;
@@ -45,8 +44,6 @@ import java.util.concurrent.TimeUnit;
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
@@ -69,22 +66,18 @@ public class PersistentEphemeralNode implements Closeable
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)
+ if ( event.getType() == EventType.NodeDeleted )
{
createNode();
}
- else if ( event.getType() == EventType.NodeDataChanged)
+ else if ( event.getType() == EventType.NodeDataChanged )
{
- try {
- watchNode();
- } catch (Exception e) {
- log.error(String.format("Unexpected error during watching of path: %s", basePath), e);
- }
+ watchNode();
}
}
};
@@ -99,21 +92,23 @@ public class PersistentEphemeralNode implements Closeable
}
}
};
- 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
@@ -219,7 +214,7 @@ public class PersistentEphemeralNode implements Closeable
* @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)
{
@@ -236,9 +231,9 @@ public class PersistentEphemeralNode implements Closeable
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() )
{
@@ -246,23 +241,23 @@ public class PersistentEphemeralNode implements Closeable
}
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
@@ -275,7 +270,7 @@ public class PersistentEphemeralNode implements Closeable
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);
@@ -360,7 +355,8 @@ public class PersistentEphemeralNode implements Closeable
}
}
- private byte[] getData() {
+ private byte[] getData()
+ {
return this.data.get();
}
@@ -377,10 +373,6 @@ public class PersistentEphemeralNode implements Closeable
{
// ignore
}
- catch ( Exception e )
- {
- throw e;
- }
}
}
@@ -413,14 +405,7 @@ public class PersistentEphemeralNode implements Closeable
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);
}
}
@@ -428,10 +413,10 @@ public class PersistentEphemeralNode implements Closeable
{
return (state.get() == State.STARTED);
}
-
+
@VisibleForTesting
boolean isAuthFailure()
{
- return authFailure.get();
+ return authFailure.get();
}
}
[2/3] curator git commit: Merge branch 'CURATOR-258' of
github.com:nkvoll/curator into CURATOR-258
Posted by ra...@apache.org.
Merge branch 'CURATOR-258' of github.com:nkvoll/curator into CURATOR-258
Project: http://git-wip-us.apache.org/repos/asf/curator/repo
Commit: http://git-wip-us.apache.org/repos/asf/curator/commit/02a03f94
Tree: http://git-wip-us.apache.org/repos/asf/curator/tree/02a03f94
Diff: http://git-wip-us.apache.org/repos/asf/curator/diff/02a03f94
Branch: refs/heads/master
Commit: 02a03f943cc1a68ac6b1450b0381fcfd86c7b392
Parents: f4f2208 cdb3a80
Author: randgalt <ra...@apache.org>
Authored: Sun Oct 11 15:46:13 2015 -0500
Committer: randgalt <ra...@apache.org>
Committed: Sun Oct 11 15:46:13 2015 -0500
----------------------------------------------------------------------
.../recipes/nodes/PersistentEphemeralNode.java | 18 +++--
.../nodes/TestPersistentEphemeralNode.java | 82 ++++++++++++++++++++
2 files changed, 95 insertions(+), 5 deletions(-)
----------------------------------------------------------------------