Author: kwright
Date: Wed Sep 17 23:41:31 2014
New Revision: 1625856
URL: http://svn.apache.org/r1625856
Log:
Fix for CONNECTORS-1036.
Added:
manifoldcf/trunk/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperEphemeralNodeObject.java
(with props)
manifoldcf/trunk/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperEphemeralNodePool.java
(with props)
Modified:
manifoldcf/trunk/CHANGES.txt
manifoldcf/trunk/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperConnection.java
manifoldcf/trunk/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperLockManager.java
Modified: manifoldcf/trunk/CHANGES.txt
URL:
http://svn.apache.org/viewvc/manifoldcf/trunk/CHANGES.txt?rev=1625856&r1=1625855&r2=1625856&view=diff
==============================================================================
--- manifoldcf/trunk/CHANGES.txt (original)
+++ manifoldcf/trunk/CHANGES.txt Wed Sep 17 23:41:31 2014
@@ -3,6 +3,10 @@ $Id$
======================= 2.0-dev =====================
+CONNECTORS-1036: Zookeeper service handling also has
+ephemeral nodes which need to be tied to sessions.
+(Karl Wright)
+
CONNECTORS-1031: Fix Zookeeper synchronization to be resilient
against short tick time settings.
(Karl Wright)
Modified:
manifoldcf/trunk/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperConnection.java
URL:
http://svn.apache.org/viewvc/manifoldcf/trunk/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperConnection.java?rev=1625856&r1=1625855&r2=1625856&view=diff
==============================================================================
---
manifoldcf/trunk/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperConnection.java
(original)
+++
manifoldcf/trunk/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperConnection.java
Wed Sep 17 23:41:31 2014
@@ -53,6 +53,8 @@ public class ZooKeeperConnection
// Transient state
protected String lockNode = null;
+ protected String nodePath = null;
+ protected byte[] nodeData = null;
/** Constructor. */
public ZooKeeperConnection(String connectString, int sessionTimeout)
@@ -86,16 +88,25 @@ public class ZooKeeperConnection
public void createNode(String nodePath, byte[] nodeData)
throws ManifoldCFException, InterruptedException
{
+ if (this.nodePath != null)
+ throw new IllegalStateException("Ephemeral node '"+this.nodePath+"'
already open; can't open '"+nodePath+"'.");
+
while (true)
{
try
{
- zookeeper.create(nodePath, nodeData, ZooDefs.Ids.OPEN_ACL_UNSAFE,
CreateMode.EPHEMERAL);
+ if (this.nodePath == null)
+ {
+ zookeeper.create(nodePath, nodeData, ZooDefs.Ids.OPEN_ACL_UNSAFE,
CreateMode.EPHEMERAL);
+ // Keep a record of the ephemeral node
+ this.nodePath = nodePath;
+ this.nodeData = nodeData;
+ }
break;
}
catch (KeeperException e)
{
- handleKeeperException(e);
+ handleEphemeralNodeKeeperException(e,false);
}
}
}
@@ -115,7 +126,7 @@ public class ZooKeeperConnection
}
catch (KeeperException e)
{
- handleKeeperException(e);
+ handleKeeperException(e,true);
}
}
}
@@ -132,27 +143,37 @@ public class ZooKeeperConnection
/** Set node data.
*/
- public void setNodeData(String nodePath, byte[] data)
+ public void setNodeData(byte[] data)
throws ManifoldCFException, InterruptedException
{
+ if (nodePath == null)
+ throw new IllegalStateException("Can't set data for a node path we did
not create: '"+nodePath+"'");
writeData(nodePath, data);
+ this.nodeData = data;
}
/** Delete a node.
*/
- public void deleteNode(String nodePath)
+ public void deleteNode()
throws ManifoldCFException, InterruptedException
{
+ if (nodePath == null)
+ throw new IllegalStateException("Can't delete ephemeral node that isn't
registered: '"+nodePath+"'");
while (true)
{
try
{
- zookeeper.delete(nodePath,-1);
+ if (nodePath != null)
+ {
+ zookeeper.delete(nodePath,-1);
+ nodePath = null;
+ nodeData = null;
+ }
return;
}
catch (KeeperException e)
{
- handleKeeperException(e);
+ handleEphemeralNodeKeeperException(e,false);
}
}
}
@@ -179,7 +200,7 @@ public class ZooKeeperConnection
}
catch (KeeperException e)
{
- handleKeeperException(e);
+ handleKeeperException(e,true);
}
}
}
@@ -211,7 +232,7 @@ public class ZooKeeperConnection
}
catch (KeeperException e)
{
- handleKeeperException(e);
+ handleKeeperException(e,true);
}
}
}
@@ -248,7 +269,7 @@ public class ZooKeeperConnection
}
catch (KeeperException e)
{
- handleKeeperException(e);
+ handleKeeperException(e,true);
}
}
}
@@ -269,7 +290,7 @@ public class ZooKeeperConnection
}
catch (KeeperException e)
{
- handleKeeperException(e);
+ handleKeeperException(e,true);
}
}
}
@@ -289,6 +310,7 @@ public class ZooKeeperConnection
try
{
// Assert that we want a write lock
+ String lockNode = this.lockNode;
if (lockNode == null)
lockNode = createSequentialChild(lockPath,WRITE_PREFIX);
String lockSequenceNumber = lockNode.substring(lockPath.length() + 1 +
WRITE_PREFIX.length());
@@ -313,11 +335,12 @@ public class ZooKeeperConnection
}
}
// We got it!
+ this.lockNode = lockNode;
return true;
}
catch (KeeperException e)
{
- handleEphemeralNodeKeeperException(e);
+ handleEphemeralNodeKeeperException(e,true);
}
}
}
@@ -336,6 +359,7 @@ public class ZooKeeperConnection
try
{
// Assert that we want a write lock
+ String lockNode = this.lockNode;
if (lockNode == null)
lockNode = createSequentialChild(lockPath,WRITE_PREFIX);
long lockSequenceNumber = new
Long(lockNode.substring(lockPath.length() + 1 +
WRITE_PREFIX.length())).longValue();
@@ -378,6 +402,7 @@ public class ZooKeeperConnection
{
// We got it!
//System.out.println("Got write lock for
'"+lockSequenceNumber+"'");
+ this.lockNode = lockNode;
return;
}
@@ -398,7 +423,7 @@ public class ZooKeeperConnection
}
catch (KeeperException e)
{
- handleEphemeralNodeKeeperException(e);
+ handleEphemeralNodeKeeperException(e,true);
}
}
}
@@ -418,6 +443,7 @@ public class ZooKeeperConnection
try
{
// Assert that we want a read lock
+ String lockNode = this.lockNode;
if (lockNode == null)
lockNode = createSequentialChild(lockPath,NONEXWRITE_PREFIX);
String lockSequenceNumber = lockNode.substring(lockPath.length() + 1 +
NONEXWRITE_PREFIX.length());
@@ -437,7 +463,7 @@ public class ZooKeeperConnection
}
catch (KeeperException e)
{
- handleKeeperException(e);
+ handleKeeperException(e,true);
}
}
if (children == null)
@@ -462,11 +488,12 @@ public class ZooKeeperConnection
}
}
// We got it!
+ this.lockNode = lockNode;
return true;
}
catch (KeeperException e)
{
- handleEphemeralNodeKeeperException(e);
+ handleEphemeralNodeKeeperException(e,true);
}
}
}
@@ -485,6 +512,7 @@ public class ZooKeeperConnection
try
{
// Assert that we want a read lock
+ String lockNode = this.lockNode;
if (lockNode == null)
lockNode = createSequentialChild(lockPath,NONEXWRITE_PREFIX);
long lockSequenceNumber = new
Long(lockNode.substring(lockPath.length() + 1 +
NONEXWRITE_PREFIX.length())).longValue();
@@ -505,7 +533,7 @@ public class ZooKeeperConnection
}
catch (KeeperException e)
{
- handleKeeperException(e);
+ handleKeeperException(e,true);
}
}
@@ -540,8 +568,11 @@ public class ZooKeeperConnection
}
if (gotLock)
+ {
// We got it!
+ this.lockNode = lockNode;
return;
+ }
// There SHOULD be a previous node immediately prior to the one we
asserted. If we didn't find one, go back around;
// the previous lock was probably created and destroyed before we
managed to get the children.
@@ -557,7 +588,7 @@ public class ZooKeeperConnection
}
catch (KeeperException e)
{
- handleEphemeralNodeKeeperException(e);
+ handleEphemeralNodeKeeperException(e,true);
}
}
}
@@ -577,6 +608,7 @@ public class ZooKeeperConnection
try
{
// Assert that we want a read lock
+ String lockNode = this.lockNode;
if (lockNode == null)
lockNode = createSequentialChild(lockPath,READ_PREFIX);
String lockSequenceNumber = lockNode.substring(lockPath.length() + 1 +
READ_PREFIX.length());
@@ -595,7 +627,7 @@ public class ZooKeeperConnection
}
catch (KeeperException e)
{
- handleKeeperException(e);
+ handleKeeperException(e,true);
}
}
if (children == null)
@@ -617,11 +649,12 @@ public class ZooKeeperConnection
}
}
// We got it!
+ this.lockNode= lockNode;
return true;
}
catch (KeeperException e)
{
- handleEphemeralNodeKeeperException(e);
+ handleEphemeralNodeKeeperException(e,true);
}
}
}
@@ -640,6 +673,7 @@ public class ZooKeeperConnection
try
{
// Assert that we want a read lock
+ String lockNode = this.lockNode;
if (lockNode == null)
lockNode = createSequentialChild(lockPath,READ_PREFIX);
long lockSequenceNumber = new
Long(lockNode.substring(lockPath.length() + 1 +
READ_PREFIX.length())).longValue();
@@ -661,7 +695,7 @@ public class ZooKeeperConnection
}
catch (KeeperException e)
{
- handleKeeperException(e);
+ handleKeeperException(e,true);
}
}
@@ -700,6 +734,7 @@ public class ZooKeeperConnection
{
// We got it!
//System.out.println("Got read lock for '"+lockSequenceNumber+"'");
+ this.lockNode = lockNode;
return;
}
@@ -721,7 +756,7 @@ public class ZooKeeperConnection
}
catch (KeeperException e)
{
- handleEphemeralNodeKeeperException(e);
+ handleEphemeralNodeKeeperException(e,true);
}
}
}
@@ -745,9 +780,14 @@ public class ZooKeeperConnection
lockNode = null;
break;
}
+ catch (InterruptedException e)
+ {
+ lockNode = null;
+ throw e;
+ }
catch (KeeperException e)
{
- handleEphemeralNodeKeeperException(e);
+ handleEphemeralNodeKeeperException(e,true);
}
}
}
@@ -767,7 +807,7 @@ public class ZooKeeperConnection
}
catch (KeeperException e)
{
- handleKeeperException(e);
+ handleKeeperException(e,true);
}
}
}
@@ -818,7 +858,7 @@ public class ZooKeeperConnection
}
catch (KeeperException e)
{
- handleKeeperException(e);
+ handleKeeperException(e,true);
}
}
}
@@ -842,7 +882,7 @@ public class ZooKeeperConnection
}
catch (KeeperException e)
{
- handleKeeperException(e);
+ handleKeeperException(e,true);
}
}
}
@@ -866,7 +906,7 @@ public class ZooKeeperConnection
}
catch (KeeperException e)
{
- handleKeeperException(e);
+ handleKeeperException(e,true);
}
}
}
@@ -883,7 +923,7 @@ public class ZooKeeperConnection
}
catch (KeeperException e)
{
- handleKeeperException(e);
+ handleKeeperException(e,true);
}
}
}
@@ -901,15 +941,37 @@ public class ZooKeeperConnection
/** Handle keeper exceptions that may involve ephemeral node creation.
*/
- protected void handleEphemeralNodeKeeperException(KeeperException e)
+ protected void handleEphemeralNodeKeeperException(KeeperException e, boolean
recreate)
throws ManifoldCFException, InterruptedException
{
if (e instanceof KeeperException.ConnectionLossException || e instanceof
KeeperException.SessionExpiredException)
{
- // Close the handle, open a new one
- lockNode = null;
- zookeeper.close();
- createSession();
+ while (true)
+ {
+ try
+ {
+ // Close the handle, open a new one
+ lockNode = null;
+ if (!recreate)
+ {
+ nodePath = null;
+ nodeData = null;
+ }
+ zookeeper.close();
+ createSession();
+ // Lock is lost, but we can (and should) recreate the ephemeral nodes
+ if (nodePath != null)
+ {
+ zookeeper.create(nodePath, nodeData, ZooDefs.Ids.OPEN_ACL_UNSAFE,
CreateMode.EPHEMERAL);
+ }
+ break;
+ }
+ catch (KeeperException e2)
+ {
+ if (!(e2 instanceof KeeperException.ConnectionLossException) && !(e2
instanceof KeeperException.SessionExpiredException))
+ throw new ManifoldCFException(e2.getMessage(),e2);
+ }
+ }
}
else
{
@@ -920,7 +982,7 @@ public class ZooKeeperConnection
/** Handle keeper exceptions that don't involve ephemeral node creation.
*/
- protected void handleKeeperException(KeeperException e)
+ protected void handleKeeperException(KeeperException e, boolean recreate)
throws ManifoldCFException, InterruptedException
{
if (e instanceof KeeperException.ConnectionLossException)
@@ -930,10 +992,32 @@ public class ZooKeeperConnection
}
else if (e instanceof KeeperException.SessionExpiredException)
{
- // Close the handle, open a new one
- lockNode = null;
- zookeeper.close();
- createSession();
+ while (true)
+ {
+ try
+ {
+ // Close the handle, open a new one
+ lockNode = null;
+ if (!recreate)
+ {
+ nodePath = null;
+ nodeData = null;
+ }
+ zookeeper.close();
+ createSession();
+ // Lock is lost, but we can (and should) recreate the ephemeral nodes
+ if (nodePath != null)
+ {
+ zookeeper.create(nodePath, nodeData, ZooDefs.Ids.OPEN_ACL_UNSAFE,
CreateMode.EPHEMERAL);
+ }
+ break;
+ }
+ catch (KeeperException e2)
+ {
+ if (!(e2 instanceof KeeperException.ConnectionLossException) && !(e2
instanceof KeeperException.SessionExpiredException))
+ throw new ManifoldCFException(e2.getMessage(),e2);
+ }
+ }
}
else
{
Added:
manifoldcf/trunk/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperEphemeralNodeObject.java
URL:
http://svn.apache.org/viewvc/manifoldcf/trunk/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperEphemeralNodeObject.java?rev=1625856&view=auto
==============================================================================
---
manifoldcf/trunk/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperEphemeralNodeObject.java
(added)
+++
manifoldcf/trunk/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperEphemeralNodeObject.java
Wed Sep 17 23:41:31 2014
@@ -0,0 +1,92 @@
+/**
+* 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.manifoldcf.core.lockmanager;
+
+import org.apache.manifoldcf.core.interfaces.*;
+import org.apache.manifoldcf.core.system.Logging;
+import org.apache.manifoldcf.core.system.ManifoldCF;
+
+/** This class keeps track of a zookeeper ephemeral node that is owned by the
+* current process.
+*/
+public class ZooKeeperEphemeralNodeObject
+{
+ private final ZooKeeperConnectionPool pool;
+ private final String nodePath;
+
+ private ZooKeeperConnection currentConnection = null;
+
+ public ZooKeeperEphemeralNodeObject(String nodePath, ZooKeeperConnectionPool
pool)
+ {
+ this.nodePath = nodePath;
+ this.pool = pool;
+ }
+
+ /** Create the specified node.
+ */
+ public synchronized void createNode(byte[] nodeData)
+ throws ManifoldCFException, InterruptedException
+ {
+ if (currentConnection != null)
+ throw new IllegalStateException("Already have a created node for
'"+nodePath+"'");
+ currentConnection = pool.grab();
+ try
+ {
+ currentConnection.createNode(nodePath,nodeData);
+ }
+ catch (Throwable t)
+ {
+ pool.release(currentConnection);
+ currentConnection = null;
+ if (t instanceof ManifoldCFException)
+ throw (ManifoldCFException)t;
+ if (t instanceof InterruptedException)
+ throw (InterruptedException)t;
+ if (t instanceof Error)
+ throw (Error)t;
+ if (t instanceof RuntimeException)
+ throw (RuntimeException)t;
+ throw new RuntimeException("Unexpected exception type:
"+t.getClass().getName()+": "+t.getMessage(),t);
+ }
+ }
+
+ /** Set the node's data.
+ */
+ public synchronized void setNodeData(byte[] nodeData)
+ throws ManifoldCFException, InterruptedException
+ {
+ if (currentConnection == null)
+ throw new IllegalStateException("Node not yet created for node path
'"+nodePath+"'");
+
+ currentConnection.setNodeData(nodeData);
+ }
+
+ /** Delete the node.
+ */
+ public synchronized void deleteNode()
+ throws ManifoldCFException, InterruptedException
+ {
+ if (currentConnection == null)
+ // It's allowed to delete the same node multiple times
+ return;
+
+ currentConnection.deleteNode();
+ pool.release(currentConnection);
+ currentConnection = null;
+ }
+
+}
Propchange:
manifoldcf/trunk/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperEphemeralNodeObject.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange:
manifoldcf/trunk/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperEphemeralNodeObject.java
------------------------------------------------------------------------------
svn:keywords = Id
Added:
manifoldcf/trunk/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperEphemeralNodePool.java
URL:
http://svn.apache.org/viewvc/manifoldcf/trunk/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperEphemeralNodePool.java?rev=1625856&view=auto
==============================================================================
---
manifoldcf/trunk/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperEphemeralNodePool.java
(added)
+++
manifoldcf/trunk/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperEphemeralNodePool.java
Wed Sep 17 23:41:31 2014
@@ -0,0 +1,88 @@
+/**
+* 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.manifoldcf.core.lockmanager;
+
+import org.apache.manifoldcf.core.interfaces.*;
+import org.apache.manifoldcf.core.system.Logging;
+import org.apache.manifoldcf.core.system.ManifoldCF;
+
+import java.util.*;
+
+/** This class represents a pool of ZooKeeperEphemeralNodeObject objects.
+* The key for this pool is the node path.
+*/
+public class ZooKeeperEphemeralNodePool
+{
+ protected final ZooKeeperConnectionPool pool;
+ protected final Map<String,ZooKeeperEphemeralNodeObject> nodes = new
HashMap<String,ZooKeeperEphemeralNodeObject>();
+
+ public ZooKeeperEphemeralNodePool(ZooKeeperConnectionPool pool)
+ {
+ this.pool = pool;
+ }
+
+ public void createNode(String nodePath, byte[] nodeData)
+ throws ManifoldCFException, InterruptedException
+ {
+ getObject(nodePath).createNode(nodeData);
+ }
+
+ public void setNodeData(String nodePath, byte[] nodeData)
+ throws ManifoldCFException, InterruptedException
+ {
+ getObject(nodePath).setNodeData(nodeData);
+ }
+
+ public void deleteNode(String nodePath)
+ throws ManifoldCFException, InterruptedException
+ {
+ synchronized (this)
+ {
+ ZooKeeperEphemeralNodeObject rval = nodes.get(nodePath);
+ if (rval != null)
+ {
+ rval.deleteNode();
+ nodes.remove(nodePath);
+ }
+ }
+ }
+
+ public void deleteAll()
+ throws ManifoldCFException, InterruptedException
+ {
+ synchronized (this)
+ {
+ while (nodes.size() > 0)
+ {
+ Iterator<String> nodePathIter = nodes.keySet().iterator();
+ String nodePath = nodePathIter.next();
+ deleteNode(nodePath);
+ }
+ }
+ }
+
+ protected synchronized ZooKeeperEphemeralNodeObject getObject(String
nodePath)
+ {
+ ZooKeeperEphemeralNodeObject rval = nodes.get(nodePath);
+ if (rval != null)
+ return rval;
+ rval = new ZooKeeperEphemeralNodeObject(nodePath,pool);
+ nodes.put(nodePath,rval);
+ return rval;
+ }
+
+}
Propchange:
manifoldcf/trunk/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperEphemeralNodePool.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange:
manifoldcf/trunk/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperEphemeralNodePool.java
------------------------------------------------------------------------------
svn:keywords = Id
Modified:
manifoldcf/trunk/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperLockManager.java
URL:
http://svn.apache.org/viewvc/manifoldcf/trunk/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperLockManager.java?rev=1625856&r1=1625855&r2=1625856&view=diff
==============================================================================
---
manifoldcf/trunk/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperLockManager.java
(original)
+++
manifoldcf/trunk/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperLockManager.java
Wed Sep 17 23:41:31 2014
@@ -54,6 +54,8 @@ public class ZooKeeperLockManager extend
protected static ZooKeeperConnectionPool pool = null;
protected static Integer zookeeperPoolLocker = new Integer(0);
protected static LockPool myZooKeeperLocks = null;
+ protected static Integer ephemeralPoolLocker = new Integer(0);
+ protected static ZooKeeperEphemeralNodePool myEphemeralNodes = null;
/** Constructor */
public ZooKeeperLockManager()
@@ -79,6 +81,13 @@ public class ZooKeeperLockManager extend
myZooKeeperLocks = new LockPool(new ZooKeeperLockObjectFactory(pool));
}
}
+ synchronized (ephemeralPoolLocker)
+ {
+ if (myEphemeralNodes == null)
+ {
+ myEphemeralNodes = new ZooKeeperEphemeralNodePool(pool);
+ }
+ }
}
// The node synchronization model involves keeping track of active agents
entities, so that other entities
@@ -213,7 +222,7 @@ public class ZooKeeperLockManager extend
}
// Last, set the appropriate active flag
- connection.createNode(activePath, initialData);
+ myEphemeralNodes.createNode(activePath, initialData);
return serviceName;
}
finally
@@ -251,7 +260,7 @@ public class ZooKeeperLockManager extend
try
{
String activePath = buildServiceTypeActivePath(serviceType,
ZooKeeperConnection.zooKeeperSafeName(serviceName));
- connection.setNodeData(activePath, (serviceData==null)?new
byte[0]:serviceData);
+ myEphemeralNodes.setNodeData(activePath, (serviceData==null)?new
byte[0]:serviceData);
}
finally
{
@@ -475,7 +484,7 @@ public class ZooKeeperLockManager extend
enterServiceRegistryWriteLock(connection, serviceType);
try
{
- connection.deleteNode(buildServiceTypeActivePath(serviceType,
ZooKeeperConnection.zooKeeperSafeName(serviceName)));
+ myEphemeralNodes.deleteNode(buildServiceTypeActivePath(serviceType,
ZooKeeperConnection.zooKeeperSafeName(serviceName)));
}
finally
{
@@ -907,6 +916,22 @@ public class ZooKeeperLockManager extend
protected static void shutdownPool()
throws ManifoldCFException
{
+ synchronized (ephemeralPoolLocker)
+ {
+ if (myEphemeralNodes != null)
+ {
+ try
+ {
+ myEphemeralNodes.deleteAll();
+ myEphemeralNodes = null;
+ }
+ catch (InterruptedException e)
+ {
+ throw new
ManifoldCFException(e.getMessage(),e,ManifoldCFException.INTERRUPTED);
+ }
+ }
+ }
+
synchronized (connectionPoolLock)
{
if (pool != null)
