You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by ma...@apache.org on 2011/09/28 22:49:16 UTC
svn commit: r1177057 - in /lucene/dev/branches/solrcloud/solr/core/src:
java/org/apache/solr/cloud/ java/org/apache/solr/cloud/lock/
test/org/apache/solr/cloud/ test/org/apache/solr/cloud/lock/
Author: markrmiller
Date: Wed Sep 28 20:49:15 2011
New Revision: 1177057
URL: http://svn.apache.org/viewvc?rev=1177057&view=rev
Log:
SOLR-2723: If you don't choose a shard name for a SolrCore, the system should auto assign shard names. Dopey strawman that's hard coded to 3 slices.
Added:
lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/AssignShard.java (with props)
lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/ZkCollectionLock.java (with props)
lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/lock/
lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/lock/LockListener.java (with props)
lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/lock/ProtocolSupport.java (with props)
lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/lock/WriteLock.java (with props)
lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/lock/ZNodeName.java (with props)
lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/lock/ZooKeeperOperation.java (with props)
lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/ZkLockTest.java (with props)
lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/lock/
lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/lock/WriteLockTest.java (with props)
Modified:
lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/ZkController.java
lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/AbstractZkTestCase.java
lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/ZkControllerTest.java
Added: lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/AssignShard.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/AssignShard.java?rev=1177057&view=auto
==============================================================================
--- lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/AssignShard.java (added)
+++ lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/AssignShard.java Wed Sep 28 20:49:15 2011
@@ -0,0 +1,95 @@
+package org.apache.solr.cloud;
+
+/**
+ * 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.
+ */
+
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.SortedMap;
+import java.util.TreeMap;
+
+import org.apache.solr.common.cloud.SolrZkClient;
+import org.apache.solr.common.cloud.ZkStateReader;
+import org.apache.zookeeper.KeeperException;
+
+public class AssignShard {
+ private SolrZkClient client;
+
+ public AssignShard(SolrZkClient client) {
+ this.client = client;
+ }
+
+ /**
+ * Assign a new unique id up to slices count - then add replicas evenly.
+ *
+ * @param collection
+ *
+ * @param slices
+ * @return
+ * @throws InterruptedException
+ * @throws KeeperException
+ */
+ public String assignShard(String collection, int slices)
+ throws KeeperException, InterruptedException {
+ // we want the collection lock
+ ZkCollectionLock lock = new ZkCollectionLock(client);
+ lock.lock();
+ String returnShardId = null;
+ try {
+ // lets read the current shards - we want to read straight from zk, and we
+ // assume we have some kind
+ // of collection level lock
+ String shardIdPaths = ZkStateReader.COLLECTIONS_ZKNODE + "/" + collection
+ + ZkStateReader.SHARDS_ZKNODE;
+
+ List<String> shardIdNames = client.getChildren(shardIdPaths, null);
+
+ if (shardIdNames.size() == 0) {
+ return "shard1";
+ }
+
+ if (shardIdNames.size() < slices) {
+ return "shard" + (shardIdNames.size() + 1);
+ }
+
+ // else figure out which shard needs more replicas
+ final Map<String,Integer> map = new HashMap<String,Integer>();
+ for (String shardId : shardIdNames) {
+ int cnt = client.getChildren(shardIdPaths + "/" + shardId, null).size();
+ map.put(shardId, cnt);
+ }
+
+ Collections.sort(shardIdNames, new Comparator<String>() {
+
+ @Override
+ public int compare(String o1, String o2) {
+ Integer one = map.get(o1);
+ Integer two = map.get(o2);
+ return one.compareTo(two);
+ }
+ });
+
+ returnShardId = shardIdNames.get(0);
+ } finally {
+ lock.unlock();
+ }
+ return returnShardId;
+ }
+}
Added: lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/ZkCollectionLock.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/ZkCollectionLock.java?rev=1177057&view=auto
==============================================================================
--- lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/ZkCollectionLock.java (added)
+++ lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/ZkCollectionLock.java Wed Sep 28 20:49:15 2011
@@ -0,0 +1,67 @@
+package org.apache.solr.cloud;
+
+import org.apache.solr.cloud.lock.LockListener;
+import org.apache.solr.cloud.lock.WriteLock;
+import org.apache.solr.common.cloud.SolrZkClient;
+import org.apache.zookeeper.KeeperException;
+
+/**
+ * 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.
+ */
+
+public class ZkCollectionLock {
+ private WriteLock lock;
+ private volatile boolean gotLock = false;
+
+ public ZkCollectionLock(SolrZkClient zkClient) {
+ lock = new WriteLock(zkClient.getSolrZooKeeper(),
+ "/collections/collection1/shards_lock", null, new LockListener() {
+
+ @Override
+ public void lockReleased() {
+ gotLock = false;
+ }
+
+ @Override
+ public void lockAcquired() {
+ System.out.println("I got the lock!");
+ gotLock = true;
+ }
+ });
+
+ }
+
+ public void lock() throws KeeperException, InterruptedException {
+ if (lock.lock()) {
+ // we got the lock, just return
+ gotLock = true;
+ return;
+ }
+
+ // we need to wait for the lock
+ int cnt = 0;
+ while (!gotLock) {
+ if (cnt++ == 15) {
+ throw new RuntimeException("Coulnd't aquire the shard lock");
+ }
+ Thread.sleep(1000);
+ }
+ }
+
+ public void unlock() {
+ lock.unlock();
+ }
+}
Modified: lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/ZkController.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/ZkController.java?rev=1177057&r1=1177056&r2=1177057&view=diff
==============================================================================
--- lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/ZkController.java (original)
+++ lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/ZkController.java Wed Sep 28 20:49:15 2011
@@ -87,6 +87,8 @@ public final class ZkController {
private String hostName;
+ private AssignShard assignShard;
+
/**
@@ -153,6 +155,7 @@ public final class ZkController {
leaderElector = new SliceLeaderElector(zkClient);
zkStateReader = new ZkStateReader(zkClient);
+ assignShard = new AssignShard(zkClient);
init();
}
@@ -463,23 +466,30 @@ public final class ZkController {
return configName;
}
+
/**
* Register shard with ZooKeeper.
*
* @param coreName
* @param cloudDesc
+ * @return
* @throws IOException
* @throws KeeperException
* @throws InterruptedException
*/
- public void register(String coreName, final CloudDescriptor cloudDesc) throws IOException,
+ public String register(String coreName, final CloudDescriptor cloudDesc) throws IOException,
KeeperException, InterruptedException {
String shardUrl = localHostName + ":" + localHostPort + "/" + localHostContext
+ "/" + coreName;
final String collection = cloudDesc.getCollectionName();
- String shardsZkPath = ZkStateReader.COLLECTIONS_ZKNODE + "/" + collection + ZkStateReader.SHARDS_ZKNODE + "/" + cloudDesc.getShardId();
+ String shardId = cloudDesc.getShardId();
+ if (shardId == null) {
+ shardId = assignShard.assignShard(collection, 3); // nocommit: hard coded
+ // number of slices
+ }
+ String shardsZkPath = ZkStateReader.COLLECTIONS_ZKNODE + "/" + collection + ZkStateReader.SHARDS_ZKNODE + "/" + shardId;
boolean shardZkNodeAlreadyExists = zkClient.exists(shardsZkPath);
@@ -502,8 +512,9 @@ public final class ZkController {
// tell everyone to update cloud info
zkClient.setData(ZkStateReader.COLLECTIONS_ZKNODE, (byte[])null);
} else {
- addZkShardsNode(cloudDesc.getShardId(), collection);
+ addZkShardsNode(shardId, collection);
try {
+ log.info("create node:" + shardZkNodeName);
zkClient.create(shardsZkPath + "/" + shardZkNodeName, bytes,
CreateMode.PERSISTENT);
@@ -521,15 +532,15 @@ public final class ZkController {
}
// leader election
- doLeaderElectionProcess(cloudDesc, collection, shardZkNodeName);
-
+ doLeaderElectionProcess(shardId, collection, shardZkNodeName);
+ return shardId;
}
- private void doLeaderElectionProcess(final CloudDescriptor cloudDesc,
+ private void doLeaderElectionProcess(String shardId,
final String collection, String shardZkNodeName) throws KeeperException,
InterruptedException, UnsupportedEncodingException {
- leaderElector.joinElection(cloudDesc.getShardId(), collection, shardZkNodeName);
+ leaderElector.joinElection(shardId, collection, shardZkNodeName);
}
/**
@@ -672,6 +683,33 @@ public final class ZkController {
}
zkClient.makePath(collectionPath, collectionProps.store(), CreateMode.PERSISTENT, null, true);
+ try {
+ // shards_lock node
+ if (!zkClient.exists(ZkStateReader.COLLECTIONS_ZKNODE + "/" + collection + "/shards_lock")) {
+ // makes shards_lock zkNode if it doesn't exist
+ zkClient.makePath(ZkStateReader.COLLECTIONS_ZKNODE + "/" + collection + "/shards_lock");
+ }
+ } catch (KeeperException e) {
+ // its okay if another beats us creating the node
+ if (e.code() != KeeperException.Code.NODEEXISTS) {
+ throw e;
+ }
+ }
+ try {
+ // shards node
+ if (!zkClient.exists(ZkStateReader.COLLECTIONS_ZKNODE + "/" + collection
+ + ZkStateReader.SHARDS_ZKNODE)) {
+ // makes shards zkNode if it doesn't exist
+ zkClient.makePath(ZkStateReader.COLLECTIONS_ZKNODE + "/" + collection
+ + ZkStateReader.SHARDS_ZKNODE);
+ }
+ } catch (KeeperException e) {
+ // its okay if another beats us creating the node
+ if (e.code() != KeeperException.Code.NODEEXISTS) {
+ throw e;
+ }
+ }
+
// ping that there is a new collection
zkClient.setData(ZkStateReader.COLLECTIONS_ZKNODE, (byte[])null);
Added: lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/lock/LockListener.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/lock/LockListener.java?rev=1177057&view=auto
==============================================================================
--- lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/lock/LockListener.java (added)
+++ lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/lock/LockListener.java Wed Sep 28 20:49:15 2011
@@ -0,0 +1,38 @@
+/**
+ *
+ * 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.solr.cloud.lock;
+
+/**
+ * This class has two methods which are call
+ * back methods when a lock is acquired and
+ * when the lock is released.
+ *
+ */
+public interface LockListener {
+ /**
+ * call back called when the lock
+ * is acquired
+ */
+ public void lockAcquired();
+
+ /**
+ * call back called when the lock is
+ * released.
+ */
+ public void lockReleased();
+}
\ No newline at end of file
Added: lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/lock/ProtocolSupport.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/lock/ProtocolSupport.java?rev=1177057&view=auto
==============================================================================
--- lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/lock/ProtocolSupport.java (added)
+++ lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/lock/ProtocolSupport.java Wed Sep 28 20:49:15 2011
@@ -0,0 +1,191 @@
+/**
+ *
+ * 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.solr.cloud.lock;
+
+import org.apache.log4j.Logger;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.ZooDefs;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.data.ACL;
+import org.apache.zookeeper.data.Stat;
+
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * A base class for protocol implementations which provides a number of higher
+ * level helper methods for working with ZooKeeper along with retrying synchronous
+ * operations if the connection to ZooKeeper closes such as
+ * {@link #retryOperation(ZooKeeperOperation)}
+ *
+ */
+class ProtocolSupport {
+ private static final Logger LOG = Logger.getLogger(ProtocolSupport.class);
+
+ protected final ZooKeeper zookeeper;
+ private AtomicBoolean closed = new AtomicBoolean(false);
+ private long retryDelay = 1000L;
+ private int retryCount = 15;
+ private List<ACL> acl = ZooDefs.Ids.OPEN_ACL_UNSAFE;
+
+ public ProtocolSupport(ZooKeeper zookeeper) {
+ this.zookeeper = zookeeper;
+ }
+
+ /**
+ * Closes this strategy and releases any ZooKeeper resources; but keeps the
+ * ZooKeeper instance open
+ */
+ public void close() {
+ if (closed.compareAndSet(false, true)) {
+ doClose();
+ }
+ }
+
+ /**
+ * return zookeeper client instance
+ * @return zookeeper client instance
+ */
+ public ZooKeeper getZookeeper() {
+ return zookeeper;
+ }
+
+ /**
+ * return the acl its using
+ * @return the acl.
+ */
+ public List<ACL> getAcl() {
+ return acl;
+ }
+
+ /**
+ * set the acl
+ * @param acl the acl to set to
+ */
+ public void setAcl(List<ACL> acl) {
+ this.acl = acl;
+ }
+
+ /**
+ * get the retry delay in milliseconds
+ * @return the retry delay
+ */
+ public long getRetryDelay() {
+ return retryDelay;
+ }
+
+ /**
+ * Sets the time waited between retry delays
+ * @param retryDelay the retry delay
+ */
+ public void setRetryDelay(long retryDelay) {
+ this.retryDelay = retryDelay;
+ }
+
+ /**
+ * Allow derived classes to perform
+ * some custom closing operations to release resources
+ */
+ protected void doClose() {
+ }
+
+
+ /**
+ * Perform the given operation, retrying if the connection fails
+ * @return object. it needs to be cast to the callee's expected
+ * return type.
+ */
+ protected Object retryOperation(ZooKeeperOperation operation)
+ throws KeeperException, InterruptedException {
+ KeeperException exception = null;
+ for (int i = 0; i < retryCount; i++) {
+ try {
+ return operation.execute();
+ } catch (KeeperException.SessionExpiredException e) {
+ LOG.warn("Session expired for: " + zookeeper + " so reconnecting due to: " + e, e);
+ throw e;
+ } catch (KeeperException.ConnectionLossException e) {
+ if (exception == null) {
+ exception = e;
+ }
+ LOG.debug("Attempt " + i + " failed with connection loss so " +
+ "attempting to reconnect: " + e, e);
+ retryDelay(i);
+ }
+ }
+ throw exception;
+ }
+
+ /**
+ * Ensures that the given path exists with no data, the current
+ * ACL and no flags
+ * @param path
+ */
+ protected void ensurePathExists(String path) {
+ ensureExists(path, null, acl, CreateMode.PERSISTENT);
+ }
+
+ /**
+ * Ensures that the given path exists with the given data, ACL and flags
+ * @param path
+ * @param acl
+ * @param flags
+ */
+ protected void ensureExists(final String path, final byte[] data,
+ final List<ACL> acl, final CreateMode flags) {
+ try {
+ retryOperation(new ZooKeeperOperation() {
+ public boolean execute() throws KeeperException, InterruptedException {
+ Stat stat = zookeeper.exists(path, false);
+ if (stat != null) {
+ return true;
+ }
+ zookeeper.create(path, data, acl, flags);
+ return true;
+ }
+ });
+ } catch (KeeperException e) {
+ LOG.warn("Caught: " + e, e);
+ } catch (InterruptedException e) {
+ LOG.warn("Caught: " + e, e);
+ }
+ }
+
+ /**
+ * Returns true if this protocol has been closed
+ * @return true if this protocol is closed
+ */
+ protected boolean isClosed() {
+ return closed.get();
+ }
+
+ /**
+ * Performs a retry delay if this is not the first attempt
+ * @param attemptCount the number of the attempts performed so far
+ */
+ protected void retryDelay(int attemptCount) {
+ if (attemptCount > 0) {
+ try {
+ Thread.sleep(attemptCount * retryDelay);
+ } catch (InterruptedException e) {
+ LOG.debug("Failed to sleep: " + e, e);
+ }
+ }
+ }
+}
Added: lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/lock/WriteLock.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/lock/WriteLock.java?rev=1177057&view=auto
==============================================================================
--- lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/lock/WriteLock.java (added)
+++ lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/lock/WriteLock.java Wed Sep 28 20:49:15 2011
@@ -0,0 +1,295 @@
+/**
+ *
+ * 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.solr.cloud.lock;
+
+import org.apache.log4j.Logger;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher;
+import static org.apache.zookeeper.CreateMode.EPHEMERAL_SEQUENTIAL;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.data.ACL;
+import org.apache.zookeeper.data.Stat;
+
+import java.util.List;
+import java.util.SortedSet;
+import java.util.TreeSet;
+
+/**
+ * A <a href="package.html">protocol to implement an exclusive
+ * write lock or to elect a leader</a>. <p/> You invoke {@link #lock()} to
+ * start the process of grabbing the lock; you may get the lock then or it may be
+ * some time later. <p/> You can register a listener so that you are invoked
+ * when you get the lock; otherwise you can ask if you have the lock
+ * by calling {@link #isOwner()}
+ *
+ */
+public class WriteLock extends ProtocolSupport {
+ private static final Logger LOG = Logger.getLogger(WriteLock.class);
+
+ private final String dir;
+ private String id;
+ private ZNodeName idName;
+ private String ownerId;
+ private String lastChildId;
+ private byte[] data = {0x12, 0x34};
+ private LockListener callback;
+ private LockZooKeeperOperation zop;
+
+ /**
+ * zookeeper contructor for writelock
+ * @param zookeeper zookeeper client instance
+ * @param dir the parent path you want to use for locking
+ * @param acls the acls that you want to use for all the paths,
+ * if null world read/write is used.
+ */
+ public WriteLock(ZooKeeper zookeeper, String dir, List<ACL> acl) {
+ super(zookeeper);
+ this.dir = dir;
+ if (acl != null) {
+ setAcl(acl);
+ }
+ this.zop = new LockZooKeeperOperation();
+ }
+
+ /**
+ * zookeeper contructor for writelock with callback
+ * @param zookeeper the zookeeper client instance
+ * @param dir the parent path you want to use for locking
+ * @param acl the acls that you want to use for all the paths
+ * @param callback the call back instance
+ */
+ public WriteLock(ZooKeeper zookeeper, String dir, List<ACL> acl,
+ LockListener callback) {
+ this(zookeeper, dir, acl);
+ this.callback = callback;
+ }
+
+ /**
+ * return the current locklistener
+ * @return the locklistener
+ */
+ public LockListener getLockListener() {
+ return this.callback;
+ }
+
+ /**
+ * register a different call back listener
+ * @param callback the call back instance
+ */
+ public void setLockListener(LockListener callback) {
+ this.callback = callback;
+ }
+
+ /**
+ * Removes the lock or associated znode if
+ * you no longer require the lock. this also
+ * removes your request in the queue for locking
+ * in case you do not already hold the lock.
+ * @throws RuntimeException throws a runtime exception
+ * if it cannot connect to zookeeper.
+ */
+ public synchronized void unlock() throws RuntimeException {
+
+ if (!isClosed() && id != null) {
+ // we don't need to retry this operation in the case of failure
+ // as ZK will remove ephemeral files and we don't wanna hang
+ // this process when closing if we cannot reconnect to ZK
+ try {
+
+ ZooKeeperOperation zopdel = new ZooKeeperOperation() {
+ public boolean execute() throws KeeperException,
+ InterruptedException {
+ zookeeper.delete(id, -1);
+ return Boolean.TRUE;
+ }
+ };
+ zopdel.execute();
+ } catch (InterruptedException e) {
+ LOG.warn("Caught: " + e, e);
+ //set that we have been interrupted.
+ Thread.currentThread().interrupt();
+ } catch (KeeperException.NoNodeException e) {
+ // do nothing
+ } catch (KeeperException e) {
+ LOG.warn("Caught: " + e, e);
+ throw (RuntimeException) new RuntimeException(e.getMessage()).
+ initCause(e);
+ }
+ finally {
+ if (callback != null) {
+ callback.lockReleased();
+ }
+ id = null;
+ }
+ }
+ }
+
+ /**
+ * the watcher called on
+ * getting watch while watching
+ * my predecessor
+ */
+ private class LockWatcher implements Watcher {
+ public void process(WatchedEvent event) {
+ // lets either become the leader or watch the new/updated node
+ LOG.debug("Watcher fired on path: " + event.getPath() + " state: " +
+ event.getState() + " type " + event.getType());
+ try {
+ lock();
+ } catch (Exception e) {
+ LOG.warn("Failed to acquire lock: " + e, e);
+ }
+ }
+ }
+
+ /**
+ * a zoookeeper operation that is mainly responsible
+ * for all the magic required for locking.
+ */
+ private class LockZooKeeperOperation implements ZooKeeperOperation {
+
+ /** find if we have been created earler if not create our node
+ *
+ * @param prefix the prefix node
+ * @param zookeeper teh zookeeper client
+ * @param dir the dir paretn
+ * @throws KeeperException
+ * @throws InterruptedException
+ */
+ private void findPrefixInChildren(String prefix, ZooKeeper zookeeper, String dir)
+ throws KeeperException, InterruptedException {
+ List<String> names = zookeeper.getChildren(dir, false);
+ for (String name : names) {
+ if (name.startsWith(prefix)) {
+ id = name;
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Found id created last time: " + id);
+ }
+ break;
+ }
+ }
+ if (id == null) {
+ id = zookeeper.create(dir + "/" + prefix, data,
+ getAcl(), EPHEMERAL_SEQUENTIAL);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Created id: " + id);
+ }
+ }
+
+ }
+
+ /**
+ * the command that is run and retried for actually
+ * obtaining the lock
+ * @return if the command was successful or not
+ */
+ public boolean execute() throws KeeperException, InterruptedException {
+ do {
+ if (id == null) {
+ long sessionId = zookeeper.getSessionId();
+ String prefix = "x-" + sessionId + "-";
+ // lets try look up the current ID if we failed
+ // in the middle of creating the znode
+ findPrefixInChildren(prefix, zookeeper, dir);
+ idName = new ZNodeName(id);
+ }
+ if (id != null) {
+ List<String> names = zookeeper.getChildren(dir, false);
+ if (names.isEmpty()) {
+ LOG.warn("No children in: " + dir + " when we've just " +
+ "created one! Lets recreate it...");
+ // lets force the recreation of the id
+ id = null;
+ } else {
+ // lets sort them explicitly (though they do seem to come back in order ususally :)
+ SortedSet<ZNodeName> sortedNames = new TreeSet<ZNodeName>();
+ for (String name : names) {
+ sortedNames.add(new ZNodeName(dir + "/" + name));
+ }
+ ownerId = sortedNames.first().getName();
+ SortedSet<ZNodeName> lessThanMe = sortedNames.headSet(idName);
+ if (!lessThanMe.isEmpty()) {
+ ZNodeName lastChildName = lessThanMe.last();
+ lastChildId = lastChildName.getName();
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("watching less than me node: " + lastChildId);
+ }
+ Stat stat = zookeeper.exists(lastChildId, new LockWatcher());
+ if (stat != null) {
+ return Boolean.FALSE;
+ } else {
+ LOG.warn("Could not find the" +
+ " stats for less than me: " + lastChildName.getName());
+ }
+ } else {
+ if (isOwner()) {
+ if (callback != null) {
+ callback.lockAcquired();
+ }
+ return Boolean.TRUE;
+ }
+ }
+ }
+ }
+ }
+ while (id == null);
+ return Boolean.FALSE;
+ }
+ };
+
+ /**
+ * Attempts to acquire the exclusive write lock returning whether or not it was
+ * acquired. Note that the exclusive lock may be acquired some time later after
+ * this method has been invoked due to the current lock owner going away.
+ */
+ public synchronized boolean lock() throws KeeperException, InterruptedException {
+ if (isClosed()) {
+ return false;
+ }
+ ensurePathExists(dir);
+
+ return (Boolean) retryOperation(zop);
+ }
+
+ /**
+ * return the parent dir for lock
+ * @return the parent dir used for locks.
+ */
+ public String getDir() {
+ return dir;
+ }
+
+ /**
+ * Returns true if this node is the owner of the
+ * lock (or the leader)
+ */
+ public boolean isOwner() {
+ return id != null && ownerId != null && id.equals(ownerId);
+ }
+
+ /**
+ * return the id for this lock
+ * @return the id for this lock
+ */
+ public String getId() {
+ return this.id;
+ }
+}
+
Added: lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/lock/ZNodeName.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/lock/ZNodeName.java?rev=1177057&view=auto
==============================================================================
--- lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/lock/ZNodeName.java (added)
+++ lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/lock/ZNodeName.java Wed Sep 28 20:49:15 2011
@@ -0,0 +1,109 @@
+/**
+ *
+ * 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.solr.cloud.lock;
+
+import org.apache.log4j.Logger;
+
+/**
+ * Represents an ephemeral znode name which has an ordered sequence number
+ * and can be sorted in order
+ *
+ */
+class ZNodeName implements Comparable<ZNodeName> {
+ private final String name;
+ private String prefix;
+ private int sequence = -1;
+ private static final Logger LOG = Logger.getLogger(ZNodeName.class);
+
+ public ZNodeName(String name) {
+ if (name == null) {
+ throw new NullPointerException("id cannot be null");
+ }
+ this.name = name;
+ this.prefix = name;
+ int idx = name.lastIndexOf('-');
+ if (idx >= 0) {
+ this.prefix = name.substring(0, idx);
+ try {
+ this.sequence = Integer.parseInt(name.substring(idx + 1));
+ // If an exception occurred we misdetected a sequence suffix,
+ // so return -1.
+ } catch (NumberFormatException e) {
+ LOG.info("Number format exception for " + idx, e);
+ } catch (ArrayIndexOutOfBoundsException e) {
+ LOG.info("Array out of bounds for " + idx, e);
+ }
+ }
+ }
+
+ @Override
+ public String toString() {
+ return name.toString();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ ZNodeName sequence = (ZNodeName) o;
+
+ if (!name.equals(sequence.name)) return false;
+
+ return true;
+ }
+
+ @Override
+ public int hashCode() {
+ return name.hashCode() + 37;
+ }
+
+ public int compareTo(ZNodeName that) {
+ int answer = this.prefix.compareTo(that.prefix);
+ if (answer == 0) {
+ int s1 = this.sequence;
+ int s2 = that.sequence;
+ if (s1 == -1 && s2 == -1) {
+ return this.name.compareTo(that.name);
+ }
+ answer = s1 == -1 ? 1 : s2 == -1 ? -1 : s1 - s2;
+ }
+ return answer;
+ }
+
+ /**
+ * Returns the name of the znode
+ */
+ public String getName() {
+ return name;
+ }
+
+ /**
+ * Returns the sequence number
+ */
+ public int getZNodeName() {
+ return sequence;
+ }
+
+ /**
+ * Returns the text prefix before the sequence number
+ */
+ public String getPrefix() {
+ return prefix;
+ }
+}
Added: lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/lock/ZooKeeperOperation.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/lock/ZooKeeperOperation.java?rev=1177057&view=auto
==============================================================================
--- lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/lock/ZooKeeperOperation.java (added)
+++ lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/lock/ZooKeeperOperation.java Wed Sep 28 20:49:15 2011
@@ -0,0 +1,38 @@
+/**
+ *
+ * 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.solr.cloud.lock;
+
+import org.apache.zookeeper.KeeperException;
+
+/**
+ * A callback object which can be used for implementing retry-able operations in the
+ * {@link org.apache.zookeeper.recipes.lock.ProtocolSupport} class
+ *
+ */
+public interface ZooKeeperOperation {
+
+ /**
+ * Performs the operation - which may be involved multiple times if the connection
+ * to ZooKeeper closes during this operation
+ *
+ * @return the result of the operation or null
+ * @throws KeeperException
+ * @throws InterruptedException
+ */
+ public boolean execute() throws KeeperException, InterruptedException;
+}
Modified: lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/AbstractZkTestCase.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/AbstractZkTestCase.java?rev=1177057&r1=1177056&r2=1177057&view=diff
==============================================================================
--- lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/AbstractZkTestCase.java (original)
+++ lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/AbstractZkTestCase.java Wed Sep 28 20:49:15 2011
@@ -123,13 +123,13 @@ public abstract class AbstractZkTestCase
zkClient.close();
}
- static void makeSolrZkNode(String zkHost) throws Exception {
+ public static void makeSolrZkNode(String zkHost) throws Exception {
SolrZkClient zkClient = new SolrZkClient(zkHost, TIMEOUT);
zkClient.makePath("/solr");
zkClient.close();
}
- static void tryCleanSolrZkNode(String zkHost) throws Exception {
+ public static void tryCleanSolrZkNode(String zkHost) throws Exception {
tryCleanPath(zkHost, "/solr");
}
Modified: lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/ZkControllerTest.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/ZkControllerTest.java?rev=1177057&r1=1177056&r2=1177057&view=diff
==============================================================================
--- lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/ZkControllerTest.java (original)
+++ lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/ZkControllerTest.java Wed Sep 28 20:49:15 2011
@@ -226,6 +226,67 @@ public class ZkControllerTest extends So
}
}
+
+ @Test
+ public void testAutoShard() throws Exception {
+ String zkDir = dataDir.getAbsolutePath() + File.separator
+ + "zookeeper/server1/data";
+
+ ZkTestServer server = new ZkTestServer(zkDir);
+
+
+ ZkController zkController = null;
+ try {
+ server.run();
+ AbstractZkTestCase.tryCleanSolrZkNode(server.getZkHost());
+ AbstractZkTestCase.makeSolrZkNode(server.getZkHost());
+
+ zkController = new ZkController(server.getZkAddress(),
+ TIMEOUT, 10000, "localhost", "8983", "solr", new CurrentCoreDescriptorProvider() {
+
+ @Override
+ public List<CoreDescriptor> getCurrentDescriptors() {
+ // do nothing
+ return null;
+ }
+ });
+
+ System.setProperty("bootstrap_confdir", getFile("solr/conf").getAbsolutePath());
+
+ // ensure our shards node for the collection exists - other tests can mess with this
+
+ CloudDescriptor cloudDesc = new CloudDescriptor();
+ cloudDesc.setCollectionName("collection1");
+ zkController.createCollectionZkNode(cloudDesc);
+
+ String shard1 = zkController.register("core1", cloudDesc);
+ String shard2 = zkController.register("core2", cloudDesc);
+ String shard3 = zkController.register("core3", cloudDesc);
+ String shard4 = zkController.register("core4", cloudDesc);
+ String shard5 = zkController.register("core5", cloudDesc);
+ String shard6 = zkController.register("core6", cloudDesc);
+
+ assertEquals("shard1", shard1);
+ assertEquals("shard2", shard2);
+ assertEquals("shard3", shard3);
+ assertEquals("shard1", shard4);
+ assertEquals("shard2", shard5);
+ assertEquals("shard3", shard6);
+
+ } finally {
+ if (DEBUG) {
+ if (zkController != null) {
+ zkController.printLayoutToStdOut();
+ }
+ }
+
+ if (zkController != null) {
+ zkController.close();
+ }
+ server.shutdown();
+ }
+
+ }
private void addShardToZk(SolrZkClient zkClient, String shardsPath,
String zkNodeName, String url) throws IOException,
Added: lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/ZkLockTest.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/ZkLockTest.java?rev=1177057&view=auto
==============================================================================
--- lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/ZkLockTest.java (added)
+++ lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/ZkLockTest.java Wed Sep 28 20:49:15 2011
@@ -0,0 +1,97 @@
+package org.apache.solr.cloud;
+
+/**
+ * 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.
+ */
+
+
+import java.io.File;
+
+import org.apache.solr.SolrTestCaseJ4;
+import org.apache.solr.cloud.lock.LockListener;
+import org.apache.solr.cloud.lock.WriteLock;
+import org.apache.solr.common.cloud.SolrZkClient;
+import org.apache.solr.core.SolrConfig;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+public class ZkLockTest extends SolrTestCaseJ4 {
+
+ static final int TIMEOUT = 10000;
+ private ZkTestServer server;
+ private SolrZkClient zkClient;
+
+ @BeforeClass
+ public static void beforeClass() throws Exception {
+ createTempDir();
+ }
+
+ @Override
+ public void setUp() throws Exception {
+ super.setUp();
+ String zkDir = dataDir.getAbsolutePath() + File.separator
+ + "zookeeper/server1/data";
+
+ server = new ZkTestServer(zkDir);
+ server.run();
+ AbstractZkTestCase.tryCleanSolrZkNode(server.getZkHost());
+ AbstractZkTestCase.makeSolrZkNode(server.getZkHost());
+ zkClient = new SolrZkClient(server.getZkAddress(), TIMEOUT);
+ }
+
+ @Test
+ public void testLock() throws Exception {
+ zkClient.makePath("/collections/collection1/shards_lock");
+ zkClient.makePath("/collections/collection1/shards_lock2");
+ zkClient.makePath("/collections/collection1/shards_lock3");
+ zkClient.getData("/collections/collection1/shards_lock", null, null);
+
+ WriteLock lock = new WriteLock(zkClient.getSolrZooKeeper(),
+ "/collections/collection1/shards_lock", null, new LockListener() {
+
+ @Override
+ public void lockReleased() {
+ // TODO Auto-generated method stub
+
+ }
+
+ @Override
+ public void lockAcquired() {
+ System.out.println("I got the lock!");
+
+ }
+ });
+
+ lock.lock();
+ System.out.println("zk:");
+ printLayout(server.getZkHost());
+ }
+
+ @Override
+ public void tearDown() throws Exception {
+ zkClient.close();
+ server.shutdown();
+ SolrConfig.severeErrors.clear();
+ super.tearDown();
+ }
+
+ private void printLayout(String zkHost) throws Exception {
+ SolrZkClient zkClient = new SolrZkClient(zkHost, AbstractZkTestCase.TIMEOUT);
+ zkClient.printLayoutToStdOut();
+ zkClient.close();
+ }
+
+}
Added: lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/lock/WriteLockTest.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/lock/WriteLockTest.java?rev=1177057&view=auto
==============================================================================
--- lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/lock/WriteLockTest.java (added)
+++ lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/lock/WriteLockTest.java Wed Sep 28 20:49:15 2011
@@ -0,0 +1,184 @@
+package org.apache.solr.cloud.lock;
+
+/**
+ * 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.
+ */
+
+import java.io.File;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.solr.SolrTestCaseJ4;
+import org.apache.solr.cloud.AbstractZkTestCase;
+import org.apache.solr.cloud.ZkTestServer;
+import org.apache.solr.common.cloud.SolrZkClient;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+/**
+ * test for writelock
+ */
+public class WriteLockTest extends SolrTestCaseJ4 {
+ private static int TIMEOUT = 10 * 1000;
+ protected String dir = "/" + getClass().getName();
+ protected WriteLock[] nodes;
+ protected CountDownLatch latch = new CountDownLatch(1);
+ private boolean restartServer = true;
+ private boolean workAroundClosingLastZNodeFails = true;
+ private boolean killLeader = true;
+ private ZkTestServer server;
+ private SolrZkClient zkClient;
+ private String zkDir;
+
+ @BeforeClass
+ public static void beforeClass() throws Exception {
+ createTempDir();
+ }
+
+ @Override
+ public void setUp() throws Exception {
+ super.setUp();
+ zkDir = dataDir.getAbsolutePath() + File.separator
+ + "zookeeper/server1/data";
+
+ server = new ZkTestServer(zkDir);
+ server.run();
+ AbstractZkTestCase.tryCleanSolrZkNode(server.getZkHost());
+ AbstractZkTestCase.makeSolrZkNode(server.getZkHost());
+ zkClient = new SolrZkClient(server.getZkAddress(), TIMEOUT);
+ }
+
+ @Test
+ public void runTest() throws Exception {
+ doTest(3);
+ }
+
+ class LockCallback implements LockListener {
+ public void lockAcquired() {
+ latch.countDown();
+ }
+
+ public void lockReleased() {
+
+ }
+
+ }
+
+ protected void doTest(int count) throws Exception {
+ nodes = new WriteLock[count];
+ for (int i = 0; i < count; i++) {
+ WriteLock leader = new WriteLock(zkClient.getSolrZooKeeper(), dir, null);
+ leader.setLockListener(new LockCallback());
+ nodes[i] = leader;
+
+ leader.lock();
+ }
+
+ // lets wait for any previous leaders to die and one of our new
+ // nodes to become the new leader
+ latch.await(30, TimeUnit.SECONDS);
+
+ WriteLock first = nodes[0];
+ dumpNodes(count);
+
+ // lets assert that the first election is the leader
+ assertTrue("The first znode should be the leader " + first.getId(),
+ first.isOwner());
+
+ for (int i = 1; i < count; i++) {
+ WriteLock node = nodes[i];
+ assertFalse("Node should not be the leader " + node.getId(),
+ node.isOwner());
+ }
+
+ if (count > 1) {
+ if (killLeader) {
+ System.out.println("Now killing the leader");
+ // now lets kill the leader
+ latch = new CountDownLatch(1);
+ first.unlock();
+ latch.await(30, TimeUnit.SECONDS);
+ // Thread.sleep(10000);
+ WriteLock second = nodes[1];
+ dumpNodes(count);
+ // lets assert that the first election is the leader
+ assertTrue("The second znode should be the leader " + second.getId(),
+ second.isOwner());
+
+ for (int i = 2; i < count; i++) {
+ WriteLock node = nodes[i];
+ assertFalse("Node should not be the leader " + node.getId(),
+ node.isOwner());
+ }
+ }
+
+ if (restartServer) {
+ // now lets stop the server
+ System.out.println("Now stopping the server");
+ server.shutdown();
+ Thread.sleep(10000);
+
+ // TODO lets assert that we are no longer the leader
+ dumpNodes(count);
+
+ Thread.sleep(300);
+
+ // try a reconnect from disconnect
+ server = new ZkTestServer(zkDir);
+ server.run();
+
+ for (int i = 0; i < count - 1; i++) {
+ System.out.println("Calling acquire for node: " + i);
+ //nodes[i].lock();
+ }
+ dumpNodes(count);
+ System.out.println("Now closing down...");
+ }
+ }
+ }
+
+ protected void dumpNodes(int count) {
+ for (int i = 0; i < count; i++) {
+ WriteLock node = nodes[i];
+ System.out.println("node: " + i + " id: " + node.getId() + " is leader: "
+ + node.isOwner());
+ }
+ }
+
+ @Override
+ public void tearDown() throws Exception {
+ if (nodes != null) {
+ for (int i = 0; i < nodes.length; i++) {
+ WriteLock node = nodes[i];
+ if (node != null) {
+ System.out.println("Closing node: " + i);
+ node.close();
+ if (workAroundClosingLastZNodeFails && i == nodes.length - 1) {
+ System.out.println("Not closing zookeeper: " + i + " due to bug!");
+ } else {
+ System.out.println("Closing zookeeper: " + i);
+ node.getZookeeper().close();
+ System.out.println("Closed zookeeper: " + i);
+ }
+ }
+ }
+ }
+ zkClient.close();
+ server.shutdown();
+ super.tearDown();
+
+ }
+}