You are viewing a plain text version of this content. The canonical link for it is here.
Posted to solr-commits@lucene.apache.org by yo...@apache.org on 2010/02/07 16:32:30 UTC
svn commit: r907432 - in /lucene/solr/branches/cloud: build.xml
src/java/org/apache/solr/cloud/ZkController.java
src/solrj/org/apache/solr/client/solrj/impl/CloudSolrServer.java
Author: yonik
Date: Sun Feb 7 15:32:29 2010
New Revision: 907432
URL: http://svn.apache.org/viewvc?rev=907432&view=rev
Log:
first cut at SolrJ client for cloud
Added:
lucene/solr/branches/cloud/src/solrj/org/apache/solr/client/solrj/impl/CloudSolrServer.java
Modified:
lucene/solr/branches/cloud/build.xml
lucene/solr/branches/cloud/src/java/org/apache/solr/cloud/ZkController.java
Modified: lucene/solr/branches/cloud/build.xml
URL: http://svn.apache.org/viewvc/lucene/solr/branches/cloud/build.xml?rev=907432&r1=907431&r2=907432&view=diff
==============================================================================
--- lucene/solr/branches/cloud/build.xml (original)
+++ lucene/solr/branches/cloud/build.xml Sun Feb 7 15:32:29 2010
@@ -119,6 +119,7 @@
<include name="*.jar" />
</fileset>
<pathelement location="${dest}/common"/>
+ <pathelement location="${src}/java"/> <!-- temp hack since CloudSolrServer depends on ZkController, etc -->
</path>
<target name="compile-solrj"
Modified: lucene/solr/branches/cloud/src/java/org/apache/solr/cloud/ZkController.java
URL: http://svn.apache.org/viewvc/lucene/solr/branches/cloud/src/java/org/apache/solr/cloud/ZkController.java?rev=907432&r1=907431&r2=907432&view=diff
==============================================================================
--- lucene/solr/branches/cloud/src/java/org/apache/solr/cloud/ZkController.java (original)
+++ lucene/solr/branches/cloud/src/java/org/apache/solr/cloud/ZkController.java Sun Feb 7 15:32:29 2010
@@ -90,6 +90,8 @@
private boolean cloudStateUpdateScheduled;
+ private boolean readonly; // temporary hack to enable reuse in SolrJ client
+
/**
* @param zkServerAddress ZooKeeper server host address
* @param zkClientTimeout
@@ -108,6 +110,7 @@
this.localHostPort = locaHostPort;
this.localHostContext = localHostContext;
this.localHost = localHost;
+ this.readonly = localHostPort==null;
cloudState = new CloudState(new HashSet<String>(0), new HashMap<String,Map<String,Slice>>(0));
zkClient = new SolrZkClient(zkServerAddress, zkClientTimeout, zkClientConnectTimeout,
// on reconnect, reload cloud info
@@ -345,7 +348,8 @@
};
try {
- zkClient.makePath(nodePath, CreateMode.EPHEMERAL);
+ if (!readonly)
+ zkClient.makePath(nodePath, CreateMode.EPHEMERAL);
} catch (KeeperException e) {
// its okay if the node already exists
if (e.code() != KeeperException.Code.NODEEXISTS) {
@@ -585,7 +589,7 @@
private void setUpCollectionsNode() throws KeeperException, InterruptedException {
try {
- if (!zkClient.exists(COLLECTIONS_ZKNODE)) {
+ if (!zkClient.exists(COLLECTIONS_ZKNODE) && !readonly) {
if (log.isInfoEnabled()) {
log.info("creating zk collections node:" + COLLECTIONS_ZKNODE);
}
Added: lucene/solr/branches/cloud/src/solrj/org/apache/solr/client/solrj/impl/CloudSolrServer.java
URL: http://svn.apache.org/viewvc/lucene/solr/branches/cloud/src/solrj/org/apache/solr/client/solrj/impl/CloudSolrServer.java?rev=907432&view=auto
==============================================================================
--- lucene/solr/branches/cloud/src/solrj/org/apache/solr/client/solrj/impl/CloudSolrServer.java (added)
+++ lucene/solr/branches/cloud/src/solrj/org/apache/solr/client/solrj/impl/CloudSolrServer.java Sun Feb 7 15:32:29 2010
@@ -0,0 +1,131 @@
+package org.apache.solr.client.solrj.impl;
+
+import org.apache.solr.client.solrj.SolrRequest;
+import org.apache.solr.client.solrj.SolrServer;
+import org.apache.solr.client.solrj.SolrServerException;
+import org.apache.solr.cloud.*;
+import org.apache.solr.common.SolrException;
+import org.apache.solr.common.util.NamedList;
+import org.apache.zookeeper.KeeperException;
+
+import java.io.IOException;
+import java.net.MalformedURLException;
+import java.util.*;
+import java.util.concurrent.TimeoutException;
+
+public class CloudSolrServer extends SolrServer {
+ private volatile ZkController zkController;
+ private String zkHost; // the zk server address
+ private int zkConnectTimeout = 10000;
+ private int zkClientTimeout = 10000;
+ private String defaultCollection;
+ private LBHttpSolrServer lbServer;
+ Random rand = new Random();
+
+ /**
+ * @param zkHost The address of the zookeeper quorum containing the cloud state
+ */
+ public CloudSolrServer(String zkHost) throws MalformedURLException {
+ this(zkHost, new LBHttpSolrServer());
+ }
+
+ /**
+ * @param zkHost The address of the zookeeper quorum containing the cloud state
+ */
+ public CloudSolrServer(String zkHost, LBHttpSolrServer lbServer) {
+ this.zkHost = zkHost;
+ this.lbServer = lbServer;
+ }
+
+ /** Sets the default collection for request */
+ public void setDefaultCollection(String collection) {
+ this.defaultCollection = collection;
+ }
+
+ /** Set the connect timeout to the zookeeper ensemble in ms */
+ public void setZkConnectTimeout(int zkConnectTimeout) {
+ this.zkConnectTimeout = zkConnectTimeout;
+ }
+
+ /** Set the timeout to the zookeeper ensemble in ms */
+ public void setZkClientTimeout(int zkClientTimeout) {
+ this.zkClientTimeout = zkClientTimeout;
+ }
+
+ /**
+ * Connect to the zookeeper ensemble.
+ * This is an optional method that may be used to force a connect before any other requests are sent.
+ *
+ * @throws IOException
+ * @throws TimeoutException
+ * @throws InterruptedException
+ */
+ public void connect() {
+ if (zkController != null) return;
+ synchronized(this) {
+ if (zkController != null) return;
+ try {
+ zkController = new ZkController(zkHost, zkConnectTimeout, zkClientTimeout, null, null, null);
+ zkController.addShardZkNodeWatches();
+ zkController.updateCloudState(true);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR, "", e);
+ } catch (KeeperException e) {
+ throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR, "", e);
+
+ } catch (IOException e) {
+ throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR, "", e);
+
+ } catch (TimeoutException e) {
+ throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR, "", e);
+ }
+ }
+ }
+
+
+ @Override
+ public NamedList<Object> request(SolrRequest request) throws SolrServerException, IOException {
+ connect();
+
+ CloudState cloudState = zkController.getCloudState();
+
+ String collection = request.getParams().get("collection", defaultCollection);
+
+ // TODO: allow multiple collections to be specified via comma separated list
+
+ Map<String,Slice> slices = cloudState.getSlices(collection);
+ Set<String> liveNodes = cloudState.getLiveNodes();
+
+ // IDEA: have versions on various things... like a global cloudState version
+ // or shardAddressVersion (which only changes when the shards change)
+ // to allow caching.
+
+ // build a map of unique nodes
+ // TODO: allow filtering by group, role, etc
+ Map<String,ZkNodeProps> nodes = new HashMap<String,ZkNodeProps>();
+ List<String> urlList = new ArrayList<String>();
+ for (Slice slice : slices.values()) {
+ for (ZkNodeProps nodeProps : slice.getShards().values()) {
+ String node = nodeProps.get(ZkController.NODE_NAME);
+ if (!liveNodes.contains(node)) continue;
+ if (nodes.put(node, nodeProps) == null) {
+ String url = nodeProps.get(ZkController.URL_PROP);
+ urlList.add(url);
+ }
+ }
+ }
+
+ Collections.shuffle(urlList, rand);
+
+ // TODO: set distrib=true if we detected more than one shard?
+ LBHttpSolrServer.Req req = new LBHttpSolrServer.Req(request, urlList);
+ LBHttpSolrServer.Rsp rsp = lbServer.request(req);
+ return rsp.getResponse();
+ }
+
+ public void close() {
+ zkController.close();
+ zkController = null;
+ }
+}