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 2012/01/25 20:49:30 UTC
svn commit: r1235888 [11/12] - in /lucene/dev/trunk: dev-tools/eclipse/
dev-tools/maven/ solr/ solr/cloud-dev/
solr/contrib/dataimporthandler/src/java/org/apache/solr/handler/dataimport/
solr/contrib/dataimporthandler/src/test/org/apache/solr/handler/d...
Modified: lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/SolrZkClient.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/SolrZkClient.java?rev=1235888&r1=1235887&r2=1235888&view=diff
==============================================================================
--- lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/SolrZkClient.java (original)
+++ lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/SolrZkClient.java Wed Jan 25 19:49:26 2012
@@ -19,18 +19,32 @@ package org.apache.solr.common.cloud;
import java.io.File;
import java.io.IOException;
+import java.io.StringReader;
+import java.io.StringWriter;
import java.io.UnsupportedEncodingException;
import java.util.List;
import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+
+import javax.xml.transform.OutputKeys;
+import javax.xml.transform.Source;
+import javax.xml.transform.Transformer;
+import javax.xml.transform.TransformerFactory;
+import javax.xml.transform.stream.StreamResult;
+import javax.xml.transform.stream.StreamSource;
import org.apache.commons.io.FileUtils;
+import org.apache.lucene.store.AlreadyClosedException;
import org.apache.solr.common.SolrException;
import org.apache.solr.common.cloud.ZkClientConnectionStrategy.ZkUpdate;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.KeeperException.NoNodeException;
+import org.apache.zookeeper.SolrZooKeeper;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.KeeperException.NodeExistsException;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
@@ -43,6 +57,10 @@ import org.slf4j.LoggerFactory;
*
*/
public class SolrZkClient {
+ // These should *only* be used for debugging or monitoring purposes
+ public static final AtomicLong numOpens = new AtomicLong();
+ public static final AtomicLong numCloses = new AtomicLong();
+
static final String NEWL = System.getProperty("line.separator");
static final int DEFAULT_CLIENT_CONNECT_TIMEOUT = 30000;
@@ -54,6 +72,10 @@ public class SolrZkClient {
private volatile SolrZooKeeper keeper;
+ private ZkCmdExecutor zkCmdExecutor = new ZkCmdExecutor();
+
+ private volatile boolean isClosed = false;
+
/**
* @param zkServerAddress
* @param zkClientTimeout
@@ -97,6 +119,7 @@ public class SolrZkClient {
public SolrZkClient(String zkServerAddress, int zkClientTimeout,
ZkClientConnectionStrategy strat, final OnReconnect onReconnect, int clientConnectTimeout) throws InterruptedException,
TimeoutException, IOException {
+ numOpens.incrementAndGet();
connManager = new ConnectionManager("ZooKeeperConnection Watcher:"
+ zkServerAddress, this, zkServerAddress, zkClientTimeout, strat, onReconnect);
strat.connect(zkServerAddress, zkClientTimeout, connManager,
@@ -131,12 +154,23 @@ public class SolrZkClient {
/**
* @param path
* @param version
+ * @param retryOnConnLoss
* @throws InterruptedException
* @throws KeeperException
*/
- public void delete(final String path, int version)
+ public void delete(final String path, final int version, boolean retryOnConnLoss)
throws InterruptedException, KeeperException {
- keeper.delete(path, version);
+ if (retryOnConnLoss) {
+ zkCmdExecutor.retryOperation(new ZkOperation() {
+ @Override
+ public Stat execute() throws KeeperException, InterruptedException {
+ keeper.delete(path, version);
+ return null;
+ }
+ });
+ } else {
+ keeper.delete(path, version);
+ }
}
/**
@@ -150,26 +184,46 @@ public class SolrZkClient {
*
* @param path the node path
* @param watcher explicit watcher
+ * @param retryOnConnLoss
* @return the stat of the node of the given path; return null if no such a
* node exists.
* @throws KeeperException If the server signals an error
* @throws InterruptedException If the server transaction is interrupted.
* @throws IllegalArgumentException if an invalid path is specified
*/
- public Stat exists(final String path, Watcher watcher)
+ public Stat exists(final String path, final Watcher watcher, boolean retryOnConnLoss)
throws KeeperException, InterruptedException {
- return keeper.exists(path, watcher);
+ if (retryOnConnLoss) {
+ return zkCmdExecutor.retryOperation(new ZkOperation() {
+ @Override
+ public Stat execute() throws KeeperException, InterruptedException {
+ return keeper.exists(path, watcher);
+ }
+ });
+ } else {
+ return keeper.exists(path, watcher);
+ }
}
/**
* @param path
* @return true if path exists
* @throws KeeperException
+ * @param retryOnConnLoss
* @throws InterruptedException
*/
- public boolean exists(final String path)
+ public Boolean exists(final String path, boolean retryOnConnLoss)
throws KeeperException, InterruptedException {
- return keeper.exists(path, null) != null;
+ if (retryOnConnLoss) {
+ return zkCmdExecutor.retryOperation(new ZkOperation() {
+ @Override
+ public Boolean execute() throws KeeperException, InterruptedException {
+ return keeper.exists(path, null) != null;
+ }
+ });
+ } else {
+ return keeper.exists(path, null) != null;
+ }
}
/**
@@ -177,51 +231,91 @@ public class SolrZkClient {
* @param data
* @param acl
* @param createMode
+ * @param retryOnConnLoss
* @return path of created node
* @throws KeeperException
* @throws InterruptedException
*/
- public String create(final String path, byte data[], List<ACL> acl,
- CreateMode createMode) throws KeeperException, InterruptedException {
- return keeper.create(path, data, acl, createMode);
+ public String create(final String path, final byte data[], final List<ACL> acl,
+ final CreateMode createMode, boolean retryOnConnLoss) throws KeeperException, InterruptedException {
+ if (retryOnConnLoss) {
+ return zkCmdExecutor.retryOperation(new ZkOperation() {
+ @Override
+ public String execute() throws KeeperException, InterruptedException {
+ return keeper.create(path, data, acl, createMode);
+ }
+ });
+ } else {
+ return keeper.create(path, data, acl, createMode);
+ }
}
/**
* @param path
* @param watcher
+ * @param retryOnConnLoss
* @return children of the node at the path
* @throws KeeperException
* @throws InterruptedException
*/
- public List<String> getChildren(final String path, Watcher watcher)
+ public List<String> getChildren(final String path, final Watcher watcher, boolean retryOnConnLoss)
throws KeeperException, InterruptedException {
- return keeper.getChildren(path, watcher);
+ if (retryOnConnLoss) {
+ return zkCmdExecutor.retryOperation(new ZkOperation() {
+ @Override
+ public List<String> execute() throws KeeperException, InterruptedException {
+ return keeper.getChildren(path, watcher);
+ }
+ });
+ } else {
+ return keeper.getChildren(path, watcher);
+ }
}
/**
* @param path
* @param watcher
* @param stat
+ * @param retryOnConnLoss
* @return node's data
* @throws KeeperException
* @throws InterruptedException
*/
- public byte[] getData(final String path, Watcher watcher, Stat stat)
+ public byte[] getData(final String path, final Watcher watcher, final Stat stat, boolean retryOnConnLoss)
throws KeeperException, InterruptedException {
- return keeper.getData(path, watcher, stat);
+ if (retryOnConnLoss) {
+ return zkCmdExecutor.retryOperation(new ZkOperation() {
+ @Override
+ public byte[] execute() throws KeeperException, InterruptedException {
+ return keeper.getData(path, watcher, stat);
+ }
+ });
+ } else {
+ return keeper.getData(path, watcher, stat);
+ }
}
/**
* @param path
* @param data
* @param version
+ * @param retryOnConnLoss
* @return node's state
* @throws KeeperException
* @throws InterruptedException
*/
- public Stat setData(final String path, byte data[], int version)
+ public Stat setData(final String path, final byte data[], final int version, boolean retryOnConnLoss)
throws KeeperException, InterruptedException {
- return keeper.setData(path, data, version);
+ if (retryOnConnLoss) {
+ return zkCmdExecutor.retryOperation(new ZkOperation() {
+ @Override
+ public Stat execute() throws KeeperException, InterruptedException {
+ return keeper.setData(path, data, version);
+ }
+ });
+ } else {
+ return keeper.setData(path, data, version);
+ }
}
/**
@@ -230,14 +324,24 @@ public class SolrZkClient {
* @param data
* @param createMode
* @return path of created node
+ * @param retryOnConnLoss
* @throws KeeperException
* @throws InterruptedException
*/
- public String create(String path, byte[] data, CreateMode createMode) throws KeeperException, InterruptedException {
-
- String zkPath = keeper.create(path, data, ZooDefs.Ids.OPEN_ACL_UNSAFE, createMode);
-
- return zkPath;
+ public String create(final String path, final byte[] data,
+ final CreateMode createMode, boolean retryOnConnLoss) throws KeeperException,
+ InterruptedException {
+ if (retryOnConnLoss) {
+ return zkCmdExecutor.retryOperation(new ZkOperation() {
+ @Override
+ public String execute() throws KeeperException, InterruptedException {
+ return keeper.create(path, data, ZooDefs.Ids.OPEN_ACL_UNSAFE,
+ createMode);
+ }
+ });
+ } else {
+ return keeper.create(path, data, ZooDefs.Ids.OPEN_ACL_UNSAFE, createMode);
+ }
}
/**
@@ -247,17 +351,34 @@ public class SolrZkClient {
* group, node exist, each will be created.
*
* @param path
+ * @param retryOnConnLoss
* @throws KeeperException
* @throws InterruptedException
*/
- public void makePath(String path) throws KeeperException,
+ public void makePath(String path, boolean retryOnConnLoss) throws KeeperException,
InterruptedException {
- makePath(path, null, CreateMode.PERSISTENT);
+ makePath(path, null, CreateMode.PERSISTENT, retryOnConnLoss);
}
- public void makePath(String path, CreateMode createMode) throws KeeperException,
+ public void makePath(String path, boolean failOnExists, boolean retryOnConnLoss) throws KeeperException,
InterruptedException {
- makePath(path, null, createMode);
+ makePath(path, null, CreateMode.PERSISTENT, null, failOnExists, retryOnConnLoss);
+ }
+
+ public void makePath(String path, File file, boolean failOnExists, boolean retryOnConnLoss)
+ throws IOException, KeeperException, InterruptedException {
+ makePath(path, FileUtils.readFileToString(file).getBytes("UTF-8"),
+ CreateMode.PERSISTENT, null, failOnExists, retryOnConnLoss);
+ }
+
+ public void makePath(String path, File file, boolean retryOnConnLoss) throws IOException,
+ KeeperException, InterruptedException {
+ makePath(path, FileUtils.readFileToString(file).getBytes("UTF-8"), retryOnConnLoss);
+ }
+
+ public void makePath(String path, CreateMode createMode, boolean retryOnConnLoss) throws KeeperException,
+ InterruptedException {
+ makePath(path, null, createMode, retryOnConnLoss);
}
/**
@@ -265,12 +386,13 @@ public class SolrZkClient {
*
* @param path
* @param data to set on the last zkNode
+ * @param retryOnConnLoss
* @throws KeeperException
* @throws InterruptedException
*/
- public void makePath(String path, byte[] data) throws KeeperException,
+ public void makePath(String path, byte[] data, boolean retryOnConnLoss) throws KeeperException,
InterruptedException {
- makePath(path, data, CreateMode.PERSISTENT);
+ makePath(path, data, CreateMode.PERSISTENT, retryOnConnLoss);
}
/**
@@ -282,12 +404,13 @@ public class SolrZkClient {
* @param path
* @param data to set on the last zkNode
* @param createMode
+ * @param retryOnConnLoss
* @throws KeeperException
* @throws InterruptedException
*/
- public void makePath(String path, byte[] data, CreateMode createMode)
+ public void makePath(String path, byte[] data, CreateMode createMode, boolean retryOnConnLoss)
throws KeeperException, InterruptedException {
- makePath(path, data, createMode, null);
+ makePath(path, data, createMode, null, retryOnConnLoss);
}
/**
@@ -300,32 +423,41 @@ public class SolrZkClient {
* @param data to set on the last zkNode
* @param createMode
* @param watcher
+ * @param retryOnConnLoss
* @throws KeeperException
* @throws InterruptedException
*/
public void makePath(String path, byte[] data, CreateMode createMode,
- Watcher watcher) throws KeeperException, InterruptedException {
- makePath(path, data, createMode, watcher, false);
+ Watcher watcher, boolean retryOnConnLoss) throws KeeperException, InterruptedException {
+ makePath(path, data, createMode, watcher, true, retryOnConnLoss);
}
+
+
/**
* Creates the path in ZooKeeper, creating each node as necessary.
*
* e.g. If <code>path=/solr/group/node</code> and none of the nodes, solr,
* group, node exist, each will be created.
*
+ * Note: retryOnConnLoss is only respected for the final node - nodes
+ * before that are always retried on connection loss.
+ *
* @param path
- * @param data to set on the last zkNode
+ * @param data
* @param createMode
* @param watcher
+ * @param failOnExists
+ * @param retryOnConnLoss
* @throws KeeperException
* @throws InterruptedException
*/
public void makePath(String path, byte[] data, CreateMode createMode,
- Watcher watcher, boolean failOnExists) throws KeeperException, InterruptedException {
+ Watcher watcher, boolean failOnExists, boolean retryOnConnLoss) throws KeeperException, InterruptedException {
if (log.isInfoEnabled()) {
log.info("makePath: " + path);
}
+ boolean retry = true;
if (path.startsWith("/")) {
path = path.substring(1, path.length());
@@ -336,24 +468,44 @@ public class SolrZkClient {
byte[] bytes = null;
String pathPiece = paths[i];
sbPath.append("/" + pathPiece);
- String currentPath = sbPath.toString();
- Object exists = exists(currentPath, watcher);
+ final String currentPath = sbPath.toString();
+ Object exists = exists(currentPath, watcher, retryOnConnLoss);
if (exists == null || ((i == paths.length -1) && failOnExists)) {
CreateMode mode = CreateMode.PERSISTENT;
if (i == paths.length - 1) {
mode = createMode;
bytes = data;
+ if (!retryOnConnLoss) retry = false;
+ }
+ try {
+ if (retry) {
+ final CreateMode finalMode = mode;
+ final byte[] finalBytes = bytes;
+ zkCmdExecutor.retryOperation(new ZkOperation() {
+ @Override
+ public Object execute() throws KeeperException, InterruptedException {
+ keeper.create(currentPath, finalBytes, ZooDefs.Ids.OPEN_ACL_UNSAFE, finalMode);
+ return null;
+ }
+ });
+ } else {
+ keeper.create(currentPath, bytes, ZooDefs.Ids.OPEN_ACL_UNSAFE, mode);
+ }
+ } catch (NodeExistsException e) {
+ // ignore unless it's the last node in the path
+ if (i == paths.length - 1) {
+ throw e;
+ }
}
- keeper.create(currentPath, bytes, ZooDefs.Ids.OPEN_ACL_UNSAFE, mode);
if(i == paths.length -1) {
// set new watch
- exists(currentPath, watcher);
+ exists(currentPath, watcher, retryOnConnLoss);
}
} else if (i == paths.length - 1) {
// TODO: version ? for now, don't worry about race
- setData(currentPath, data, -1);
+ setData(currentPath, data, -1, retryOnConnLoss);
// set new watch
- exists(currentPath, watcher);
+ exists(currentPath, watcher, retryOnConnLoss);
}
}
}
@@ -362,12 +514,13 @@ public class SolrZkClient {
* @param zkPath
* @param createMode
* @param watcher
+ * @param retryOnConnLoss
* @throws KeeperException
* @throws InterruptedException
*/
- public void makePath(String zkPath, CreateMode createMode, Watcher watcher)
+ public void makePath(String zkPath, CreateMode createMode, Watcher watcher, boolean retryOnConnLoss)
throws KeeperException, InterruptedException {
- makePath(zkPath, null, createMode, watcher);
+ makePath(zkPath, null, createMode, watcher, retryOnConnLoss);
}
/**
@@ -375,20 +528,13 @@ public class SolrZkClient {
*
* @param path
* @param data
+ * @param retryOnConnLoss
* @throws KeeperException
* @throws InterruptedException
*/
- public void setData(String path, byte[] data) throws KeeperException,
+ public void setData(String path, byte[] data, boolean retryOnConnLoss) throws KeeperException,
InterruptedException {
-
- makePath(path);
-
- Object exists = exists(path, null);
- if (exists != null) {
- setData(path, data, -1);
- } else {
- create(path, data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
- }
+ setData(path, data, -1, retryOnConnLoss);
}
/**
@@ -396,18 +542,19 @@ public class SolrZkClient {
*
* @param path path to upload file to e.g. /solr/conf/solrconfig.xml
* @param file path to file to be uploaded
+ * @param retryOnConnLoss
* @throws IOException
* @throws KeeperException
* @throws InterruptedException
*/
- public void setData(String path, File file) throws IOException,
+ public void setData(String path, File file, boolean retryOnConnLoss) throws IOException,
KeeperException, InterruptedException {
if (log.isInfoEnabled()) {
log.info("Write to ZooKeepeer " + file.getAbsolutePath() + " to " + path);
}
String data = FileUtils.readFileToString(file);
- setData(path, data.getBytes("UTF-8"));
+ setData(path, data.getBytes("UTF-8"), retryOnConnLoss);
}
/**
@@ -420,8 +567,8 @@ public class SolrZkClient {
*/
public void printLayout(String path, int indent, StringBuilder string)
throws KeeperException, InterruptedException {
- byte[] data = getData(path, null, null);
- List<String> children = getChildren(path, null);
+ byte[] data = getData(path, null, null, true);
+ List<String> children = getChildren(path, null, true);
StringBuilder dent = new StringBuilder();
for (int i = 0; i < indent; i++) {
dent.append(" ");
@@ -430,7 +577,12 @@ public class SolrZkClient {
if (data != null) {
try {
String dataString = new String(data, "UTF-8");
- if (!path.endsWith(".txt") && !path.endsWith(".xml")) {
+ if ((!path.endsWith(".txt") && !path.endsWith(".xml")) || path.endsWith(ZkStateReader.CLUSTER_STATE)) {
+ if (path.endsWith(".xml")) {
+ // this is the cluster state in xml format - lets pretty print
+ dataString = prettyPrint(dataString);
+ }
+
string.append(dent + "DATA:\n" + dent + " "
+ dataString.replaceAll("\n", "\n" + dent + " ") + NEWL);
} else {
@@ -444,8 +596,12 @@ public class SolrZkClient {
for (String child : children) {
if (!child.equals("quota")) {
- printLayout(path + (path.equals("/") ? "" : "/") + child, indent + 1,
- string);
+ try {
+ printLayout(path + (path.equals("/") ? "" : "/") + child, indent + 1,
+ string);
+ } catch (NoNodeException e) {
+ // must have gone away
+ }
}
}
@@ -463,12 +619,41 @@ public class SolrZkClient {
printLayout("/", 0, sb);
System.out.println(sb.toString());
}
+
+ public static String prettyPrint(String input, int indent) {
+ try {
+ Source xmlInput = new StreamSource(new StringReader(input));
+ StringWriter stringWriter = new StringWriter();
+ StreamResult xmlOutput = new StreamResult(stringWriter);
+ TransformerFactory transformerFactory = TransformerFactory.newInstance();
+ transformerFactory.setAttribute("indent-number", indent);
+ Transformer transformer = transformerFactory.newTransformer();
+ transformer.setOutputProperty(OutputKeys.INDENT, "yes");
+ transformer.transform(xmlInput, xmlOutput);
+ return xmlOutput.getWriter().toString();
+ } catch (Exception e) {
+ throw new RuntimeException("Problem pretty printing XML", e);
+ }
+ }
+
+ private static String prettyPrint(String input) {
+ return prettyPrint(input, 2);
+ }
/**
* @throws InterruptedException
*/
public void close() throws InterruptedException {
+ if (isClosed) {
+ throw new AlreadyClosedException("This client has already been closed");
+ }
+ isClosed = true;
keeper.close();
+ numCloses.incrementAndGet();
+ }
+
+ public boolean isClosed() {
+ return isClosed;
}
/**
Modified: lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ZkClientConnectionStrategy.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ZkClientConnectionStrategy.java?rev=1235888&r1=1235887&r2=1235888&view=diff
==============================================================================
--- lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ZkClientConnectionStrategy.java (original)
+++ lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ZkClientConnectionStrategy.java Wed Jan 25 19:49:26 2012
@@ -20,6 +20,7 @@ package org.apache.solr.common.cloud;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
+import org.apache.zookeeper.SolrZooKeeper;
import org.apache.zookeeper.Watcher;
/**
Added: lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ZkCmdExecutor.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ZkCmdExecutor.java?rev=1235888&view=auto
==============================================================================
--- lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ZkCmdExecutor.java (added)
+++ lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ZkCmdExecutor.java Wed Jan 25 19:49:26 2012
@@ -0,0 +1,112 @@
+package org.apache.solr.common.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.IOException;
+import java.util.List;
+
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.KeeperException.NodeExistsException;
+import org.apache.zookeeper.ZooDefs;
+import org.apache.zookeeper.data.ACL;
+
+
+public class ZkCmdExecutor {
+ private long retryDelay = 1000L;
+ private int retryCount = 15;
+ private List<ACL> acl = ZooDefs.Ids.OPEN_ACL_UNSAFE;
+
+ public ZkCmdExecutor() {
+ }
+
+ public List<ACL> getAcl() {
+ return acl;
+ }
+
+ public void setAcl(List<ACL> acl) {
+ this.acl = acl;
+ }
+
+ public long getRetryDelay() {
+ return retryDelay;
+ }
+
+ public void setRetryDelay(long retryDelay) {
+ this.retryDelay = retryDelay;
+ }
+
+ /**
+ * Perform the given operation, retrying if the connection fails
+ *
+ * @return
+ * @throws IOException
+ */
+ @SuppressWarnings("unchecked")
+ public <T> T retryOperation(ZkOperation operation)
+ throws KeeperException, InterruptedException {
+ KeeperException exception = null;
+ for (int i = 0; i < retryCount; i++) {
+ try {
+ return (T) operation.execute();
+ } catch (KeeperException.ConnectionLossException e) {
+ if (exception == null) {
+ exception = e;
+ }
+ if (Thread.currentThread().isInterrupted()) {
+ Thread.currentThread().interrupt();
+ throw new InterruptedException();
+ }
+ retryDelay(i);
+ }
+ }
+ throw exception;
+ }
+
+ public void ensureExists(String path, final SolrZkClient zkClient) throws KeeperException, InterruptedException {
+ ensureExists(path, null, CreateMode.PERSISTENT, zkClient);
+ }
+
+ public void ensureExists(final String path, final byte[] data,
+ CreateMode createMode, final SolrZkClient zkClient) throws KeeperException, InterruptedException {
+
+ if (zkClient.exists(path, true)) {
+ return;
+ }
+ try {
+ zkClient.makePath(path, data, true);
+ } catch (NodeExistsException e) {
+ // its okay if another beats us creating the node
+ }
+
+ }
+
+ /**
+ * Performs a retry delay if this is not the first attempt
+ *
+ * @param attemptCount
+ * the number of the attempts performed so far
+ * @throws InterruptedException
+ */
+ protected void retryDelay(int attemptCount) throws InterruptedException {
+ if (attemptCount > 0) {
+ Thread.sleep(attemptCount * retryDelay);
+ }
+ }
+
+}
Added: lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ZkCoreNodeProps.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ZkCoreNodeProps.java?rev=1235888&view=auto
==============================================================================
--- lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ZkCoreNodeProps.java (added)
+++ lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ZkCoreNodeProps.java Wed Jan 25 19:49:26 2012
@@ -0,0 +1,74 @@
+package org.apache.solr.common.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.
+ */
+
+public class ZkCoreNodeProps {
+ private ZkNodeProps nodeProps;
+
+ public ZkCoreNodeProps(ZkNodeProps nodeProps) {
+ this.nodeProps = nodeProps;
+ }
+
+ // may return null
+ public String getCoreUrl() {
+ return getCoreUrl(nodeProps.get(ZkStateReader.BASE_URL_PROP), nodeProps.get(ZkStateReader.CORE_NAME_PROP));
+ }
+
+ public String getNodeName() {
+ return nodeProps.get(ZkStateReader.NODE_NAME_PROP);
+ }
+
+ public String getState() {
+ return nodeProps.get(ZkStateReader.STATE_PROP);
+ }
+
+ public String getBaseUrl() {
+ return nodeProps.get(ZkStateReader.BASE_URL_PROP);
+ }
+
+ public String getCoreName() {
+ return nodeProps.get(ZkStateReader.CORE_NAME_PROP);
+ }
+
+ public static String getCoreUrl(String baseUrl, String coreName) {
+ StringBuilder sb = new StringBuilder();
+ if (baseUrl == null) return null;
+ if (coreName == null) return null;
+ sb.append(baseUrl);
+ if (!baseUrl.endsWith("/")) sb.append("/");
+ sb.append(coreName == null ? "" : coreName);
+ if (!(sb.substring(sb.length() - 1).equals("/"))) sb.append("/");
+
+ return sb.toString();
+ }
+
+ @Override
+ public String toString() {
+ return nodeProps.toString();
+ }
+
+ public String getCoreNodeName() {
+ return getNodeName() + "_" + getCoreName();
+ }
+
+ public ZkNodeProps getNodeProps() {
+ return nodeProps;
+ }
+
+
+}
Modified: lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ZkNodeProps.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ZkNodeProps.java?rev=1235888&r1=1235887&r2=1235888&view=diff
==============================================================================
--- lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ZkNodeProps.java (original)
+++ lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ZkNodeProps.java Wed Jan 25 19:49:26 2012
@@ -17,43 +17,77 @@ package org.apache.solr.common.cloud;
* limitations under the License.
*/
-import java.io.IOException;
+import java.util.Collections;
import java.util.HashMap;
+import java.util.Map;
import java.util.Set;
import java.util.Map.Entry;
-public class ZkNodeProps extends HashMap<String,String> {
+import org.apache.noggit.JSONWriter;
- private static final long serialVersionUID = 1L;
+// Immutable
+public class ZkNodeProps implements JSONWriter.Writable {
- public void load(byte[] bytes) throws IOException {
- String stringRep = new String(bytes, "UTF-8");
- String[] lines = stringRep.split("\n");
- for (String line : lines) {
- int sepIndex = line.indexOf('=');
- String key = line.substring(0, sepIndex);
- String value = line.substring(sepIndex + 1, line.length());
- put(key, value);
+ private final Map<String,String> propMap;
+
+ public ZkNodeProps(Map<String,String> propMap) {
+ this.propMap = new HashMap<String,String>();
+ this.propMap.putAll(propMap);
+ }
+
+ public ZkNodeProps(ZkNodeProps zkNodeProps) {
+ this.propMap = new HashMap<String,String>();
+ this.propMap.putAll(zkNodeProps.propMap);
+ }
+
+ public ZkNodeProps() {
+ propMap = new HashMap<String,String>();
+ }
+
+ public ZkNodeProps(String... keyVals) {
+ if (keyVals.length % 2 != 0) {
+ throw new IllegalArgumentException("arguments should be key,value");
+ }
+ propMap = new HashMap<String,String>();
+ for (int i = 0; i < keyVals.length; i+=2) {
+ propMap.put(keyVals[i], keyVals[i+1]);
}
}
+
+ public Set<String> keySet() {
+ return Collections.unmodifiableSet(propMap.keySet());
+ }
- public byte[] store() throws IOException {
- StringBuilder sb = new StringBuilder();
- Set<Entry<String,String>> entries = entrySet();
- for(Entry<String,String> entry : entries) {
- sb.append(entry.getKey() + "=" + entry.getValue() + "\n");
- }
- return sb.toString().getBytes("UTF-8");
+ public Map<String,String> getProperties() {
+ return Collections.unmodifiableMap(propMap);
+ }
+
+ public static ZkNodeProps load(byte[] bytes) {
+ Map<String, String> props = (Map<String, String>) ZkStateReader.fromJSON(bytes);
+ return new ZkNodeProps(props);
+ }
+
+ @Override
+ public void write(JSONWriter jsonWriter) {
+ jsonWriter.write(propMap);
+ }
+
+ public String get(String key) {
+ return propMap.get(key);
}
@Override
public String toString() {
StringBuilder sb = new StringBuilder();
- Set<Entry<String,String>> entries = entrySet();
+ Set<Entry<String,String>> entries = propMap.entrySet();
for(Entry<String,String> entry : entries) {
sb.append(entry.getKey() + "=" + entry.getValue() + "\n");
}
return sb.toString();
}
+
+ public boolean containsKey(String key) {
+ return propMap.containsKey(key);
+ }
}
Added: lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ZkOperation.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ZkOperation.java?rev=1235888&view=auto
==============================================================================
--- lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ZkOperation.java (added)
+++ lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ZkOperation.java Wed Jan 25 19:49:26 2012
@@ -0,0 +1,41 @@
+/**
+ *
+ * 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.common.cloud;
+
+import java.io.IOException;
+
+import org.apache.zookeeper.KeeperException;
+
+/**
+ * A callback object which can be used for implementing retry-able operations in the
+ * {@link org.apache.solr.common.cloud.ZkCmdExecutor.lock.ProtocolSupport} class
+ *
+ */
+public abstract class ZkOperation {
+
+ /**
+ * 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
+ * @throws IOException
+ */
+ public abstract Object execute() throws KeeperException, InterruptedException;
+}
Modified: lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java?rev=1235888&r1=1235887&r2=1235888&view=diff
==============================================================================
--- lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java (original)
+++ lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java Wed Jan 25 19:49:26 2012
@@ -18,10 +18,12 @@ package org.apache.solr.common.cloud;
*/
import java.io.IOException;
-import java.util.HashMap;
+import java.util.ArrayList;
+import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
@@ -29,26 +31,83 @@ import java.util.concurrent.ThreadFactor
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import org.apache.lucene.util.BytesRef;
+import org.apache.lucene.util.CharsRef;
+import org.apache.lucene.util.UnicodeUtil;
+import org.apache.noggit.CharArr;
+import org.apache.noggit.JSONParser;
+import org.apache.noggit.JSONWriter;
+import org.apache.noggit.ObjectBuilder;
import org.apache.solr.common.SolrException;
+import org.apache.solr.common.SolrException.ErrorCode;
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;
public class ZkStateReader {
private static Logger log = LoggerFactory.getLogger(ZkStateReader.class);
+ public static final String BASE_URL_PROP = "base_url";
+ public static final String NODE_NAME_PROP = "node_name";
+ public static final String ROLES_PROP = "roles";
+ public static final String STATE_PROP = "state";
+ public static final String CORE_NAME_PROP = "core";
+ public static final String COLLECTION_PROP = "collection";
+ public static final String SHARD_ID_PROP = "shard_id";
+ public static final String NUM_SHARDS_PROP = "numShards";
+ public static final String LEADER_PROP = "leader";
+
public static final String COLLECTIONS_ZKNODE = "/collections";
- public static final String URL_PROP = "url";
- public static final String NODE_NAME = "node_name";
- public static final String SHARDS_ZKNODE = "/shards";
public static final String LIVE_NODES_ZKNODE = "/live_nodes";
+ public static final String CLUSTER_STATE = "/clusterstate.json";
+
+ public static final String RECOVERING = "recovering";
+ public static final String RECOVERY_FAILED = "recovery_failed";
+ public static final String ACTIVE = "active";
+ public static final String DOWN = "down";
+ public static final String SYNC = "sync";
- private volatile CloudState cloudState = new CloudState(new HashSet<String>(0), new HashMap<String,Map<String,Slice>>(0));
+ private volatile CloudState cloudState;
+
+ private static final long SOLRCLOUD_UPDATE_DELAY = Long.parseLong(System.getProperty("solrcloud.update.delay", "5000"));
+
+ public static final String LEADER_ELECT_ZKNODE = "/leader_elect";
+
+ public static final String SHARD_LEADERS_ZKNODE = "leaders";
+
+
+
- private static final long CLOUD_UPDATE_DELAY = Long.parseLong(System.getProperty("CLOUD_UPDATE_DELAY", "5000"));
+ //
+ // convenience methods... should these go somewhere else?
+ //
+ public static byte[] toJSON(Object o) {
+ CharArr out = new CharArr();
+ new JSONWriter(out, 2).write(o); // indentation by default
+ return toUTF8(out);
+ }
+
+ public static byte[] toUTF8(CharArr out) {
+ BytesRef br = new BytesRef(out);
+ return Arrays.copyOf(br.bytes, br.length);
+ }
+
+ public static Object fromJSON(byte[] utf8) {
+ // convert directly from bytes to chars
+ // and parse directly from that instead of going through
+ // intermediate strings or readers
+ CharsRef chars = new CharsRef();
+ UnicodeUtil.UTF8toUTF16(utf8, 0, utf8.length, chars); // TODO: this method currently oversizes the array
+ JSONParser parser = new JSONParser(chars.chars, chars.offset, chars.length);
+ try {
+ return ObjectBuilder.getVal(parser);
+ } catch (IOException e) {
+ throw new RuntimeException(e); // should never happen w/o using real IO
+ }
+ }
+
private static class ZKTF implements ThreadFactory {
private static ThreadGroup tg = new ThreadGroup("ZkStateReader");
@@ -66,6 +125,8 @@ public class ZkStateReader {
private SolrZkClient zkClient;
private boolean closeClient = false;
+
+ private ZkCmdExecutor cmdExecutor = new ZkCmdExecutor();
public ZkStateReader(SolrZkClient zkClient) {
this.zkClient = zkClient;
@@ -79,9 +140,7 @@ public class ZkStateReader {
public void command() {
try {
- makeCollectionsNodeWatches();
- makeShardsWatches(true);
- updateCloudState(false);
+ ZkStateReader.this.createClusterStateWatchersAndUpdate();
} catch (KeeperException e) {
log.error("", e);
throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR,
@@ -92,46 +151,146 @@ public class ZkStateReader {
log.error("", e);
throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR,
"", e);
- } catch (IOException e) {
- log.error("", e);
- throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR,
- "", e);
- }
+ }
}
});
}
// load and publish a new CollectionInfo
- public void updateCloudState(boolean immediate) throws KeeperException, InterruptedException,
- IOException {
+ public void updateCloudState(boolean immediate) throws KeeperException, InterruptedException {
updateCloudState(immediate, false);
}
// load and publish a new CollectionInfo
- public void updateLiveNodes() throws KeeperException, InterruptedException,
- IOException {
+ public void updateLiveNodes() throws KeeperException, InterruptedException {
updateCloudState(true, true);
}
- // load and publish a new CollectionInfo
- private synchronized void updateCloudState(boolean immediate, final boolean onlyLiveNodes) throws KeeperException, InterruptedException,
- IOException {
+ public synchronized void createClusterStateWatchersAndUpdate() throws KeeperException,
+ InterruptedException {
+ // We need to fetch the current cluster state and the set of live nodes
+
+ synchronized (getUpdateLock()) {
+ cmdExecutor.ensureExists(CLUSTER_STATE, zkClient);
+
+ log.info("Updating cluster state from ZooKeeper... ");
+
+ zkClient.exists(CLUSTER_STATE, new Watcher() {
+
+ @Override
+ public void process(WatchedEvent event) {
+ log.info("A cluster state change has occurred");
+ try {
+
+ // delayed approach
+ // ZkStateReader.this.updateCloudState(false, false);
+ synchronized (ZkStateReader.this.getUpdateLock()) {
+ // remake watch
+ final Watcher thisWatch = this;
+ byte[] data = zkClient.getData(CLUSTER_STATE, thisWatch, null,
+ true);
+
+ CloudState clusterState = CloudState.load(data,
+ ZkStateReader.this.cloudState.getLiveNodes());
+ // update volatile
+ cloudState = clusterState;
+ }
+ } catch (KeeperException e) {
+ if (e.code() == KeeperException.Code.SESSIONEXPIRED
+ || e.code() == KeeperException.Code.CONNECTIONLOSS) {
+ log.warn("ZooKeeper watch triggered, but Solr cannot talk to ZK");
+ return;
+ }
+ log.error("", e);
+ throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR,
+ "", e);
+ } catch (InterruptedException e) {
+ // Restore the interrupted status
+ Thread.currentThread().interrupt();
+ log.warn("", e);
+ return;
+ }
+ }
+
+ }, true);
+ }
+
+
+ synchronized (ZkStateReader.this.getUpdateLock()) {
+ List<String> liveNodes = zkClient.getChildren(LIVE_NODES_ZKNODE,
+ new Watcher() {
+
+ @Override
+ public void process(WatchedEvent event) {
+ log.info("Updating live nodes");
+ try {
+ // delayed approach
+ // ZkStateReader.this.updateCloudState(false, true);
+ synchronized (ZkStateReader.this.getUpdateLock()) {
+ List<String> liveNodes = zkClient.getChildren(
+ LIVE_NODES_ZKNODE, this, true);
+ Set<String> liveNodesSet = new HashSet<String>();
+ liveNodesSet.addAll(liveNodes);
+ CloudState clusterState = new CloudState(liveNodesSet,
+ ZkStateReader.this.cloudState.getCollectionStates());
+ ZkStateReader.this.cloudState = clusterState;
+ }
+ } catch (KeeperException e) {
+ if (e.code() == KeeperException.Code.SESSIONEXPIRED
+ || e.code() == KeeperException.Code.CONNECTIONLOSS) {
+ log.warn("ZooKeeper watch triggered, but Solr cannot talk to ZK");
+ return;
+ }
+ log.error("", e);
+ throw new ZooKeeperException(
+ SolrException.ErrorCode.SERVER_ERROR, "", e);
+ } catch (InterruptedException e) {
+ // Restore the interrupted status
+ Thread.currentThread().interrupt();
+ log.warn("", e);
+ return;
+ }
+ }
+
+ }, true);
+ Set<String> liveNodeSet = new HashSet<String>();
+ liveNodeSet.addAll(liveNodes);
+ CloudState clusterState = CloudState.load(zkClient, liveNodeSet);
+ this.cloudState = clusterState;
+ }
+ }
+
+
+ // load and publish a new CollectionInfo
+ private synchronized void updateCloudState(boolean immediate,
+ final boolean onlyLiveNodes) throws KeeperException,
+ InterruptedException {
+ log.info("Manual update of cluster state initiated");
// build immutable CloudInfo
- if(immediate) {
- if(!onlyLiveNodes) {
- log.info("Updating cloud state from ZooKeeper... ");
- } else {
- log.info("Updating live nodes from ZooKeeper... ");
+ if (immediate) {
+ CloudState clusterState;
+ synchronized (getUpdateLock()) {
+ List<String> liveNodes = zkClient.getChildren(LIVE_NODES_ZKNODE, null, true);
+ Set<String> liveNodesSet = new HashSet<String>();
+ liveNodesSet.addAll(liveNodes);
+
+ if (!onlyLiveNodes) {
+ log.info("Updating cloud state from ZooKeeper... ");
+
+ clusterState = CloudState.load(zkClient, liveNodesSet);
+ } else {
+ log.info("Updating live nodes from ZooKeeper... ");
+ clusterState = new CloudState(liveNodesSet,
+ ZkStateReader.this.cloudState.getCollectionStates());
+ }
}
- CloudState cloudState;
- cloudState = CloudState.buildCloudState(zkClient, this.cloudState, onlyLiveNodes);
- // update volatile
- this.cloudState = cloudState;
+
+ this.cloudState = clusterState;
} else {
- if(cloudStateUpdateScheduled) {
+ if (cloudStateUpdateScheduled) {
log.info("Cloud state update for ZooKeeper already scheduled");
return;
}
@@ -140,15 +299,30 @@ public class ZkStateReader {
updateCloudExecutor.schedule(new Runnable() {
public void run() {
- log.info("Updating cloud state from ZooKeeper...");
+ log.info("Updating cluster state from ZooKeeper...");
synchronized (getUpdateLock()) {
cloudStateUpdateScheduled = false;
- CloudState cloudState;
+ CloudState clusterState;
try {
- cloudState = CloudState.buildCloudState(zkClient,
- ZkStateReader.this.cloudState, onlyLiveNodes);
+ List<String> liveNodes = zkClient.getChildren(LIVE_NODES_ZKNODE,
+ null, true);
+ Set<String> liveNodesSet = new HashSet<String>();
+ liveNodesSet.addAll(liveNodes);
+
+ if (!onlyLiveNodes) {
+ log.info("Updating cloud state from ZooKeeper... ");
+
+ clusterState = CloudState.load(zkClient, liveNodesSet);
+ } else {
+ log.info("Updating live nodes from ZooKeeper... ");
+ clusterState = new CloudState(liveNodesSet, ZkStateReader.this.cloudState.getCollectionStates());
+ }
+
+ ZkStateReader.this.cloudState = clusterState;
+
} catch (KeeperException e) {
- if(e.code() == KeeperException.Code.SESSIONEXPIRED || e.code() == KeeperException.Code.CONNECTIONLOSS) {
+ if (e.code() == KeeperException.Code.SESSIONEXPIRED
+ || e.code() == KeeperException.Code.CONNECTIONLOSS) {
log.warn("ZooKeeper watch triggered, but Solr cannot talk to ZK");
return;
}
@@ -161,140 +335,16 @@ public class ZkStateReader {
log.error("", e);
throw new ZooKeeperException(
SolrException.ErrorCode.SERVER_ERROR, "", e);
- } catch (IOException e) {
- log.error("", e);
- throw new ZooKeeperException(
- SolrException.ErrorCode.SERVER_ERROR, "", e);
- }
+ }
// update volatile
ZkStateReader.this.cloudState = cloudState;
}
}
- }, CLOUD_UPDATE_DELAY, TimeUnit.MILLISECONDS);
+ }, SOLRCLOUD_UPDATE_DELAY, TimeUnit.MILLISECONDS);
}
-
- }
-
- public void makeShardZkNodeWatches(boolean makeWatchesForReconnect) throws KeeperException, InterruptedException {
- CloudState cloudState = getCloudState();
- Set<String> knownCollections = cloudState.getCollections();
- List<String> collections = zkClient.getChildren(COLLECTIONS_ZKNODE, null);
-
- for(final String collection : collections) {
- if(makeWatchesForReconnect || !knownCollections.contains(collection)) {
- log.info("Found new collection:" + collection);
- Watcher watcher = new Watcher() {
- public void process(WatchedEvent event) {
- log.info("Detected changed ShardId in collection:" + collection);
- try {
- makeShardsWatches(collection, false);
- updateCloudState(false);
- } catch (KeeperException e) {
- if(e.code() == KeeperException.Code.SESSIONEXPIRED || e.code() == KeeperException.Code.CONNECTIONLOSS) {
- log.warn("ZooKeeper watch triggered, but Solr cannot talk to ZK");
- return;
- }
- log.error("", e);
- throw new ZooKeeperException(
- SolrException.ErrorCode.SERVER_ERROR, "", e);
- } catch (InterruptedException e) {
- // Restore the interrupted status
- Thread.currentThread().interrupt();
- log.error("", e);
- throw new ZooKeeperException(
- SolrException.ErrorCode.SERVER_ERROR, "", e);
- } catch (IOException e) {
- log.error("", e);
- throw new ZooKeeperException(
- SolrException.ErrorCode.SERVER_ERROR, "", e);
- }
- }
- };
- boolean madeWatch = true;
- String shardZkNode = COLLECTIONS_ZKNODE + "/" + collection
- + SHARDS_ZKNODE;
- for (int i = 0; i < 5; i++) {
- try {
- zkClient.getChildren(shardZkNode, watcher);
- } catch (KeeperException.NoNodeException e) {
- // most likely, the collections node has been created, but not the
- // shards node yet -- pause and try again
- madeWatch = false;
- if (i == 4) {
- log.error("Could not set shards zknode watch, because the zknode does not exist:" + shardZkNode);
- break;
- }
- Thread.sleep(100);
- }
- if (madeWatch) {
- log.info("Made shard watch:" + shardZkNode);
- break;
- }
- }
- }
- }
- }
-
- public void makeShardsWatches(final String collection, boolean makeWatchesForReconnect) throws KeeperException,
- InterruptedException {
- if (zkClient.exists(COLLECTIONS_ZKNODE + "/" + collection + SHARDS_ZKNODE)) {
- List<String> shardIds = zkClient.getChildren(COLLECTIONS_ZKNODE + "/"
- + collection + SHARDS_ZKNODE, null);
- CloudState cloudState = getCloudState();
- Set<String> knownShardIds;
- Map<String,Slice> slices = cloudState.getSlices(collection);
- if (slices != null) {
- knownShardIds = slices.keySet();
- } else {
- knownShardIds = new HashSet<String>(0);
- }
- for (final String shardId : shardIds) {
- if (makeWatchesForReconnect || !knownShardIds.contains(shardId)) {
- zkClient.getChildren(COLLECTIONS_ZKNODE + "/" + collection
- + SHARDS_ZKNODE + "/" + shardId, new Watcher() {
-
- public void process(WatchedEvent event) {
- log.info("Detected a shard change under ShardId:" + shardId + " in collection:" + collection);
- try {
- updateCloudState(false);
- } catch (KeeperException e) {
- if(e.code() == KeeperException.Code.SESSIONEXPIRED || e.code() == KeeperException.Code.CONNECTIONLOSS) {
- log.warn("ZooKeeper watch triggered, but Solr cannot talk to ZK");
- return;
- }
- log.error("", e);
- throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR,
- "", e);
- } catch (InterruptedException e) {
- // Restore the interrupted status
- Thread.currentThread().interrupt();
- log.error("", e);
- throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR,
- "", e);
- } catch (IOException e) {
- log.error("", e);
- throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR,
- "", e);
- }
- }
- });
- }
- }
- }
}
-
- /**
- * @throws KeeperException
- * @throws InterruptedException
- */
- public void makeShardsWatches(boolean makeWatchesForReconnect) throws KeeperException, InterruptedException {
- List<String> collections = zkClient.getChildren(COLLECTIONS_ZKNODE, null);
- for (final String collection : collections) {
- makeShardsWatches(collection, makeWatchesForReconnect);
- }
- }
-
+
/**
* @return information about the cluster from ZooKeeper
*/
@@ -319,80 +369,96 @@ public class ZkStateReader {
}
}
}
+
+ abstract class RunnableWatcher implements Runnable {
+ Watcher watcher;
+ public RunnableWatcher(Watcher watcher){
+ this.watcher = watcher;
+ }
- public void makeCollectionsNodeWatches() throws KeeperException, InterruptedException {
- log.info("Start watching collections zk node for changes");
- zkClient.getChildren(ZkStateReader.COLLECTIONS_ZKNODE, new Watcher(){
-
- public void process(WatchedEvent event) {
- try {
-
- log.info("Detected a new or removed collection");
- synchronized (getUpdateLock()) {
- makeShardZkNodeWatches(false);
- updateCloudState(false);
- }
- // re-watch
- String path = event.getPath();
- if (path != null) {
- zkClient.getChildren(path, this);
- }
- } catch (KeeperException e) {
- if(e.code() == KeeperException.Code.SESSIONEXPIRED || e.code() == KeeperException.Code.CONNECTIONLOSS) {
- log.warn("ZooKeeper watch triggered, but Solr cannot talk to ZK");
- return;
- }
- log.error("", e);
- throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR,
- "", e);
- } catch (InterruptedException e) {
- // Restore the interrupted status
- Thread.currentThread().interrupt();
- log.error("", e);
- throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR,
- "", e);
- } catch (IOException e) {
- log.error("", e);
- throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR,
- "", e);
- }
+ }
+
+ public String getLeaderUrl(String collection, String shard) throws InterruptedException, KeeperException {
+ return getLeaderUrl(collection, shard, 1000);
+ }
+
+ public String getLeaderUrl(String collection, String shard, int timeout)
+ throws InterruptedException, KeeperException {
+ ZkCoreNodeProps props = new ZkCoreNodeProps(getLeaderProps(collection,
+ shard, timeout));
+ return props.getCoreUrl();
+ }
+
+ public ZkNodeProps getLeaderProps(String collection, String shard) throws InterruptedException {
+ return getLeaderProps(collection, shard, 1000);
+ }
+
+ public ZkNodeProps getLeaderProps(String collection, String shard, int timeout) throws InterruptedException {
+ long timeoutAt = System.currentTimeMillis() + timeout;
+ while (System.currentTimeMillis() < timeoutAt) {
+ if (cloudState != null) {
+ final CloudState currentState = cloudState;
+ final ZkNodeProps nodeProps = currentState.getLeader(collection, shard);
+ if (nodeProps != null) {
+ return nodeProps;
+ }
+ }
+ Thread.sleep(50);
+ }
+ throw new RuntimeException("No registered leader was found, collection:" + collection + " slice:" + shard);
+ }
- }});
+ public static String getShardLeadersPath(String collection, String shardId) {
+ return COLLECTIONS_ZKNODE + "/" + collection + "/"
+ + SHARD_LEADERS_ZKNODE + (shardId != null ? ("/" + shardId)
+ : "");
+ }
+
+ public List<ZkCoreNodeProps> getReplicaProps(String collection,
+ String shardId, String thisNodeName, String coreName) {
+ return getReplicaProps(collection, shardId, thisNodeName, coreName, null);
+ }
+
+ public List<ZkCoreNodeProps> getReplicaProps(String collection,
+ String shardId, String thisNodeName, String coreName, String stateFilter) {
+ CloudState cloudState = this.cloudState;
+ if (cloudState == null) {
+ return null;
+ }
+ Map<String,Slice> slices = cloudState.getSlices(collection);
+ if (slices == null) {
+ throw new ZooKeeperException(ErrorCode.BAD_REQUEST,
+ "Could not find collection in zk: " + collection + " "
+ + cloudState.getCollections());
+ }
- zkClient.exists(ZkStateReader.COLLECTIONS_ZKNODE, new Watcher(){
-
- public void process(WatchedEvent event) {
- if(event.getType() != EventType.NodeDataChanged) {
- return;
- }
- log.info("Notified of CloudState change");
- try {
- synchronized (getUpdateLock()) {
- makeShardZkNodeWatches(false);
- updateCloudState(false);
- }
- zkClient.exists(ZkStateReader.COLLECTIONS_ZKNODE, this);
- } catch (KeeperException e) {
- if(e.code() == KeeperException.Code.SESSIONEXPIRED || e.code() == KeeperException.Code.CONNECTIONLOSS) {
- log.warn("ZooKeeper watch triggered, but Solr cannot talk to ZK");
- return;
- }
- log.error("", e);
- throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR,
- "", e);
- } catch (InterruptedException e) {
- // Restore the interrupted status
- Thread.currentThread().interrupt();
- log.error("", e);
- throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR,
- "", e);
- } catch (IOException e) {
- log.error("", e);
- throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR,
- "", e);
- }
-
- }});
+ Slice replicas = slices.get(shardId);
+ if (replicas == null) {
+ throw new ZooKeeperException(ErrorCode.BAD_REQUEST, "Could not find shardId in zk: " + shardId);
+ }
+ Map<String,ZkNodeProps> shardMap = replicas.getShards();
+ List<ZkCoreNodeProps> nodes = new ArrayList<ZkCoreNodeProps>(shardMap.size());
+ String filterNodeName = thisNodeName + "_" + coreName;
+ for (Entry<String,ZkNodeProps> entry : shardMap.entrySet()) {
+ ZkCoreNodeProps nodeProps = new ZkCoreNodeProps(entry.getValue());
+ String coreNodeName = nodeProps.getNodeName() + "_" + coreName;
+ if (cloudState.liveNodesContain(nodeProps.getNodeName()) && !coreNodeName.equals(filterNodeName)) {
+ if (stateFilter == null || stateFilter.equals(nodeProps.getState())) {
+ nodes.add(nodeProps);
+ }
+ }
+ }
+ if (nodes.size() == 0) {
+ // no replicas - go local
+ return null;
+ }
+
+ return nodes;
}
+
+ public SolrZkClient getZkClient() {
+ return zkClient;
+ }
+
}
Modified: lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/params/CoreAdminParams.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/params/CoreAdminParams.java?rev=1235888&r1=1235887&r2=1235888&view=diff
==============================================================================
--- lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/params/CoreAdminParams.java (original)
+++ lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/params/CoreAdminParams.java Wed Jan 25 19:49:26 2012
@@ -86,7 +86,10 @@ public interface CoreAdminParams
RENAME,
@Deprecated
ALIAS,
- MERGEINDEXES;
+ MERGEINDEXES,
+ PREPRECOVERY,
+ REQUESTRECOVERY,
+ DISTRIBURL;
public static CoreAdminAction get( String p )
{
Modified: lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/util/FastInputStream.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/util/FastInputStream.java?rev=1235888&r1=1235887&r2=1235888&view=diff
==============================================================================
--- lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/util/FastInputStream.java (original)
+++ lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/util/FastInputStream.java Wed Jan 25 19:49:26 2012
@@ -23,10 +23,11 @@ import java.io.*;
* Internal Solr use only, subject to change.
*/
public class FastInputStream extends InputStream implements DataInput {
- private final InputStream in;
- private final byte[] buf;
- private int pos;
- private int end;
+ protected final InputStream in;
+ protected final byte[] buf;
+ protected int pos;
+ protected int end;
+ protected long readFromStream; // number of bytes read from the underlying inputstream
public FastInputStream(InputStream in) {
// use default BUFSIZE of BufferedOutputStream so if we wrap that
@@ -55,10 +56,21 @@ public class FastInputStream extends Inp
return buf[pos++] & 0xff;
}
+ public int peek() throws IOException {
+ if (pos >= end) {
+ refill();
+ if (pos >= end) return -1;
+ }
+ return buf[pos] & 0xff;
+ }
+
+
public int readUnsignedByte() throws IOException {
if (pos >= end) {
refill();
- if (pos >= end) throw new EOFException();
+ if (pos >= end) {
+ throw new EOFException();
+ }
}
return buf[pos++] & 0xff;
}
@@ -67,9 +79,14 @@ public class FastInputStream extends Inp
return in.read(target, offset, len);
}
+ public long position() {
+ return readFromStream - (end - pos);
+ }
+
public void refill() throws IOException {
// this will set end to -1 at EOF
end = readWrappedStream(buf, 0, buf.length);
+ if (end > 0) readFromStream += end;
pos = 0;
}
@@ -80,11 +97,12 @@ public class FastInputStream extends Inp
@Override
public int read(byte b[], int off, int len) throws IOException {
- int r=0; // number of bytes read
+ int r=0; // number of bytes we have read
+
// first read from our buffer;
if (end-pos > 0) {
r = Math.min(end-pos, len);
- System.arraycopy(buf, pos, b, off, r);
+ System.arraycopy(buf, pos, b, off, r);
pos += r;
}
@@ -93,14 +111,19 @@ public class FastInputStream extends Inp
// amount left to read is >= buffer size
if (len-r >= buf.length) {
int ret = readWrappedStream(b, off+r, len-r);
- if (ret==-1) return r==0 ? -1 : r;
- r += ret;
- return r;
+ if (ret >= 0) {
+ readFromStream += ret;
+ r += ret;
+ return r;
+ } else {
+ // negative return code
+ return r > 0 ? r : -1;
+ }
}
refill();
- // first read from our buffer;
+ // read rest from our buffer
if (end-pos > 0) {
int toRead = Math.min(end-pos, len-r);
System.arraycopy(buf, pos, b, off+r, toRead);
@@ -108,7 +131,7 @@ public class FastInputStream extends Inp
r += toRead;
return r;
}
-
+
return r > 0 ? r : -1;
}
Modified: lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/util/FastOutputStream.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/util/FastOutputStream.java?rev=1235888&r1=1235887&r2=1235888&view=diff
==============================================================================
--- lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/util/FastOutputStream.java (original)
+++ lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/util/FastOutputStream.java Wed Jan 25 19:49:26 2012
@@ -25,7 +25,7 @@ import java.io.*;
public class FastOutputStream extends OutputStream implements DataOutput {
private final OutputStream out;
private final byte[] buf;
- private long written; // how many bytes written
+ private long written; // how many bytes written to the underlying stream
private int pos;
public FastOutputStream(OutputStream w) {
@@ -191,4 +191,17 @@ public class FastOutputStream extends Ou
public long size() {
return written + pos;
}
+
+ /** Returns the number of bytes actually written to the underlying OutputStream, not including
+ * anything currently buffered by this class itself.
+ */
+ public long written() {
+ return written;
+ }
+
+ /** Resets the count returned by written() */
+ public void setWritten(long written) {
+ this.written = written;
+ }
+
}
Modified: lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/util/Hash.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/util/Hash.java?rev=1235888&r1=1235887&r2=1235888&view=diff
==============================================================================
--- lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/util/Hash.java (original)
+++ lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/util/Hash.java Wed Jan 25 19:49:26 2012
@@ -239,4 +239,59 @@ public class Hash {
return c + (((long)b) << 32);
}
+
+ /** Returns the MurmurHash3_x86_32 hash.
+ * Original source/tests at https://github.com/yonik/java_util/
+ */
+ public static int murmurhash3_x86_32(byte[] data, int offset, int len, int seed) {
+
+ final int c1 = 0xcc9e2d51;
+ final int c2 = 0x1b873593;
+
+ int h1 = seed;
+ int roundedEnd = offset + (len & 0xfffffffc); // round down to 4 byte block
+
+ for (int i=offset; i<roundedEnd; i+=4) {
+ // little endian load order
+ int k1 = (data[i] & 0xff) | ((data[i+1] & 0xff) << 8) | ((data[i+2] & 0xff) << 16) | (data[i+3] << 24);
+ k1 *= c1;
+ k1 = (k1 << 15) | (k1 >>> 17); // ROTL32(k1,15);
+ k1 *= c2;
+
+ h1 ^= k1;
+ h1 = (h1 << 13) | (h1 >>> 19); // ROTL32(h1,13);
+ h1 = h1*5+0xe6546b64;
+ }
+
+ // tail
+ int k1 = 0;
+
+ switch(len & 0x03) {
+ case 3:
+ k1 = (data[roundedEnd + 2] & 0xff) << 16;
+ // fallthrough
+ case 2:
+ k1 |= (data[roundedEnd + 1] & 0xff) << 8;
+ // fallthrough
+ case 1:
+ k1 |= (data[roundedEnd] & 0xff);
+ k1 *= c1;
+ k1 = (k1 << 15) | (k1 >>> 17); // ROTL32(k1,15);
+ k1 *= c2;
+ h1 ^= k1;
+ }
+
+ // finalization
+ h1 ^= len;
+
+ // fmix(h1);
+ h1 ^= h1 >>> 16;
+ h1 *= 0x85ebca6b;
+ h1 ^= h1 >>> 13;
+ h1 *= 0xc2b2ae35;
+ h1 ^= h1 >>> 16;
+
+ return h1;
+ }
+
}
Modified: lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/util/StrUtils.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/util/StrUtils.java?rev=1235888&r1=1235887&r2=1235888&view=diff
==============================================================================
--- lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/util/StrUtils.java (original)
+++ lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/util/StrUtils.java Wed Jan 25 19:49:26 2012
@@ -144,10 +144,11 @@ public class StrUtils {
}
/** Creates a backslash escaped string, joining all the items. */
- public static String join(List<String> items, char separator) {
+ public static String join(List<?> items, char separator) {
StringBuilder sb = new StringBuilder(items.size() << 3);
boolean first=true;
- for (String item : items) {
+ for (Object o : items) {
+ String item = o.toString();
if (first) {
first = false;
} else {
Added: lucene/dev/trunk/solr/solrj/src/java/org/apache/zookeeper/SolrZooKeeper.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/solrj/src/java/org/apache/zookeeper/SolrZooKeeper.java?rev=1235888&view=auto
==============================================================================
--- lucene/dev/trunk/solr/solrj/src/java/org/apache/zookeeper/SolrZooKeeper.java (added)
+++ lucene/dev/trunk/solr/solrj/src/java/org/apache/zookeeper/SolrZooKeeper.java Wed Jan 25 19:49:26 2012
@@ -0,0 +1,41 @@
+package org.apache.zookeeper;
+
+import java.io.IOException;
+import java.nio.channels.SocketChannel;
+
+// we use this class to expose nasty stuff for tests
+public class SolrZooKeeper extends ZooKeeper {
+
+ public SolrZooKeeper(String connectString, int sessionTimeout,
+ Watcher watcher) throws IOException {
+ super(connectString, sessionTimeout, watcher);
+ }
+
+ public ClientCnxn getConnection() {
+ return cnxn;
+ }
+
+ /**
+ * Cause this ZooKeeper object to stop receiving from the ZooKeeperServer
+ * for the given number of milliseconds.
+ * @param ms the number of milliseconds to pause.
+ */
+ public void pauseCnxn(final long ms) {
+ new Thread() {
+ public void run() {
+ synchronized (cnxn) {
+ try {
+ try {
+ ((SocketChannel) cnxn.sendThread.sockKey.channel()).socket()
+ .close();
+ } catch (Exception e) {
+
+ }
+ Thread.sleep(ms);
+ } catch (InterruptedException e) {}
+ }
+ }
+ }.start();
+ }
+
+}
Modified: lucene/dev/trunk/solr/solrj/src/test/org/apache/solr/client/solrj/TestLBHttpSolrServer.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/solrj/src/test/org/apache/solr/client/solrj/TestLBHttpSolrServer.java?rev=1235888&r1=1235887&r2=1235888&view=diff
==============================================================================
--- lucene/dev/trunk/solr/solrj/src/test/org/apache/solr/client/solrj/TestLBHttpSolrServer.java (original)
+++ lucene/dev/trunk/solr/solrj/src/test/org/apache/solr/client/solrj/TestLBHttpSolrServer.java Wed Jan 25 19:49:26 2012
@@ -276,8 +276,7 @@ public class TestLBHttpSolrServer extend
}
public void startJetty() throws Exception {
- jetty = new JettySolrRunner("/solr", port);
- System.setProperty("solr.solr.home", getHomeDir());
+ jetty = new JettySolrRunner(getHomeDir(), "/solr", port, "bad_solrconfig.xml", null);
System.setProperty("solr.data.dir", getDataDir());
jetty.start();
int newPort = jetty.getLocalPort();
Modified: lucene/dev/trunk/solr/solrj/src/test/org/apache/solr/client/solrj/embedded/MultiCoreExampleJettyTest.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/solrj/src/test/org/apache/solr/client/solrj/embedded/MultiCoreExampleJettyTest.java?rev=1235888&r1=1235887&r2=1235888&view=diff
==============================================================================
--- lucene/dev/trunk/solr/solrj/src/test/org/apache/solr/client/solrj/embedded/MultiCoreExampleJettyTest.java (original)
+++ lucene/dev/trunk/solr/solrj/src/test/org/apache/solr/client/solrj/embedded/MultiCoreExampleJettyTest.java Wed Jan 25 19:49:26 2012
@@ -53,7 +53,7 @@ public class MultiCoreExampleJettyTest e
System.clearProperty("solr.directoryFactory");
super.setUp();
- jetty = new JettySolrRunner( context, 0 );
+ jetty = new JettySolrRunner(getSolrHome(), context, 0 );
jetty.start(false);
port = jetty.getLocalPort();
Modified: lucene/dev/trunk/solr/test-framework/src/java/org/apache/solr/BaseDistributedSearchTestCase.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/test-framework/src/java/org/apache/solr/BaseDistributedSearchTestCase.java?rev=1235888&r1=1235887&r2=1235888&view=diff
==============================================================================
--- lucene/dev/trunk/solr/test-framework/src/java/org/apache/solr/BaseDistributedSearchTestCase.java (original)
+++ lucene/dev/trunk/solr/test-framework/src/java/org/apache/solr/BaseDistributedSearchTestCase.java Wed Jan 25 19:49:26 2012
@@ -36,11 +36,14 @@ import org.apache.solr.client.solrj.Solr
import org.apache.solr.client.solrj.SolrServerException;
import org.apache.solr.client.solrj.embedded.JettySolrRunner;
import org.apache.solr.client.solrj.impl.CommonsHttpSolrServer;
+import org.apache.solr.client.solrj.request.UpdateRequest;
import org.apache.solr.client.solrj.response.QueryResponse;
+import org.apache.solr.client.solrj.response.UpdateResponse;
import org.apache.solr.common.SolrDocument;
import org.apache.solr.common.SolrDocumentList;
import org.apache.solr.common.SolrInputDocument;
import org.apache.solr.common.params.ModifiableSolrParams;
+import org.apache.solr.common.params.SolrParams;
import org.apache.solr.common.util.NamedList;
import org.apache.solr.schema.TrieDateField;
import org.apache.solr.util.AbstractSolrTestCase;
@@ -69,6 +72,7 @@ public abstract class BaseDistributedSea
protected JettySolrRunner controlJetty;
protected List<SolrServer> clients = new ArrayList<SolrServer>();
protected List<JettySolrRunner> jettys = new ArrayList<JettySolrRunner>();
+
protected String context = "/solr";
protected String shards;
protected String[] shardsArr;
@@ -159,7 +163,6 @@ public abstract class BaseDistributedSea
super.setUp();
System.setProperty("solr.test.sys.prop1", "propone");
System.setProperty("solr.test.sys.prop2", "proptwo");
- System.setProperty("solr.solr.home", getSolrHome());
testDir = new File(TEMP_DIR,
getClass().getName() + "-" + System.currentTimeMillis());
testDir.mkdirs();
@@ -175,14 +178,17 @@ public abstract class BaseDistributedSea
}
protected void createServers(int numShards) throws Exception {
- controlJetty = createJetty(testDir, testDir + "/control/data");
+ controlJetty = createJetty(testDir, testDir + "/control/data", null, getSolrConfigFile(), getSchemaFile());
+
controlClient = createNewSolrServer(controlJetty.getLocalPort());
shardsArr = new String[numShards];
StringBuilder sb = new StringBuilder();
for (int i = 0; i < numShards; i++) {
if (sb.length() > 0) sb.append(',');
- JettySolrRunner j = createJetty(testDir, testDir + "/shard" + i + "/data");
+ JettySolrRunner j = createJetty(testDir,
+ testDir + "/shard" + i + "/data", null, getSolrConfigFile(),
+ getSchemaFile());
jettys.add(j);
clients.add(createNewSolrServer(j.getLocalPort()));
String shardStr = "localhost:" + j.getLocalPort() + context;
@@ -223,27 +229,28 @@ public abstract class BaseDistributedSea
protected void destroyServers() throws Exception {
controlJetty.stop();
+ ((CommonsHttpSolrServer) controlClient).shutdown();
for (JettySolrRunner jetty : jettys) jetty.stop();
+ for (SolrServer client : clients) ((CommonsHttpSolrServer) client).shutdown();
clients.clear();
jettys.clear();
}
public JettySolrRunner createJetty(File baseDir, String dataDir) throws Exception {
- return createJetty(baseDir, dataDir, null, null);
+ return createJetty(baseDir, dataDir, null, null, null);
}
public JettySolrRunner createJetty(File baseDir, String dataDir, String shardId) throws Exception {
- return createJetty(baseDir, dataDir, shardId, null);
+ return createJetty(baseDir, dataDir, shardId, null, null);
}
- public JettySolrRunner createJetty(File baseDir, String dataDir, String shardList, String solrConfigOverride) throws Exception {
- System.setProperty("solr.data.dir", dataDir);
- JettySolrRunner jetty = new JettySolrRunner("/solr", 0, solrConfigOverride);
- if(shardList != null) {
- System.setProperty("shard", shardList);
- }
+ public JettySolrRunner createJetty(File baseDir, String dataDir, String shardList, String solrConfigOverride, String schemaOverride) throws Exception {
+
+ JettySolrRunner jetty = new JettySolrRunner(getSolrHome(), "/solr", 0, solrConfigOverride, schemaOverride);
+ jetty.setShards(shardList);
+ jetty.setDataDir(dataDir);
jetty.start();
- System.clearProperty("shard");
+
return jetty;
}
@@ -272,9 +279,14 @@ public abstract class BaseDistributedSea
SolrInputDocument doc = new SolrInputDocument();
addFields(doc, fields);
addFields(doc, "rnd_b", true);
- addFields(doc, getRandFields(getFieldNames(), getRandValues()));
+ addRandFields(doc);
indexDoc(doc);
}
+
+ protected SolrInputDocument addRandFields(SolrInputDocument sdoc) {
+ addFields(sdoc, getRandFields(getFieldNames(), getRandValues()));
+ return sdoc;
+ }
protected void index(Object... fields) throws Exception {
SolrInputDocument doc = new SolrInputDocument();
@@ -289,6 +301,33 @@ public abstract class BaseDistributedSea
SolrServer client = clients.get(which);
client.add(doc);
}
+
+ protected UpdateResponse add(SolrServer server, SolrParams params, SolrInputDocument... sdocs) throws IOException, SolrServerException {
+ UpdateRequest ureq = new UpdateRequest();
+ ureq.setParams(new ModifiableSolrParams(params));
+ for (SolrInputDocument sdoc : sdocs) {
+ ureq.add(sdoc);
+ }
+ return ureq.process(server);
+ }
+
+ protected UpdateResponse del(SolrServer server, SolrParams params, Object... ids) throws IOException, SolrServerException {
+ UpdateRequest ureq = new UpdateRequest();
+ ureq.setParams(new ModifiableSolrParams(params));
+ for (Object id: ids) {
+ ureq.deleteById(id.toString());
+ }
+ return ureq.process(server);
+ }
+
+ protected UpdateResponse delQ(SolrServer server, SolrParams params, String... queries) throws IOException, SolrServerException {
+ UpdateRequest ureq = new UpdateRequest();
+ ureq.setParams(new ModifiableSolrParams(params));
+ for (String q: queries) {
+ ureq.deleteByQuery(q);
+ }
+ return ureq.process(server);
+ }
protected void index_specific(int serverNumber, Object... fields) throws Exception {
SolrInputDocument doc = new SolrInputDocument();
@@ -310,7 +349,9 @@ public abstract class BaseDistributedSea
protected void commit() throws Exception {
controlClient.commit();
- for (SolrServer client : clients) client.commit();
+ for (SolrServer client : clients) {
+ client.commit();
+ }
}
protected QueryResponse queryServer(ModifiableSolrParams params) throws SolrServerException {
@@ -322,14 +363,17 @@ public abstract class BaseDistributedSea
}
protected void query(Object... q) throws Exception {
+
final ModifiableSolrParams params = new ModifiableSolrParams();
for (int i = 0; i < q.length; i += 2) {
params.add(q[i].toString(), q[i + 1].toString());
}
-
+ // TODO: look into why passing true causes fails
+ params.set("distrib", "false");
final QueryResponse controlRsp = controlClient.query(params);
+ params.remove("distrib");
setDistributedParams(params);
QueryResponse rsp = queryServer(params);
@@ -365,6 +409,20 @@ public abstract class BaseDistributedSea
}
}
}
+
+ public QueryResponse queryAndCompare(SolrParams params, SolrServer... servers) throws SolrServerException {
+ QueryResponse first = null;
+ for (SolrServer server : servers) {
+ QueryResponse rsp = server.query(new ModifiableSolrParams(params));
+ if (first == null) {
+ first = rsp;
+ } else {
+ compareResponses(first, rsp);
+ }
+ }
+
+ return first;
+ }
public static boolean eq(String a, String b) {
return a == b || (a != null && a.equals(b));
@@ -377,6 +435,8 @@ public abstract class BaseDistributedSea
}
public static String compare(NamedList a, NamedList b, int flags, Map<String, Integer> handle) {
+// System.out.println("resp a:" + a);
+// System.out.println("resp b:" + b);
boolean ordered = (flags & UNORDERED) == 0;
int posa = 0, posb = 0;
@@ -578,9 +638,19 @@ public abstract class BaseDistributedSea
protected void compareResponses(QueryResponse a, QueryResponse b) {
String cmp;
+ if (System.getProperty("remove.version.field") != null) {
+ // we don't care if one has a version and the other doesnt -
+ // control vs distrib
+ for (SolrDocument doc : a.getResults()) {
+ doc.removeFields("_version_");
+ }
+ for (SolrDocument doc : b.getResults()) {
+ doc.removeFields("_version_");
+ }
+ }
cmp = compare(a.getResponse(), b.getResponse(), flags, handle);
if (cmp != null) {
- log.info("Mismatched responses:\n" + a + "\n" + b);
+ log.error("Mismatched responses:\n" + a + "\n" + b);
TestCase.fail(cmp);
}
}
Modified: lucene/dev/trunk/solr/test-framework/src/java/org/apache/solr/SolrJettyTestBase.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/test-framework/src/java/org/apache/solr/SolrJettyTestBase.java?rev=1235888&r1=1235887&r2=1235888&view=diff
==============================================================================
--- lucene/dev/trunk/solr/test-framework/src/java/org/apache/solr/SolrJettyTestBase.java (original)
+++ lucene/dev/trunk/solr/test-framework/src/java/org/apache/solr/SolrJettyTestBase.java Wed Jan 25 19:49:26 2012
@@ -44,12 +44,11 @@ abstract public class SolrJettyTestBase
ignoreException("maxWarmingSearchers");
// this sets the property for jetty starting SolrDispatchFilter
- System.setProperty( "solr.solr.home", solrHome);
System.setProperty( "solr.data.dir", dataDir.getCanonicalPath() );
context = context==null ? "/solr" : context;
SolrJettyTestBase.context = context;
- jetty = new JettySolrRunner( context, 0, configFile );
+ jetty = new JettySolrRunner(solrHome, context, 0, configFile, null);
jetty.start();
port = jetty.getLocalPort();