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/09/28 00:00:09 UTC
curator git commit: First pass implementation of group membership
Repository: curator
Updated Branches:
refs/heads/CURATOR-267 [created] 3029856c8
First pass implementation of group membership
Project: http://git-wip-us.apache.org/repos/asf/curator/repo
Commit: http://git-wip-us.apache.org/repos/asf/curator/commit/3029856c
Tree: http://git-wip-us.apache.org/repos/asf/curator/tree/3029856c
Diff: http://git-wip-us.apache.org/repos/asf/curator/diff/3029856c
Branch: refs/heads/CURATOR-267
Commit: 3029856c894a72ed2c5762e0c6acd2c0a8cd3937
Parents: f8f05be
Author: randgalt <ra...@apache.org>
Authored: Sun Sep 27 16:59:48 2015 -0500
Committer: randgalt <ra...@apache.org>
Committed: Sun Sep 27 16:59:48 2015 -0500
----------------------------------------------------------------------
.../framework/CuratorFrameworkFactory.java | 31 ++--
.../framework/recipes/nodes/GroupMember.java | 140 +++++++++++++++++++
.../src/site/confluence/group-member.confluence | 42 ++++++
.../src/site/confluence/index.confluence | 3 +-
.../recipes/nodes/TestGroupMember.java | 79 +++++++++++
5 files changed, 281 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/curator/blob/3029856c/curator-framework/src/main/java/org/apache/curator/framework/CuratorFrameworkFactory.java
----------------------------------------------------------------------
diff --git a/curator-framework/src/main/java/org/apache/curator/framework/CuratorFrameworkFactory.java b/curator-framework/src/main/java/org/apache/curator/framework/CuratorFrameworkFactory.java
index dcb2ee6..41ff9cd 100644
--- a/curator-framework/src/main/java/org/apache/curator/framework/CuratorFrameworkFactory.java
+++ b/curator-framework/src/main/java/org/apache/curator/framework/CuratorFrameworkFactory.java
@@ -100,6 +100,24 @@ public class CuratorFrameworkFactory
build();
}
+ /**
+ * Return the local address as bytes that can be used as a node payload
+ *
+ * @return local address bytes
+ */
+ public static byte[] getLocalAddress()
+ {
+ try
+ {
+ return InetAddress.getLocalHost().getHostAddress().getBytes();
+ }
+ catch ( UnknownHostException ignore )
+ {
+ // ignore
+ }
+ return new byte[0];
+ }
+
public static class Builder
{
private EnsembleProvider ensembleProvider;
@@ -465,19 +483,6 @@ public class CuratorFrameworkFactory
}
}
- private static byte[] getLocalAddress()
- {
- try
- {
- return InetAddress.getLocalHost().getHostAddress().getBytes();
- }
- catch ( UnknownHostException ignore )
- {
- // ignore
- }
- return new byte[0];
- }
-
private CuratorFrameworkFactory()
{
}
http://git-wip-us.apache.org/repos/asf/curator/blob/3029856c/curator-recipes/src/main/java/org/apache/curator/framework/recipes/nodes/GroupMember.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/nodes/GroupMember.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/nodes/GroupMember.java
new file mode 100644
index 0000000..5aa8ca2
--- /dev/null
+++ b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/nodes/GroupMember.java
@@ -0,0 +1,140 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.curator.framework.recipes.nodes;
+
+import com.google.common.base.Preconditions;
+import com.google.common.base.Throwables;
+import com.google.common.collect.ImmutableMap;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.framework.recipes.cache.ChildData;
+import org.apache.curator.framework.recipes.cache.PathChildrenCache;
+import org.apache.curator.utils.CloseableUtils;
+import org.apache.curator.utils.ZKPaths;
+import java.io.Closeable;
+import java.util.Arrays;
+import java.util.Map;
+
+/**
+ * Group membership management. Adds this instance into a group and
+ * keeps a cache of members in the group
+ */
+public class GroupMember implements Closeable
+{
+ private final PersistentEphemeralNode pen;
+ private final PathChildrenCache cache;
+ private final String thisId;
+ private final byte[] payload;
+
+ /**
+ * @param client client
+ * @param membershipPath the path to use for membership
+ * @param thisId ID of this group member. MUST be unique for the group
+ */
+ public GroupMember(CuratorFramework client, String membershipPath, String thisId)
+ {
+ this(client, membershipPath, thisId, CuratorFrameworkFactory.getLocalAddress());
+ }
+
+ /**
+ * @param client client
+ * @param membershipPath the path to use for membership
+ * @param thisId ID of this group member. MUST be unique for the group
+ * @param payload the payload to write in our member node
+ */
+ public GroupMember(CuratorFramework client, String membershipPath, String thisId, byte[] payload)
+ {
+ this.thisId = Preconditions.checkNotNull(thisId, "thisId cannot be null");
+ this.payload = Arrays.copyOf(payload, payload.length);
+
+ cache = newPathChildrenCache(client, membershipPath);
+ pen = newPersistentEphemeralNode(client, membershipPath, thisId, payload);
+ }
+
+ /**
+ * Start the group membership. Register thisId as a member and begin
+ * caching all members
+ */
+ public void start()
+ {
+ pen.start();
+ try
+ {
+ cache.start();
+ }
+ catch ( Exception e )
+ {
+ Throwables.propagate(e);
+ }
+ }
+
+ /**
+ * Have thisId leave the group and stop caching membership
+ */
+ @Override
+ public void close()
+ {
+ CloseableUtils.closeQuietly(cache);
+ CloseableUtils.closeQuietly(pen);
+ }
+
+ /**
+ * Return the current view of membership. The keys are the IDs
+ * of the members. The values are each member's payload
+ *
+ * @return membership
+ */
+ public Map<String, byte[]> getCurrentMembers()
+ {
+ ImmutableMap.Builder<String, byte[]> builder = ImmutableMap.builder();
+ boolean thisIdAdded = false;
+ for ( ChildData data : cache.getCurrentData() )
+ {
+ String id = idFromPath(data.getPath());
+ thisIdAdded = thisIdAdded || id.equals(thisId);
+ builder.put(id, data.getData());
+ }
+ if ( !thisIdAdded )
+ {
+ builder.put(thisId, payload); // this instance is always a member
+ }
+ return builder.build();
+ }
+
+ /**
+ * Given a full ZNode path, return the member ID
+ *
+ * @param path full ZNode path
+ * @return id
+ */
+ public String idFromPath(String path)
+ {
+ return ZKPaths.getNodeFromPath(path);
+ }
+
+ protected PersistentEphemeralNode newPersistentEphemeralNode(CuratorFramework client, String membershipPath, String thisId, byte[] payload)
+ {
+ return new PersistentEphemeralNode(client, PersistentEphemeralNode.Mode.EPHEMERAL, ZKPaths.makePath(membershipPath, thisId), payload);
+ }
+
+ protected PathChildrenCache newPathChildrenCache(CuratorFramework client, String membershipPath)
+ {
+ return new PathChildrenCache(client, membershipPath, true);
+ }
+}
http://git-wip-us.apache.org/repos/asf/curator/blob/3029856c/curator-recipes/src/site/confluence/group-member.confluence
----------------------------------------------------------------------
diff --git a/curator-recipes/src/site/confluence/group-member.confluence b/curator-recipes/src/site/confluence/group-member.confluence
new file mode 100644
index 0000000..dcb27b3
--- /dev/null
+++ b/curator-recipes/src/site/confluence/group-member.confluence
@@ -0,0 +1,42 @@
+h1. Group Member
+
+h2. Description
+Group membership management. Adds this instance into a group and keeps a cache of members in the group.
+
+h2. Participating Classes
+* GroupMember
+
+h2. Usage
+h3. Creating a GroupMember
+{code}
+public GroupMember(CuratorFramework client,
+ String membershipPath,
+ String thisId,
+ byte[] payload)
+Parameters:
+client - client instance
+membershipPath - the path to use for membership
+thisId - ID of this group member. MUST be unique for the group
+payload - the payload to write in our member node
+{code}
+
+h3. General Usage
+GroupMember must be started:
+{code}
+group.start();
+{code}
+
+When you are through with the GroupMember instance, you should call close:
+{code}
+group.close();
+{code}
+
+NOTE: this will remove the instance from the group
+
+You can get a current view of the members by calling:
+{code}
+group.getCurrentMembers();
+{code}
+
+h2. Error Handling
+GroupMember instances internally handle all error states recreating the node as necessary.
http://git-wip-us.apache.org/repos/asf/curator/blob/3029856c/curator-recipes/src/site/confluence/index.confluence
----------------------------------------------------------------------
diff --git a/curator-recipes/src/site/confluence/index.confluence b/curator-recipes/src/site/confluence/index.confluence
index 4f3a032..71de8df 100644
--- a/curator-recipes/src/site/confluence/index.confluence
+++ b/curator-recipes/src/site/confluence/index.confluence
@@ -29,7 +29,8 @@ regarding "Curator Recipes Own Their ZNode/Paths".
|[[Tree Cache|tree-cache.html]] \- A utility that attempts to keep all data from all children of a ZK path locally cached. This class will watch the ZK path, respond to update/create/delete events, pull down the data, etc. You can register a listener that will get notified when changes occur.|
||Nodes||
-|[[Persistent Ephemeral Node|persistent-ephemeral-node.html]] \- An ephemeral node that attempts to stay present in ZooKeeper, even through connection and session interruptions..|
+|[[Persistent Ephemeral Node|persistent-ephemeral-node.html]] \- An ephemeral node that attempts to stay present in ZooKeeper, even through connection and session interruptions.|
+|[Group Member|group-member-node.html]] \- Group membership management. Adds this instance into a group and keeps a cache of members in the group.|
||Queues||
|[[Distributed Queue|distributed-queue.html]] \- An implementation of the Distributed Queue ZK recipe. Items put into the queue are guaranteed to be ordered (by means of ZK's PERSISTENT_SEQUENTIAL node). If a single consumer takes items out of the queue, they will be ordered FIFO. If ordering is important, use a LeaderSelector to nominate a single consumer.|
http://git-wip-us.apache.org/repos/asf/curator/blob/3029856c/curator-recipes/src/test/java/org/apache/curator/framework/recipes/nodes/TestGroupMember.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/nodes/TestGroupMember.java b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/nodes/TestGroupMember.java
new file mode 100644
index 0000000..3d4b951
--- /dev/null
+++ b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/nodes/TestGroupMember.java
@@ -0,0 +1,79 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.curator.framework.recipes.nodes;
+
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.retry.RetryOneTime;
+import org.apache.curator.test.BaseClassForTests;
+import org.apache.curator.test.Timing;
+import org.apache.curator.utils.CloseableUtils;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+import java.util.Map;
+
+public class TestGroupMember extends BaseClassForTests
+{
+ // NOTE - don't need many tests as this class is just a wrapper around two existing recipes
+
+ @Test
+ public void testBasic() throws Exception
+ {
+ Timing timing = new Timing();
+ GroupMember groupMember1 = null;
+ GroupMember groupMember2 = null;
+ CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1));
+ try
+ {
+ client.start();
+
+ groupMember1 = new GroupMember(client, "/member", "1");
+ Assert.assertTrue(groupMember1.getCurrentMembers().containsKey("1"));
+ groupMember1.start();
+
+ groupMember2 = new GroupMember(client, "/member", "2");
+ groupMember2.start();
+
+ timing.sleepABit();
+
+ Map<String, byte[]> currentMembers1 = groupMember1.getCurrentMembers();
+ Map<String, byte[]> currentMembers2 = groupMember2.getCurrentMembers();
+ Assert.assertEquals(currentMembers1.size(), 2);
+ Assert.assertEquals(currentMembers2.size(), 2);
+ Assert.assertEquals(currentMembers1, currentMembers2);
+ Assert.assertTrue(currentMembers1.containsKey("1"));
+ Assert.assertTrue(currentMembers1.containsKey("2"));
+
+ groupMember2.close();
+
+ timing.sleepABit();
+
+ currentMembers1 = groupMember1.getCurrentMembers();
+ Assert.assertEquals(currentMembers1.size(), 1);
+ Assert.assertTrue(currentMembers1.containsKey("1"));
+ Assert.assertFalse(currentMembers1.containsKey("2"));
+ }
+ finally
+ {
+ CloseableUtils.closeQuietly(groupMember1);
+ CloseableUtils.closeQuietly(groupMember2);
+ CloseableUtils.closeQuietly(client);
+ }
+ }
+}