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/12/22 20:43:09 UTC
svn commit: r1222409 [1/2] - in /lucene/dev/branches/solrcloud/solr: ./
cloud-dev/ core/src/java/org/apache/solr/client/solrj/embedded/
core/src/java/org/apache/solr/cloud/
core/src/java/org/apache/solr/handler/admin/
core/src/java/org/apache/solr/upda...
Author: markrmiller
Date: Thu Dec 22 19:43:08 2011
New Revision: 1222409
URL: http://svn.apache.org/viewvc?rev=1222409&view=rev
Log:
sync up solrcloud work
Added:
lucene/dev/branches/solrcloud/solr/cloud-dev/example3.sh
lucene/dev/branches/solrcloud/solr/lib/zookeeper-3.3.4.jar (with props)
Removed:
lucene/dev/branches/solrcloud/solr/lib/zookeeper-3.4.0.jar
Modified:
lucene/dev/branches/solrcloud/solr/ (props changed)
lucene/dev/branches/solrcloud/solr/CHANGES.txt
lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/client/solrj/embedded/JettySolrRunner.java
lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/RecoveryStrat.java
lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/ZkController.java
lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/handler/admin/CoreAdminHandler.java
lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/update/SolrCmdDistributor.java
lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/update/TransactionLog.java
lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/update/processor/DistributedUpdateProcessor.java
lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/ChaosMonkey.java
lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/ChaosMonkeySolrCloudTest.java
lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/FullSolrCloudTest.java
lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/RecoveryZkTest.java
lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/ZkTestServer.java
lucene/dev/branches/solrcloud/solr/solrj/src/java/org/apache/solr/client/solrj/impl/LBHttpSolrServer.java
lucene/dev/branches/solrcloud/solr/solrj/src/java/org/apache/solr/client/solrj/request/CoreAdminRequest.java
lucene/dev/branches/solrcloud/solr/solrj/src/java/org/apache/solr/common/params/CoreAdminParams.java
lucene/dev/branches/solrcloud/solr/solrj/src/java/org/apache/zookeeper/SolrZooKeeper.java
lucene/dev/branches/solrcloud/solr/solrj/src/test/org/apache/solr/client/solrj/TestLBHttpSolrServer.java
Modified: lucene/dev/branches/solrcloud/solr/CHANGES.txt
URL: http://svn.apache.org/viewvc/lucene/dev/branches/solrcloud/solr/CHANGES.txt?rev=1222409&r1=1222408&r2=1222409&view=diff
==============================================================================
--- lucene/dev/branches/solrcloud/solr/CHANGES.txt (original)
+++ lucene/dev/branches/solrcloud/solr/CHANGES.txt Thu Dec 22 19:43:08 2011
@@ -28,7 +28,7 @@ Apache Tika 0.10
Carrot2 3.5.0
Velocity 1.6.4 and Velocity Tools 2.0
Apache UIMA 2.3.1
-Apache ZooKeeper 3.4
+Apache ZooKeeper 3.3.4
Upgrading from Solr 3.5-dev
Added: lucene/dev/branches/solrcloud/solr/cloud-dev/example3.sh
URL: http://svn.apache.org/viewvc/lucene/dev/branches/solrcloud/solr/cloud-dev/example3.sh?rev=1222409&view=auto
==============================================================================
--- lucene/dev/branches/solrcloud/solr/cloud-dev/example3.sh (added)
+++ lucene/dev/branches/solrcloud/solr/cloud-dev/example3.sh Thu Dec 22 19:43:08 2011
@@ -0,0 +1,34 @@
+#!/usr/bin/env bash
+
+cd ..
+
+rm -r -f example2
+rm -r -f example3
+rm -r -f example3
+
+rm -r -f dist
+rm -r -f build
+rm -r -f example/solr/zoo_data
+rm -f example/example.log
+
+ant example dist
+
+cp -r -f example example2
+cp -r -f example example3
+cp -r -f example example4
+
+
+cd example
+java -DzkRun -DnumShards=2 -DSTOP.PORT=7983 -DSTOP.KEY=key -Dbootstrap_confdir=solr/conf -DzkHost=localhost:9983,localhost:8574,localhost:8575 -jar start.jar 1>example.log 2>&1 &
+
+sleep 10
+
+cd ../example2
+java -Djetty.port=7574 -DzkRun -DzkHost=localhost:9983,localhost:8574,localhost:8575 -DnumShards=2 -DSTOP.PORT=6574 -DSTOP.KEY=key -jar start.jar 1>example2.log 2>&1 &
+
+cd ../example3
+java -Djetty.port=7575 -DzkRun -DzkHost=localhost:9983,localhost:8574,localhost:8575 -DnumShards=2 -DSTOP.PORT=6575 -DSTOP.KEY=key -jar start.jar 1>example3.log 2>&1 &
+
+cd ../example4
+java -Djetty.port=7576 -DzkHost=localhost:9983,localhost:8574,localhost:8575 -DnumShards=2 -DSTOP.PORT=6576 -DSTOP.KEY=key -jar start.jar 1>example4.log 2>&1 &
+
Modified: lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/client/solrj/embedded/JettySolrRunner.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/client/solrj/embedded/JettySolrRunner.java?rev=1222409&r1=1222408&r2=1222409&view=diff
==============================================================================
--- lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/client/solrj/embedded/JettySolrRunner.java (original)
+++ lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/client/solrj/embedded/JettySolrRunner.java Thu Dec 22 19:43:08 2011
@@ -35,6 +35,8 @@ import org.mortbay.jetty.servlet.FilterH
import org.mortbay.jetty.servlet.HashSessionIdManager;
import org.mortbay.log.Logger;
+import com.sun.net.httpserver.Filter;
+
/**
* Run solr using jetty
*
@@ -116,6 +118,8 @@ public class JettySolrRunner {
System.setProperty("hostPort", Integer.toString(lastPort));
if (solrConfigFilename != null) System.setProperty("solrconfig",
solrConfigFilename);
+// SolrDispatchFilter filter = new SolrDispatchFilter();
+// FilterHolder fh = new FilterHolder(filter);
dispatchFilter = root.addFilter(SolrDispatchFilter.class, "*",
Handler.REQUEST);
if (solrConfigFilename != null) System.clearProperty("solrconfig");
@@ -152,11 +156,14 @@ public class JettySolrRunner {
public void start(boolean waitForSolr) throws Exception {
// if started before, make a new server
if (startedBefore) {
+
init(solrHome, context, lastPort, stopAtShutdown);
} else {
startedBefore = true;
}
+ System.out.println("starting up on port " + lastPort);
+
if( dataDir != null) {
System.setProperty("solr.data.dir", dataDir);
}
@@ -182,7 +189,9 @@ public class JettySolrRunner {
}
public void stop() throws Exception {
- server.stop();
+ if (!server.isStopped() && !server.isStopping()) {
+ server.stop();
+ }
server.join();
}
Modified: lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/RecoveryStrat.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/RecoveryStrat.java?rev=1222409&r1=1222408&r2=1222409&view=diff
==============================================================================
--- lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/RecoveryStrat.java (original)
+++ lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/cloud/RecoveryStrat.java Thu Dec 22 19:43:08 2011
@@ -20,28 +20,35 @@ package org.apache.solr.cloud;
import java.io.IOException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.apache.lucene.search.MatchAllDocsQuery;
import org.apache.solr.client.solrj.SolrServerException;
-import org.apache.solr.client.solrj.embedded.EmbeddedSolrServer;
import org.apache.solr.client.solrj.impl.CommonsHttpSolrServer;
-import org.apache.solr.client.solrj.request.AbstractUpdateRequest;
-import org.apache.solr.client.solrj.request.UpdateRequest;
+import org.apache.solr.client.solrj.request.CoreAdminRequest.PrepRecovery;
+import org.apache.solr.common.SolrException;
+import org.apache.solr.common.SolrException.ErrorCode;
import org.apache.solr.common.cloud.ZkStateReader;
+import org.apache.solr.common.params.CoreAdminParams.CoreAdminAction;
import org.apache.solr.common.params.ModifiableSolrParams;
import org.apache.solr.core.RequestHandlers.LazyRequestHandlerWrapper;
import org.apache.solr.core.SolrCore;
import org.apache.solr.handler.ReplicationHandler;
import org.apache.solr.request.SolrRequestHandler;
import org.apache.solr.search.SolrIndexSearcher;
+import org.apache.solr.update.UpdateLog;
import org.apache.solr.update.UpdateLog.RecoveryInfo;
-import org.apache.solr.update.processor.DistributedUpdateProcessor;
+import org.apache.solr.update.UpdateLog.State;
import org.apache.solr.util.RefCounted;
-import org.apache.zookeeper.KeeperException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class RecoveryStrat {
+ private static final int MAX_RETRIES = 10;
+
private static final String REPLICATION_HANDLER = "/replication";
private static Logger log = LoggerFactory.getLogger(RecoveryStrat.class);
@@ -50,10 +57,10 @@ public class RecoveryStrat {
private volatile boolean close = false;
-
- interface OnFinish {
- public void run();
- }
+ private final AtomicInteger recoveryAttempts = new AtomicInteger();
+ private final AtomicInteger recoverySuccesses = new AtomicInteger();
+
+ private static final Lock RECOVERY_LOCK = new ReentrantLock();
// for now, just for tests
public interface RecoveryListener {
@@ -62,16 +69,28 @@ public class RecoveryStrat {
public void finishedRecovery();
}
+ // make sure any threads stop retrying
public void close() {
close = true;
}
// TODO: we want to be pretty noisy if we don't properly recover?
- public void recover(final SolrCore core, final ZkStateReader zkStateReader,
- final String shardUrl, final OnFinish onFinish) {
+ public void recover(final SolrCore core) {
+
log.info("Start recovery process");
if (recoveryListener != null) recoveryListener.startRecovery();
- core.getUpdateHandler().getUpdateLog().bufferUpdates();
+
+ final ZkController zkController = core.getCoreDescriptor()
+ .getCoreContainer().getZkController();
+ final ZkStateReader zkStateReader = zkController.getZkStateReader();
+ final String shardUrl = zkController.getShardUrl(core.getName());
+ final String shardZkNodeName = zkController.getNodeName() + "_"
+ + core.getName();
+ final CloudDescriptor cloudDesc = core.getCoreDescriptor()
+ .getCloudDescriptor();
+
+ zkController.publishAsRecoverying(shardUrl, cloudDesc, shardZkNodeName);
+
Thread thread = new Thread() {
{
setDaemon(true);
@@ -79,100 +98,134 @@ public class RecoveryStrat {
@Override
public void run() {
- boolean succesfulRecovery = false;
- int retries = 0;
- while (!succesfulRecovery && !close) {
- try {
- CloudDescriptor cloudDesc = core.getCoreDescriptor()
- .getCloudDescriptor();
- String leaderUrl = zkStateReader.getLeaderUrl(
- cloudDesc.getCollectionName(), cloudDesc.getShardId());
- doRecovery(core, leaderUrl, leaderUrl.equals(shardUrl));
- System.out.println("apply buffered updates");
- Future<RecoveryInfo> future = core.getUpdateHandler()
- .getUpdateLog().applyBufferedUpdates();
- if (future == null) {
- // no replay needed\
- log.info("No replay needed");
- } else {
- // wait for replay
- future.get();
- }
- System.out.println("replay done");
- EmbeddedSolrServer server = new EmbeddedSolrServer(core);
- server.commit();
-
- // nocommit: remove this
- RefCounted<SolrIndexSearcher> searcher = core.getSearcher(true,
- true, null);
- System.out.println("DOCS AFTER REPLAY:"
- + searcher.get().search(new MatchAllDocsQuery(), 1).totalHits);
- searcher.decref();
- if (recoveryListener != null) recoveryListener.finishedRecovery();
- onFinish.run();
-
- // TODO: what if the problem was in onFinish.run which sets the
- // state?
- succesfulRecovery = true;
- } catch (SolrServerException e) {
- log.warn("", e);
- } catch (IOException e) {
- log.warn("", e);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- log.warn("", e);
- } catch (ExecutionException e) {
- log.error("", e);
- } catch (KeeperException e) {
- log.warn("", e);
+ RECOVERY_LOCK.lock();
+ try {
+ UpdateLog ulog = core.getUpdateHandler().getUpdateLog();
+ // TODO: consider any races issues here - if we failed though, an
+ // assert
+ // exception is thrown if we are already buffering...
+ if (ulog.getState() != UpdateLog.State.BUFFERING) {
+ ulog.bufferUpdates();
}
- if (!succesfulRecovery) {
- // lets pause for a moment and we need to try again...
- // TODO: we don't want to retry for some problems?
- // Or do a fall off retry...
- log.warn("Recovery failed - trying again...");
- retries++;
- if (retries > 10) {
- // nocommit: for now, give up after 10 tries
- }
-
+ boolean succesfulRecovery = false;
+ int retries = 0;
+ while (!succesfulRecovery && !close) {
+ recoveryAttempts.incrementAndGet();
try {
- Thread.sleep(5000);
+
+ String leaderUrl = zkStateReader.getLeaderUrl(
+ cloudDesc.getCollectionName(), cloudDesc.getShardId());
+ System.out.println("leaderUrl:" + leaderUrl);
+ System.out.println("shardUrl:" + shardUrl);
+
+ replicate(core, shardZkNodeName, leaderUrl,
+ leaderUrl.equals(shardUrl));
+
+ // nocommit:
+ RefCounted<SolrIndexSearcher> searcher = core.getSearcher(true,
+ true, null);
+ SolrIndexSearcher is = searcher.get();
+ int afterReplicateDocs = 0;
+ if (!core.isClosed()) {
+ afterReplicateDocs = is.search(new MatchAllDocsQuery(), 1).totalHits;
+ searcher.decref();
+ }
+
+ System.out.println("apply buffered updates");
+ replay(core);
+ System.out.println("replay done");
+
+ // nocommit: remove this
+ searcher = core.getSearcher(true, true, null);
+ int afterReplayDocs = searcher.get().search(
+ new MatchAllDocsQuery(), 1).totalHits;
+ searcher.decref();
+
+ if (recoveryListener != null) recoveryListener.finishedRecovery();
+
+ zkController
+ .publishAsActive(shardUrl, cloudDesc, shardZkNodeName);
+
+ System.out.println("url: " + shardUrl + " docs after replicate: "
+ + afterReplicateDocs + " docs after replay:"
+ + afterReplayDocs + " leader:" + leaderUrl);
+ // TODO: what if the problem was in onFinish.run which sets the
+ // state?
+ succesfulRecovery = true;
+ recoverySuccesses.incrementAndGet();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
+ log.error("Recovery was interrupted", e);
+ retries = MAX_RETRIES;
+ } catch (Exception e) {
+ log.error("Error while trying to recover", e);
+ }
+
+ if (!succesfulRecovery) {
+ // lets pause for a moment and we need to try again...
+ // TODO: we don't want to retry for some problems?
+ // Or do a fall off retry...
+ log.error("Recovery failed - trying again...");
+ retries++;
+ if (retries >= MAX_RETRIES) {
+ // nocommit: for now, give up after 10 tries
+ close = true;
+ }
+
+ try {
+ Thread.sleep(500);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
}
}
+ log.info("Finished recovery process");
+ System.out.println("recovery done: " + succesfulRecovery);
+ } finally {
+ RECOVERY_LOCK.unlock();
}
- log.info("Finished recovery process");
- System.out.println("recovery done");
+ }
+
+ private Future<RecoveryInfo> replay(final SolrCore core)
+ throws InterruptedException, ExecutionException {
+ Future<RecoveryInfo> future = core.getUpdateHandler().getUpdateLog()
+ .applyBufferedUpdates();
+ if (future == null) {
+ // no replay needed\
+ log.info("No replay needed");
+ } else {
+ // wait for replay
+ future.get();
+ }
+ return future;
}
};
thread.start();
-
}
- private void doRecovery(SolrCore core, String leaderUrl, boolean iamleader)
+ private void replicate(SolrCore core, String shardZkNodeName, String leaderUrl, boolean iamleader)
throws SolrServerException, IOException {
-
+ System.out.println("replicate");
// start buffer updates to tran log
// and do recovery - either replay via realtime get (eventually)
// or full index replication
-
+
+ // if we are the leader, either we are trying to recover faster
+ // then our ephemeral timed out or we are the only node
if (!iamleader) {
- // if we are the leader, either we are trying to recover faster
- // then our ephemeral timed out or we are the only node
-
- // TODO: first, issue a hard commit?
- // nocommit: require /update?
CommonsHttpSolrServer server = new CommonsHttpSolrServer(leaderUrl);
-
- UpdateRequest uReq = new UpdateRequest();
- uReq.setAction(AbstractUpdateRequest.ACTION.COMMIT, true, true);
- // we do not need to distrib this commit
- uReq.setParam(DistributedUpdateProcessor.COMMIT_END_POINT, "true");
- uReq.process(server);
+ System.out.println("send prep cmd");
+ PrepRecovery prepCmd = new PrepRecovery();
+ prepCmd.setAction(CoreAdminAction.PREPRECOVERY);
+ // nocommit: the replica core name may not matcher the leader core name!
+ prepCmd.setCoreName(core.getName());
+ prepCmd.setNodeName(shardZkNodeName);
+
+ server.request(prepCmd);
+
// use rep handler directly, so we can do this sync rather than async
SolrRequestHandler handler = core.getRequestHandler(REPLICATION_HANDLER);
@@ -182,9 +235,7 @@ public class RecoveryStrat {
ReplicationHandler replicationHandler = (ReplicationHandler) handler;
if (replicationHandler == null) {
- // nocommit: we should not just return - we don't want to falsely advertise as active
- log.error("Skipping recovery, no /replication handler found");
- return;
+ throw new SolrException(ErrorCode.SERVICE_UNAVAILABLE, "Skipping recovery, no /replication handler found");
}
ModifiableSolrParams solrParams = new ModifiableSolrParams();
@@ -192,15 +243,8 @@ public class RecoveryStrat {
replicationHandler.doFetch(solrParams);
- // nocommit:
- RefCounted<SolrIndexSearcher> searcher = core.getSearcher(true, true,
- null);
- SolrIndexSearcher is = searcher.get();
- if (!core.isClosed()) {
- System.out.println("DOCS AFTER REPLICATE:"
- + is.search(new MatchAllDocsQuery(), 1).totalHits);
- searcher.decref();
- }
+
+ System.out.println("recover from: " + leaderUrl + " ");
if (recoveryListener != null) recoveryListener.finishedReplication();
}
@@ -213,4 +257,12 @@ public class RecoveryStrat {
public void setRecoveryListener(RecoveryListener recoveryListener) {
this.recoveryListener = recoveryListener;
}
+
+ public AtomicInteger getRecoveryAttempts() {
+ return recoveryAttempts;
+ }
+
+ public AtomicInteger getRecoverySuccesses() {
+ return recoverySuccesses;
+ }
}
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=1222409&r1=1222408&r2=1222409&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 Thu Dec 22 19:43:08 2011
@@ -30,7 +30,6 @@ import java.util.concurrent.TimeoutExcep
import java.util.regex.Matcher;
import java.util.regex.Pattern;
-import org.apache.solr.cloud.RecoveryStrat.OnFinish;
import org.apache.solr.common.SolrException;
import org.apache.solr.common.SolrException.ErrorCode;
import org.apache.solr.common.cloud.*;
@@ -458,8 +457,7 @@ public final class ZkController {
* @throws Exception
*/
public String register(String coreName, final CoreDescriptor desc) throws Exception {
- final String shardUrl = localHostName + ":" + localHostPort + "/" + localHostContext
- + "/" + coreName;
+ final String shardUrl = getShardUrl(coreName);
final CloudDescriptor cloudDesc = desc.getCloudDescriptor();
final String collection = cloudDesc.getCollectionName();
@@ -562,14 +560,7 @@ public final class ZkController {
}
if (doRecovery) {
- recoveryStrat.recover(core, zkStateReader, shardUrl, new OnFinish() {
-
- @Override
- public void run() {
- // publish new props
- publishAsActive(shardUrl, cloudDesc, shardZkNodeName);
-
- }});
+ recoveryStrat.recover(core);
}
} finally {
if (core != null) {
@@ -584,7 +575,14 @@ public final class ZkController {
}
- private void publishAsActive(String shardUrl,
+ public String getShardUrl(String coreName) {
+ final String shardUrl = localHostName + ":" + localHostPort + "/" + localHostContext
+ + "/" + coreName;
+ return shardUrl;
+ }
+
+
+ void publishAsActive(String shardUrl,
final CloudDescriptor cloudDesc, String shardZkNodeName) {
Map<String,String> finalProps = new HashMap<String,String>();
finalProps.put(ZkStateReader.URL_PROP, shardUrl);
@@ -593,6 +591,16 @@ public final class ZkController {
finalProps.put(ZkStateReader.SHARD_ID_PROP, cloudDesc.getShardId());
publishState(cloudDesc, shardZkNodeName, finalProps);
}
+
+ void publishAsRecoverying(String shardUrl,
+ final CloudDescriptor cloudDesc, String shardZkNodeName) {
+ Map<String,String> finalProps = new HashMap<String,String>();
+ finalProps.put(ZkStateReader.URL_PROP, shardUrl);
+ finalProps.put(ZkStateReader.NODE_NAME_PROP, getNodeName());
+ finalProps.put(ZkStateReader.STATE_PROP, ZkStateReader.RECOVERING);
+ finalProps.put(ZkStateReader.SHARD_ID_PROP, cloudDesc.getShardId());
+ publishState(cloudDesc, shardZkNodeName, finalProps);
+ }
private boolean getShardId(final CoreDescriptor desc,
Modified: lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/handler/admin/CoreAdminHandler.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/handler/admin/CoreAdminHandler.java?rev=1222409&r1=1222408&r2=1222409&view=diff
==============================================================================
--- lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/handler/admin/CoreAdminHandler.java (original)
+++ lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/handler/admin/CoreAdminHandler.java Thu Dec 22 19:43:08 2011
@@ -22,11 +22,15 @@ import org.apache.lucene.index.IndexRead
import org.apache.lucene.store.Directory;
import org.apache.lucene.util.IOUtils;
import org.apache.solr.cloud.CloudDescriptor;
+import org.apache.solr.cloud.RecoveryStrat;
import org.apache.solr.common.SolrException;
+import org.apache.solr.common.SolrException.ErrorCode;
+import org.apache.solr.common.cloud.ZkNodeProps;
+import org.apache.solr.common.cloud.ZkStateReader;
import org.apache.solr.common.params.CoreAdminParams;
import org.apache.solr.common.params.CoreAdminParams.CoreAdminAction;
+import org.apache.solr.common.params.ModifiableSolrParams;
import org.apache.solr.common.params.SolrParams;
-import org.apache.solr.common.params.UpdateParams;
import org.apache.solr.common.util.NamedList;
import org.apache.solr.common.util.SimpleOrderedMap;
import org.apache.solr.core.*;
@@ -38,7 +42,9 @@ import org.apache.solr.search.SolrIndexS
import org.apache.solr.util.NumberUtils;
import org.apache.solr.util.RefCounted;
import org.apache.solr.util.SolrPluginUtils;
+import org.apache.solr.update.CommitUpdateCommand;
import org.apache.solr.update.MergeIndexesCommand;
+import org.apache.solr.update.processor.DistributedUpdateProcessor;
import org.apache.solr.update.processor.UpdateRequestProcessor;
import org.apache.solr.update.processor.UpdateRequestProcessorChain;
import org.slf4j.Logger;
@@ -161,6 +167,16 @@ public class CoreAdminHandler extends Re
break;
}
+ case PREPRECOVERY: {
+ this.handlePrepRecoveryAction(req, rsp);
+ break;
+ }
+
+ case REQUESTRECOVERY: {
+ this.handleRequestRecoveryAction(req, rsp);
+ break;
+ }
+
default: {
doPersist = this.handleCustomAction(req, rsp);
break;
@@ -554,6 +570,96 @@ public class CoreAdminHandler extends Re
return doPersist;
}
+
+ protected void handleRequestRecoveryAction(SolrQueryRequest req,
+ SolrQueryResponse rsp) throws IOException {
+ final SolrParams params = req.getParams();
+
+ String cname = params.get(CoreAdminParams.CORE);
+ if (cname == null) {
+ cname = "";
+ }
+
+ SolrCore core = coreContainer.getCore(cname);
+ try {
+ RecoveryStrat recoveryStrat = new RecoveryStrat();
+ recoveryStrat.recover(core);
+ } finally {
+ // no recoveryStrat close for now
+ core.close();
+ }
+ }
+
+ protected void handlePrepRecoveryAction(SolrQueryRequest req, SolrQueryResponse rsp) throws IOException {
+ final SolrParams params = req.getParams();
+
+ String cname = params.get(CoreAdminParams.CORE);
+ if (cname == null) {
+ cname = "";
+ }
+
+ String nodeName = params.get("nodeName");
+
+ SolrCore core = coreContainer.getCore(cname);
+ try {
+ String state;
+ while (true) {
+ // wait until we are sure the recovering node is ready
+ // to accept updates
+ CloudDescriptor cloudDescriptor = core.getCoreDescriptor()
+ .getCloudDescriptor();
+ ZkNodeProps nodeProps = coreContainer
+ .getZkController()
+ .getCloudState()
+ .getSlice(cloudDescriptor.getCollectionName(),
+ cloudDescriptor.getShardId()).getShards().get(nodeName);
+ state = nodeProps.get(ZkStateReader.STATE_PROP);
+ if (nodeProps != null && state.equals(ZkStateReader.RECOVERING)) {
+ // nocommit: this should time out
+ break;
+ }
+ }
+
+ if (core != null) {
+ // small safety net for any updates that started with state that
+ // kept it from sending the update to be buffered -
+ // pause for a while to let any outstanding updates finish
+ try {
+ Thread.sleep(1000);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+
+ UpdateRequestProcessorChain processorChain = core
+ .getUpdateProcessingChain(SolrPluginUtils.resolveUpdateChainParam(
+ params, log));
+ System.out.println("prep commit");
+ ModifiableSolrParams reqParams = new ModifiableSolrParams(
+ req.getParams());
+ reqParams.set(DistributedUpdateProcessor.COMMIT_END_POINT, "true");
+
+ SolrQueryRequest sqr = new LocalSolrQueryRequest(core, reqParams);
+ UpdateRequestProcessor processor = processorChain.createProcessor(sqr,
+ new SolrQueryResponse());
+ CommitUpdateCommand cuc = new CommitUpdateCommand(req, false);
+
+ processor.processCommit(cuc);
+ processor.finish();
+
+ RefCounted<SolrIndexSearcher> searcherRef = core.getSearcher();
+ System.out.println("approx docs to replicate:"
+ + searcherRef.get().getIndexReader().numDocs());
+ searcherRef.decref();
+ } else {
+ throw new SolrException(ErrorCode.BAD_REQUEST, "Could not find core: "
+ + core);
+ }
+ } finally {
+ if (core != null) {
+ core.close();
+ }
+ }
+ }
protected NamedList<Object> getCoreStatus(CoreContainer cores, String cname) throws IOException {
NamedList<Object> info = new SimpleOrderedMap<Object>();
Modified: lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/update/SolrCmdDistributor.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/update/SolrCmdDistributor.java?rev=1222409&r1=1222408&r2=1222409&view=diff
==============================================================================
--- lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/update/SolrCmdDistributor.java (original)
+++ lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/update/SolrCmdDistributor.java Thu Dec 22 19:43:08 2011
@@ -33,16 +33,13 @@ import java.util.concurrent.TimeUnit;
import org.apache.commons.httpclient.HttpClient;
import org.apache.commons.httpclient.MultiThreadedHttpConnectionManager;
-import org.apache.solr.client.solrj.SolrServer;
import org.apache.solr.client.solrj.impl.CommonsHttpSolrServer;
import org.apache.solr.client.solrj.request.AbstractUpdateRequest;
import org.apache.solr.client.solrj.request.UpdateRequestExt;
import org.apache.solr.common.SolrException;
-import org.apache.solr.common.SolrException.ErrorCode;
import org.apache.solr.common.params.ModifiableSolrParams;
import org.apache.solr.common.util.NamedList;
import org.apache.solr.core.SolrCore;
-import org.apache.solr.response.SolrQueryResponse;
// TODO: we are not really using the buffering anymore due to DistribUpdateProc...
// we might want to bring back a form of slots...
@@ -64,7 +61,7 @@ public class SolrCmdDistributor {
Set<Future<Request>> pending;
//private final SolrQueryRequest req;
- private final SolrQueryResponse rsp;
+ //private final SolrQueryResponse rsp;
//private final SchemaField idField;
@@ -74,70 +71,51 @@ public class SolrCmdDistributor {
private List<AddUpdateCommand> alist;
private ArrayList<DeleteUpdateCommand> dlist;
- public static class CmdRequest {
- public List<ShardInfo> shardInfos;
- public ModifiableSolrParams params;
- // we have to retry updates that are just being forwarded to the leader
- public boolean forwarding;
-
- public CmdRequest(List<ShardInfo> shardInfos, ModifiableSolrParams params) {
- this.shardInfos = shardInfos;
- this.params = params;
- }
-
- public CmdRequest(List<ShardInfo> shardInfos, ModifiableSolrParams params, boolean forwarding) {
- this(shardInfos, params);
- this.forwarding = forwarding;
- }
-
- public void updateUrlsForRetry() {
- for (ShardInfo shardInfo : shardInfos) {
- if (shardInfo.retryUrl != null) {
- shardInfo.url = shardInfo.retryUrl.getRetryUrl();
- }
- }
- }
- }
+ Response response = new Response();
public static class ShardInfo {
public String url;
public RetryUrl retryUrl;
+ @Override
+ public String toString() {
+ return "ShardInfo [url=" + url + "]";
+ }
}
public static interface RetryUrl {
String getRetryUrl();
}
- public SolrCmdDistributor(SolrQueryResponse rsp) {
- this.rsp = rsp;
+ public SolrCmdDistributor() {
+ //this.rsp = rsp;
}
- public void finish(CmdRequest cmdRequest) {
+ public Response finish(List<String> urls, ModifiableSolrParams params) {
// piggyback on any outstanding adds or deletes if possible.
- flushAdds(1, null, cmdRequest);
- flushDeletes(1, null, cmdRequest);
+ flushAdds(1, null, urls, params);
+ flushDeletes(1, null, urls, params);
- checkResponses(true, cmdRequest);
+ return checkResponses(true, urls);
}
- public void distribDelete(DeleteUpdateCommand cmd, CmdRequest cmdRequest) throws IOException {
- checkResponses(false, cmdRequest);
+ public void distribDelete(DeleteUpdateCommand cmd, List<String> urls, ModifiableSolrParams params) throws IOException {
+ checkResponses(false, urls);
if (cmd.isDeleteById()) {
- doDelete(cmd, cmdRequest);
+ doDelete(cmd, urls, params);
} else {
// TODO: query must be broadcast to all ??
- doDelete(cmd, cmdRequest);
+ doDelete(cmd, urls, params);
}
}
- public void distribAdd(AddUpdateCommand cmd, CmdRequest cmdRequest) throws IOException {
+ public void distribAdd(AddUpdateCommand cmd, List<String> urls, ModifiableSolrParams params) throws IOException {
- checkResponses(false, cmdRequest);
+ checkResponses(false, urls);
// make sure any pending deletes are flushed
- flushDeletes(1, null, cmdRequest);
+ flushDeletes(1, null, urls, params);
// TODO: this is brittle
// need to make a clone since these commands may be reused
@@ -158,48 +136,48 @@ public class SolrCmdDistributor {
}
alist.add(clone);
- flushAdds(maxBufferedAddsPerServer, null, cmdRequest);
+ flushAdds(maxBufferedAddsPerServer, null, urls, params);
}
- public void distribCommit(CommitUpdateCommand cmd, CmdRequest cmdRequest)
+ public void distribCommit(CommitUpdateCommand cmd, List<String> urls, ModifiableSolrParams params)
throws IOException {
// Wait for all outstanding repsonses to make sure that a commit
// can't sneak in ahead of adds or deletes we already sent.
// We could do this on a per-server basis, but it's more complex
// and this solution will lead to commits happening closer together.
- checkResponses(true, cmdRequest);
+ checkResponses(true, urls);
// piggyback on any outstanding adds or deletes if possible.
// TODO: review this
- flushAdds(1, cmd, cmdRequest);
+ flushAdds(1, cmd, urls, params);
- flushDeletes(1, cmd, cmdRequest);
+ flushDeletes(1, cmd, urls, params);
UpdateRequestExt ureq = new UpdateRequestExt();
- ureq.setParams(cmdRequest.params);
+ ureq.setParams(params);
addCommit(ureq, cmd);
- submit(ureq, cmdRequest);
+ submit(ureq, urls);
// if the command wanted to block until everything was committed,
// then do that here.
// nocommit
if (/* cmd.waitFlush || */cmd.waitSearcher) {
- checkResponses(true, cmdRequest);
+ checkResponses(true, urls);
}
}
- private void doDelete(DeleteUpdateCommand cmd, CmdRequest cmdRequest) throws IOException {
+ private void doDelete(DeleteUpdateCommand cmd, List<String> urls, ModifiableSolrParams params) throws IOException {
- flushAdds(1, null, cmdRequest);
+ flushAdds(1, null, urls, params);
if (dlist == null) {
dlist = new ArrayList<DeleteUpdateCommand>(2);
}
dlist.add(clone(cmd));
- flushDeletes(maxBufferedDeletesPerServer, null, cmdRequest);
+ flushDeletes(maxBufferedDeletesPerServer, null, urls, params);
}
void addCommit(UpdateRequestExt ureq, CommitUpdateCommand cmd) {
@@ -209,12 +187,12 @@ public class SolrCmdDistributor {
: AbstractUpdateRequest.ACTION.COMMIT, false, cmd.waitSearcher);
}
- boolean flushAdds(int limit, CommitUpdateCommand ccmd, CmdRequest cmdRequest) {
+ boolean flushAdds(int limit, CommitUpdateCommand ccmd, List<String> urls, ModifiableSolrParams params) {
// check for pending deletes
if (alist == null || alist.size() < limit) return false;
UpdateRequestExt ureq = new UpdateRequestExt();
- ureq.setParams(cmdRequest.params);
+ ureq.setParams(params);
addCommit(ureq, ccmd);
@@ -223,16 +201,16 @@ public class SolrCmdDistributor {
}
alist = null;
- submit(ureq, cmdRequest);
+ submit(ureq, urls);
return true;
}
- boolean flushDeletes(int limit, CommitUpdateCommand ccmd, CmdRequest cmdRequest) {
+ boolean flushDeletes(int limit, CommitUpdateCommand ccmd, List<String> urls, ModifiableSolrParams params) {
// check for pending deletes
if (dlist == null || dlist.size() < limit) return false;
UpdateRequestExt ureq = new UpdateRequestExt();
- ureq.setParams(cmdRequest.params);
+ ureq.setParams(params);
addCommit(ureq, ccmd);
@@ -245,7 +223,7 @@ public class SolrCmdDistributor {
}
dlist = null;
- submit(ureq, cmdRequest);
+ submit(ureq, urls);
return true;
}
@@ -257,9 +235,9 @@ public class SolrCmdDistributor {
return c;
}
- static class Request {
+ public static class Request {
// TODO: we may need to look at deep cloning this?
- CmdRequest cmdRequest;
+ public List<String> urls;
UpdateRequestExt ureq;
NamedList<Object> ursp;
int rspCode;
@@ -268,20 +246,20 @@ public class SolrCmdDistributor {
int retries = 0;
}
- void submit(UpdateRequestExt ureq, CmdRequest cmdRequest) {
+ void submit(UpdateRequestExt ureq, List<String> urls) {
Request sreq = new Request();
- sreq.cmdRequest = cmdRequest;
+ sreq.urls = urls;
sreq.ureq = ureq;
submit(sreq);
}
- void submit(final Request sreq) {
+ public void submit(final Request sreq) {
if (completionService == null) {
completionService = new ExecutorCompletionService<Request>(commExecutor);
pending = new HashSet<Future<Request>>();
}
- for (final ShardInfo shardInfo : sreq.cmdRequest.shardInfos) {
+ for (final String url : sreq.urls) {
// TODO: when we break up shards, we might forward
// to self again - makes things simple here, but we could
// also have realized this before, done the req locally, and
@@ -291,19 +269,24 @@ public class SolrCmdDistributor {
@Override
public Request call() throws Exception {
Request clonedRequest = new Request();
- clonedRequest.cmdRequest = sreq.cmdRequest;
+ clonedRequest.urls = sreq.urls;
clonedRequest.ureq = sreq.ureq;
- clonedRequest.url = shardInfo.url;
+ clonedRequest.url = url;
+ clonedRequest.retries = sreq.retries;
+
try {
// TODO: what about https?
- String url;
- if (!shardInfo.url.startsWith("http://")) {
- url = "http://" + shardInfo.url;
+ String fullUrl;
+ if (!url.startsWith("http://")) {
+ fullUrl = "http://" + url;
} else {
- url = shardInfo.url;
+ fullUrl = url;
}
- SolrServer server = new CommonsHttpSolrServer(url, client);
+ CommonsHttpSolrServer server = new CommonsHttpSolrServer(fullUrl, client);
+
+ //server.setSoTimeout(1000);
+ //server.setConnectionTimeout(1000);
clonedRequest.ursp = server.request(clonedRequest.ureq);
// currently no way to get the request body.
@@ -323,17 +306,14 @@ public class SolrCmdDistributor {
}
}
- void checkResponses(boolean block, CmdRequest cmdRequest) {
-
- int expectedResponses = pending == null ? 0 : pending.size();
- int nonConnectionErrors = 0;
- int failed = 0;
- Request failedFowardingRequest = null;
+ Response checkResponses(boolean block, List<String> urls) {
+
+
while (pending != null && pending.size() > 0) {
try {
Future<Request> future = block ? completionService.take()
: completionService.poll();
- if (future == null) return;
+ if (future == null) return null;
pending.remove(future);
try {
@@ -341,44 +321,13 @@ public class SolrCmdDistributor {
if (sreq.rspCode != 0) {
// error during request
Exception e = sreq.exception;
-
- // if it failed due to connect, assume we simply have not yet
- // learned it is down TODO: how about if we are cut off? Are we assuming too much?
- // the problem is there are other exceptions thrown due to a machine going down mid connection... I've
- // seen Interrupted exceptions.
-
- // nocommit:
- // we have to match against the msg...:(
- if (!e.getMessage().contains(
- "java.net.ConnectException: Connection refused")
- || e.getMessage().contains(
- "java.net.SocketException: Connection reset")) nonConnectionErrors++;
-
- failed++;
- // use the first exception encountered
- // TODO: perhaps we should do more?
- if (rsp.getException() == null) {
-
- String newMsg = "shard update error (" + sreq.cmdRequest.shardInfos + "):"
- + e.getMessage();
- if (e instanceof SolrException) {
- SolrException se = (SolrException) e;
- e = new SolrException(ErrorCode.getErrorCode(se.code()),
- newMsg, se.getCause());
- } else {
- e = new SolrException(SolrException.ErrorCode.SERVER_ERROR,
- newMsg, e);
- }
- rsp.setException(e);
- }
-
- if (cmdRequest.forwarding) {
- // this shold be fine because forwarding requests are only to one shard
- failedFowardingRequest = sreq;
- }
-
+ Error error = new Error();
+ error.e = e;
+ error.url = sreq.url;
+ response.errors.add(error);
+ response.sreq = sreq;
SolrException.logOnce(SolrCore.log, "shard update error " + sreq.url + " ("
- + sreq.cmdRequest.shardInfos + ")", sreq.exception);
+ + sreq.urls + ")", sreq.exception);
}
} catch (ExecutionException e) {
@@ -392,43 +341,18 @@ public class SolrCmdDistributor {
"interrupted waiting for shard update response", e);
}
}
-// if (failed > 0) {
-// System.out.println("expected:" + expectedResponses + " failed:" + failed + " failedAfterConnect:" + nonConnectionErrors + " forward:" + forwarding);
-// }
- // TODO: this is a somewhat weak success guarantee - if the request was successful on every replica considered up
- // and that does not return a connect exception, it was successful.
- //should we optionally fail when there is only a single leader for a shard? (no replication)
-
- // TODO: now we should tell those that failed to try and recover?
-
- if (failed > 0 && nonConnectionErrors == 0) {
- if (failed == expectedResponses && cmdRequest.forwarding) {
- // this is a pure forwarding request (single url for the leader) and it fully failed -
- // don't reset the exception - TODO: most likely there is now a new
- // leader - first we should retry the request...
-
- // TODO: we really need to clean this up and apis that allow it...
- if (failedFowardingRequest != null) {
- try {
- Thread.sleep(500);
- } catch (InterruptedException e1) {
- // TODO Auto-generated catch block
- e1.printStackTrace();
- }
- failedFowardingRequest.cmdRequest.updateUrlsForRetry();
- failedFowardingRequest.retries++;
-
- if (failedFowardingRequest.retries < 10) {
- rsp.setException(null);
- submit(failedFowardingRequest);
- checkResponses(block, cmdRequest);
- }
- }
- } else {
- // System.out.println("clear exception");
- rsp.setException(null);
- }
- }
+ return response;
+
+ }
+
+ public static class Response {
+ public Request sreq;
+ public List<Error> errors = new ArrayList<Error>();
+ }
+
+ public static class Error {
+ public String url;
+ public Exception e;
}
}
Modified: lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/update/TransactionLog.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/update/TransactionLog.java?rev=1222409&r1=1222408&r2=1222409&view=diff
==============================================================================
--- lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/update/TransactionLog.java (original)
+++ lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/update/TransactionLog.java Thu Dec 22 19:43:08 2011
@@ -23,7 +23,6 @@ import org.apache.solr.common.SolrInputD
import org.apache.solr.common.util.FastInputStream;
import org.apache.solr.common.util.FastOutputStream;
import org.apache.solr.common.util.JavaBinCodec;
-import org.apache.zookeeper.Transaction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Modified: lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/update/processor/DistributedUpdateProcessor.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/update/processor/DistributedUpdateProcessor.java?rev=1222409&r1=1222408&r2=1222409&view=diff
==============================================================================
--- lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/update/processor/DistributedUpdateProcessor.java (original)
+++ lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/update/processor/DistributedUpdateProcessor.java Thu Dec 22 19:43:08 2011
@@ -18,13 +18,19 @@ package org.apache.solr.update.processor
*/
import java.io.IOException;
+import java.net.MalformedURLException;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import org.apache.lucene.util.BytesRef;
import org.apache.lucene.util.CharsRef;
+import org.apache.solr.client.solrj.SolrServerException;
+import org.apache.solr.client.solrj.impl.CommonsHttpSolrServer;
+import org.apache.solr.client.solrj.request.CoreAdminRequest.PrepRecovery;
+import org.apache.solr.client.solrj.request.CoreAdminRequest.RequestRecovery;
import org.apache.solr.cloud.CloudDescriptor;
import org.apache.solr.cloud.ZkController;
import org.apache.solr.common.SolrException;
@@ -36,6 +42,7 @@ import org.apache.solr.common.cloud.ZkNo
import org.apache.solr.common.cloud.ZkStateReader;
import org.apache.solr.common.cloud.ZooKeeperException;
import org.apache.solr.common.params.ModifiableSolrParams;
+import org.apache.solr.common.params.CoreAdminParams.CoreAdminAction;
import org.apache.solr.common.util.Hash;
import org.apache.solr.common.util.NamedList;
import org.apache.solr.core.CoreDescriptor;
@@ -47,9 +54,7 @@ import org.apache.solr.update.AddUpdateC
import org.apache.solr.update.CommitUpdateCommand;
import org.apache.solr.update.DeleteUpdateCommand;
import org.apache.solr.update.SolrCmdDistributor;
-import org.apache.solr.update.SolrCmdDistributor.CmdRequest;
-import org.apache.solr.update.SolrCmdDistributor.RetryUrl;
-import org.apache.solr.update.SolrCmdDistributor.ShardInfo;
+import org.apache.solr.update.SolrCmdDistributor.Response;
import org.apache.solr.update.UpdateCommand;
import org.apache.solr.update.UpdateHandler;
import org.apache.solr.update.UpdateLog;
@@ -58,6 +63,7 @@ import org.apache.solr.update.VersionInf
import org.apache.zookeeper.KeeperException;
// NOT mt-safe... create a new processor for each add thread
+// TODO: we really should not wait for distrib after local? unless a certain replication factor is asked for
public class DistributedUpdateProcessor extends UpdateRequestProcessor {
public static final String SEEN_LEADER = "leader";
public static final String COMMIT_END_POINT = "commit_end_point";
@@ -128,8 +134,7 @@ public class DistributedUpdateProcessor
collection = cloudDesc.getCollectionName();
}
- cmdDistrib = new SolrCmdDistributor(rsp); // TODO: we put the last result (which could be complicated due to
- // multiple docs per req) in the rsp - this is whack
+ cmdDistrib = new SolrCmdDistributor();
}
private List<String> setupRequest(int hash) {
@@ -150,6 +155,10 @@ public class DistributedUpdateProcessor
collection, shardId);
String leaderUrl = leaderProps.get(ZkStateReader.URL_PROP);
+ if (leaderUrl == null) {
+ throw new SolrException(ErrorCode.SERVER_ERROR, "Cound could not leader url in:" + leaderUrl);
+ }
+
String leaderNodeName = leaderProps.get(ZkStateReader.NODE_NAME_PROP);
String nodeName = zkController.getNodeName();
@@ -170,7 +179,7 @@ public class DistributedUpdateProcessor
}
} catch (KeeperException e) {
- throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR, "",
+ throw new SolrException(SolrException.ErrorCode.SERVICE_UNAVAILABLE, "",
e);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
@@ -218,7 +227,7 @@ public class DistributedUpdateProcessor
if (isLeader) {
params.set(SEEN_LEADER, true);
}
- cmdDistrib.distribAdd(cmd, new CmdRequest(getShardInfos(urls), params, forwardToLeader));
+ cmdDistrib.distribAdd(cmd, urls, params);
} else {
// nocommit: At a minimum, local updates must be protected by synchronization
// right now we count on versionAdd to do the local add
@@ -237,9 +246,7 @@ public class DistributedUpdateProcessor
}
if (urls != null) {
- CmdRequest cmdRequest = new CmdRequest(getShardInfos(urls), params);
- cmdRequest.forwarding = forwardToLeader;
- cmdDistrib.finish(cmdRequest);
+ finish(params);
}
// TODO: keep track of errors? needs to be done at a higher level though since
@@ -247,38 +254,95 @@ public class DistributedUpdateProcessor
// Given that, it may also make sense to move the version reporting out of this
// processor too.
}
-
- private List<ShardInfo> getShardInfos(List<String> urls) {
- List<ShardInfo> shardInfos = new ArrayList<ShardInfo>(urls.size());
- for (String url : urls) {
- ShardInfo shardInfo = new ShardInfo();
- shardInfo.url = url;
- shardInfo.retryUrl = new RetryUrl() {
-
- @Override
- public String getRetryUrl() {
- // TODO: if we are now the leader, we forward through http...
- ZkNodeProps leaderProps = null;
- try {
- leaderProps = zkController.getZkStateReader().getLeaderProps(
- collection, shardId);
- } catch (InterruptedException e) {
- // TODO Auto-generated catch block
- e.printStackTrace();
- } catch (KeeperException e) {
- // TODO Auto-generated catch block
- e.printStackTrace();
+
+ // TODO: optionally fail if n replicas are not reached...
+ // nocommit: what the hell - doesnt seem to fail when cannot forward - need to check that...
+ private void finish(ModifiableSolrParams params) {
+ boolean retry = false;
+ int retries = 0;
+
+ Response response;
+ do {
+ retry = false;
+ response = cmdDistrib.finish(urls, params);
+ // nocommit - we may need to tell about more than one error...
+ if (response.errors.size() > 0) {
+ if (urls.size() == 1 && forwardToLeader) {
+ // we should retry a failed forward...
+ retry = true;
+ retries++;
+ if (retries > 10) {
+ // nocommit
+ log.error("we totally failed");
+ retry = false;
+ rsp.setException(response.errors.get(0).e);
+ } else {
+ try {
+ response.sreq.urls = Collections.singletonList(zkController
+ .getZkStateReader().getLeaderUrl(collection, shardId));
+ System.out.println("FORWARD FAILED:" + urls + " retry with "
+ + response.sreq.urls);
+ } catch (InterruptedException e2) {
+ Thread.currentThread().interrupt();
+ throw new SolrException(ErrorCode.SERVER_ERROR, e2);
+ } catch (KeeperException e2) {
+ throw new SolrException(ErrorCode.SERVER_ERROR, e2);
+ }
+ cmdDistrib.submit(response.sreq);
+
+ try {
+ Thread.sleep(100);
+ } catch (InterruptedException e1) {
+ Thread.currentThread().interrupt();
+ throw new SolrException(ErrorCode.SERVER_ERROR, "");
+ }
}
+ }
+ } else {
+ // System.out.println("success:" + urls);
+ rsp.setException(null);
+ }
+ } while(retry);
+
+ // if it is not a forward request, for each fail, try to tell them to
+ // recover
+ if (!forwardToLeader) {
+ for (SolrCmdDistributor.Error error : response.errors) {
+
+ // nocommit:
+ System.out.println("try and tell " + error.url + " to recover");
+ // TODO: we should force their state to recovering ??
+
+ // TODO: do retries??
+ // TODO: what if its is already recovering? Right now they line up - should they?
+ CommonsHttpSolrServer server;
+ try {
+ server = new CommonsHttpSolrServer(error.url);
- String leaderUrl = leaderProps.get(ZkStateReader.URL_PROP);
- return leaderUrl;
+ System.out.println("send recover cmd");
+ RequestRecovery recoverRequestCmd = new RequestRecovery();
+ recoverRequestCmd.setAction(CoreAdminAction.REQUESTRECOVERY);
+ // nocommit: the replica core name may not matcher the leader core
+ // name!
+ recoverRequestCmd.setCoreName(req.getCore().getName());
+
+ server.request(recoverRequestCmd);
+ System.out.println("send recover request worked");
+ } catch (MalformedURLException e) {
+ // nocommit
+ e.printStackTrace();
+ } catch (SolrServerException e) {
+ // nocommit
+ e.printStackTrace();
+ } catch (IOException e) {
+ // nocommit
+ e.printStackTrace();
}
- };
- shardInfos.add(shardInfo);
+ }
}
- return shardInfos;
}
+
// must be synchronized by bucket
private void doLocalAdd(AddUpdateCommand cmd) throws IOException {
super.processAdd(cmd);
@@ -419,15 +483,14 @@ public class DistributedUpdateProcessor
// TODO: do we need to add anything to the response?
return;
}
- CmdRequest cmdRequest = null;
+
ModifiableSolrParams params = null;
if (urls != null) {
params = new ModifiableSolrParams(req.getParams());
if (isLeader) {
params.set(SEEN_LEADER, true);
}
- cmdRequest = new CmdRequest(getShardInfos(urls), params, forwardToLeader);
- cmdDistrib.distribDelete(cmd, cmdRequest);
+ cmdDistrib.distribDelete(cmd, urls, params);
} else {
// super.processDelete(cmd);
}
@@ -444,7 +507,7 @@ public class DistributedUpdateProcessor
}
if (urls != null) {
- cmdDistrib.finish(cmdRequest);
+ finish(params);
}
}
@@ -601,7 +664,10 @@ public class DistributedUpdateProcessor
if (zkEnabled) {
ModifiableSolrParams params = new ModifiableSolrParams(req.getParams());
+ System.out.println("end point:" + params.getBool(COMMIT_END_POINT, false));
+ System.out.println("commit params:" + params);
if (!params.getBool(COMMIT_END_POINT, false)) {
+ System.out.println("distrib commit");
params.set(COMMIT_END_POINT, true);
String nodeName = req.getCore().getCoreDescriptor().getCoreContainer()
@@ -611,9 +677,9 @@ public class DistributedUpdateProcessor
.getCloudDescriptor().getCollectionName(), shardZkNodeName);
if (urls != null) {
- CmdRequest cmdRequest = new CmdRequest(getShardInfos(urls), params);
- cmdDistrib.distribCommit(cmd, cmdRequest);
- cmdDistrib.finish(cmdRequest);
+
+ cmdDistrib.distribCommit(cmd, urls, params);
+ finish(params);
}
}
}
Modified: lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/ChaosMonkey.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/ChaosMonkey.java?rev=1222409&r1=1222408&r2=1222409&view=diff
==============================================================================
--- lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/ChaosMonkey.java (original)
+++ lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/ChaosMonkey.java Thu Dec 22 19:43:08 2011
@@ -74,13 +74,20 @@ public class ChaosMonkey {
}
public void stopJetty(JettySolrRunner jetty) throws Exception {
+ if (jetty.isRunning()) {
+ stops.incrementAndGet();
+ }
+
stop(jetty);
- stops.incrementAndGet();
}
public void killJetty(JettySolrRunner jetty) throws Exception {
+ if (jetty.isRunning()) {
+ stops.incrementAndGet();
+ }
+
kill(jetty);
- stops.incrementAndGet();
+
}
public static void stop(JettySolrRunner jetty) throws Exception {
@@ -238,6 +245,7 @@ public class ChaosMonkey {
sleep(2000);
jetty.start();
}
+ System.out.println("started on port:" + jetty.getLocalPort());
starts.incrementAndGet();
continue;
}
@@ -245,11 +253,15 @@ public class ChaosMonkey {
JettySolrRunner jetty;
if (random.nextBoolean()) {
- System.out.println("looking to stop");
jetty = stopRandomShard();
+ if (jetty != null) {
+ System.out.println("looking to stop " + jetty.getLocalPort());
+ }
} else {
- System.out.println("looking to kill");
jetty = killRandomShard();
+ if (jetty != null) {
+ System.out.println("looking to kill " + jetty.getLocalPort());
+ }
}
if (jetty == null) {
System.out.println("we cannot kill");
@@ -273,4 +285,8 @@ public class ChaosMonkey {
stop = true;
}
+ public int getStarts() {
+ return starts.get();
+ }
+
}
\ No newline at end of file
Modified: lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/ChaosMonkeySolrCloudTest.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/ChaosMonkeySolrCloudTest.java?rev=1222409&r1=1222408&r2=1222409&view=diff
==============================================================================
--- lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/ChaosMonkeySolrCloudTest.java (original)
+++ lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/ChaosMonkeySolrCloudTest.java Thu Dec 22 19:43:08 2011
@@ -19,16 +19,21 @@ package org.apache.solr.cloud;
import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
import org.apache.solr.client.solrj.SolrQuery;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.cloud.CloudState;
+import org.apache.solr.common.cloud.Slice;
+import org.apache.solr.common.cloud.ZkNodeProps;
+import org.apache.solr.common.cloud.ZkStateReader;
+import org.apache.zookeeper.KeeperException;
import org.junit.BeforeClass;
-import org.junit.Ignore;
/**
* TODO: sometimes the shards are off by a doc or two, even with the
* retries on index failure...perhaps because of leader dying mid update?
*/
-@Ignore("test not ready to pass all the time")
public class ChaosMonkeySolrCloudTest extends FullSolrCloudTest {
@BeforeClass
@@ -49,16 +54,13 @@ public class ChaosMonkeySolrCloudTest ex
handle.put("QTime", SKIPVAL);
handle.put("timestamp", SKIPVAL);
- del("*:*");
-
List<StopableIndexingThread> threads = new ArrayList<StopableIndexingThread>();
for (int i = 0; i < 3; i++) {
- StopableIndexingThread indexThread = new StopableIndexingThread(0);
+ StopableIndexingThread indexThread = new StopableIndexingThread(i * 50000, true);
threads.add(indexThread);
indexThread.start();
}
-
chaosMonkey.startTheMonkey();
Thread.sleep(12000);
@@ -69,13 +71,31 @@ public class ChaosMonkeySolrCloudTest ex
indexThread.safeStop();
}
+ // wait for stop...
+ for (StopableIndexingThread indexThread : threads) {
+ indexThread.join();
+ }
+
+ Thread.sleep(2000);
+
+ for (StopableIndexingThread indexThread : threads) {
+ assertEquals(0, indexThread.getFails());
+ }
// try and wait for any replications and what not to finish...
- // TODO: I suppose we should poll zk here about state
- Thread.sleep(8000);
+
+ Thread.sleep(3000);
+
+ // wait until there are no recoveries...
+ waitForRecoveriesToFinish();
+
+ Thread.sleep(1000);
commit();
+ //assertEquals(chaosMonkey.getStarts(), getNumberOfRecoveryAttempts() - shardCount - sliceCount);
+
+
// does not always pass yet
checkShardConsistency();
@@ -88,5 +108,13 @@ public class ChaosMonkeySolrCloudTest ex
public void tearDown() throws Exception {
super.tearDown();
}
+
+ // skip the randoms - they can deadlock...
+ protected void indexr(Object... fields) throws Exception {
+ SolrInputDocument doc = new SolrInputDocument();
+ addFields(doc, fields);
+ addFields(doc, "rnd_b", true);
+ indexDoc(doc);
+ }
}
Modified: lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/FullSolrCloudTest.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/FullSolrCloudTest.java?rev=1222409&r1=1222408&r2=1222409&view=diff
==============================================================================
--- lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/FullSolrCloudTest.java (original)
+++ lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/FullSolrCloudTest.java Thu Dec 22 19:43:08 2011
@@ -17,17 +17,22 @@ package org.apache.solr.cloud;
* limitations under the License.
*/
-import java.io.File;
import java.io.IOException;
import java.net.MalformedURLException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorCompletionService;
+import java.util.concurrent.Future;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
@@ -63,7 +68,11 @@ public class FullSolrCloudTest extends A
private static final String SHARD2 = "shard2";
- private static final String DEFAULT_COLLECTION = "collection1";
+ protected static final String DEFAULT_COLLECTION = "collection1";
+
+ static ThreadPoolExecutor JETTY_EXECUTOR = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 5, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());
+ ExecutorCompletionService completionService = new ExecutorCompletionService<JettySolrRunner>(JETTY_EXECUTOR);
+ Set<Future<JettySolrRunner>> pending = new HashSet<Future<JettySolrRunner>>();
String t1="a_t";
String i1="a_si";
@@ -89,7 +98,7 @@ public class FullSolrCloudTest extends A
protected Map<CloudSolrServerClient,ZkNodeProps> clientToInfo = new HashMap<CloudSolrServerClient,ZkNodeProps>();
protected Map<String,List<SolrServer>> shardToClient = new HashMap<String,List<SolrServer>>();
protected Map<String,List<CloudJettyRunner>> shardToJetty = new HashMap<String,List<CloudJettyRunner>>();
- private AtomicInteger i = new AtomicInteger(0);
+ private AtomicInteger jettyIntCntr = new AtomicInteger(0);
protected ChaosMonkey chaosMonkey;
protected volatile ZkStateReader zkStateReader;
@@ -133,8 +142,6 @@ public class FullSolrCloudTest extends A
}
-
-
@Before
@Override
public void setUp() throws Exception {
@@ -228,17 +235,19 @@ public class FullSolrCloudTest extends A
StringBuilder sb = new StringBuilder();
for (int i = 1; i <= numJettys; i++) {
if (sb.length() > 0) sb.append(',');
- JettySolrRunner j = createJetty(testDir, testDir + "/jetty" + this.i.incrementAndGet(), null, "solrconfig.xml");
+ JettySolrRunner j = createJetty(testDir, testDir + "/jetty" + this.jettyIntCntr.incrementAndGet(), null, "solrconfig.xml");
jettys.add(j);
SolrServer client = createNewSolrServer(j.getLocalPort());
clients.add(client);
}
initCloud();
- updateMappingsFromZk(jettys, clients);
this.jettys.addAll(jettys);
this.clients.addAll(clients);
+
+ updateMappingsFromZk(this.jettys, this.clients);
+
// build the shard string
for (int i = 1; i <= numJettys/2; i++) {
JettySolrRunner j = this.jettys.get(i);
@@ -248,10 +257,13 @@ public class FullSolrCloudTest extends A
sb.append("|localhost:").append(j2.getLocalPort()).append(context);
}
shards = sb.toString();
+
+ waitForRecoveriesToFinish();
+
return jettys;
}
- public JettySolrRunner createJetty(File baseDir, String dataDir, String shardList, String solrConfigOverride) throws Exception {
+ public JettySolrRunner createJetty(String dataDir, String shardList, String solrConfigOverride) throws Exception {
JettySolrRunner jetty = new JettySolrRunner(getSolrHome(), "/solr", 0, solrConfigOverride, false);
jetty.setShards(shardList);
@@ -265,13 +277,9 @@ public class FullSolrCloudTest extends A
List<SolrServer> clients) throws Exception,
IOException, KeeperException, URISyntaxException {
zkStateReader.updateCloudState(true);
-
- while(!zkStateReader.getCloudState().getCollections().contains(DEFAULT_COLLECTION)) {
- Thread.sleep(500);
- }
- while(zkStateReader.getCloudState().getSlices(DEFAULT_COLLECTION).size() != sliceCount) {
- Thread.sleep(500);
- }
+ shardToClient.clear();
+ shardToJetty.clear();
+ jettyToInfo.clear();
for (SolrServer client : clients) {
// find info for this client in zk
@@ -334,6 +342,18 @@ public class FullSolrCloudTest extends A
}
}
+ protected int getNumberOfRecoveryAttempts() {
+ int cnt = 0;
+ for (JettySolrRunner jetty : jettys) {
+ RecoveryStrat recoveryStrat = ((SolrDispatchFilter) jetty.getDispatchFilter().getFilter()).getCores()
+ .getZkController().getRecoveryStrat();
+ cnt += recoveryStrat.getRecoveryAttempts().get();
+ }
+
+
+ return cnt;
+ }
+
@Override
protected void setDistributedParams(ModifiableSolrParams params) {
@@ -549,6 +569,50 @@ public class FullSolrCloudTest extends A
assertDocCounts(VERBOSE);
}
+ protected void waitForRecoveriesToFinish() throws KeeperException,
+ InterruptedException {
+ boolean cont = true;
+ int cnt = 0;
+
+ System.out.println("WAIT FOR RECOVERY");
+
+ while (cont) {
+ System.out.println("-");
+ boolean sawLiveRecovering = false;
+ zkStateReader.updateCloudState(true);
+ CloudState cloudState = zkStateReader.getCloudState();
+ Map<String,Slice> slices = cloudState.getSlices(DEFAULT_COLLECTION);
+ for (Map.Entry<String,Slice> entry : slices.entrySet()) {
+ Map<String,ZkNodeProps> shards = entry.getValue().getShards();
+ for (Map.Entry<String,ZkNodeProps> shard : shards.entrySet()) {
+ System.out.println("rstate:"
+ + shard.getValue().get(ZkStateReader.STATE_PROP)
+ + " live:"
+ + cloudState.liveNodesContain(shard.getValue().get(
+ ZkStateReader.NODE_NAME_PROP)));
+ if (shard.getValue().get(ZkStateReader.STATE_PROP)
+ .equals(ZkStateReader.RECOVERING)
+ && cloudState.liveNodesContain(shard.getValue().get(
+ ZkStateReader.NODE_NAME_PROP))) {
+ sawLiveRecovering = true;
+ }
+ }
+ }
+ if (!sawLiveRecovering || cnt == 90) {
+ if (!sawLiveRecovering) {
+ System.out.println("no one is recoverying");
+ } else {
+ System.out.println("gave up waiting for recovery to finish..");
+ }
+ cont = false;
+ } else {
+ Thread.sleep(2000);
+ }
+ cnt++;
+ }
+ System.out.println("DONE WAIT FOR RECOVERY");
+ }
+
private void brindDownShardIndexSomeDocsAndRecover() throws Exception,
SolrServerException, IOException, InterruptedException {
// kill a shard
@@ -776,12 +840,17 @@ public class FullSolrCloudTest extends A
}
}
- protected void checkShardConsistency(String shard) throws SolrServerException {
+ protected void checkShardConsistency(String shard) throws Exception {
+
List<SolrServer> solrClients = shardToClient.get(shard);
+ if (solrClients == null) {
+ throw new RuntimeException("shard not found:" + shard + " keys:" + shardToClient.keySet());
+ }
long num = -1;
long lastNum = -1;
String failMessage = null;
System.out.println("check const of " + shard);
+ int cnt = 0;
for (SolrServer client : solrClients) {
try {
num = client.query(new SolrQuery("*:*")).getResults().getNumFound();
@@ -789,11 +858,20 @@ public class FullSolrCloudTest extends A
System.err.println("error contacting client:" + e.getMessage());
continue;
}
- System.out.println(" num:" + num + "\n");
- if (lastNum > -1 && lastNum != num && failMessage == null) {
- failMessage = "shard is not consistent, expected:" + lastNum + " and got:" + num;
+ ZkNodeProps props = clientToInfo.get(new CloudSolrServerClient(client));
+ System.out.println("client" + cnt++);
+ System.out.println("PROPS:" + props);
+
+ boolean recovering = props.get(ZkStateReader.STATE_PROP).equals(ZkStateReader.RECOVERING);
+ System.out.println(" num:" + num + "\n" + (recovering ? "recovering" : ""));
+
+ if (!recovering) {
+ if (lastNum > -1 && lastNum != num && failMessage == null) {
+ failMessage = "shard is not consistent, expected:" + lastNum
+ + " and got:" + num;
+ }
+ lastNum = num;
}
- lastNum = num;
}
if (failMessage != null) {
@@ -802,22 +880,33 @@ public class FullSolrCloudTest extends A
}
- protected void checkShardConsistency() throws SolrServerException {
+ protected void checkShardConsistency() throws Exception {
+ long docs = controlClient.query(new SolrQuery("*:*")).getResults().getNumFound();
+ System.out.println("Control Docs:" + docs);
+
+ updateMappingsFromZk(jettys, clients);
+
Set<String> theShards = shardToClient.keySet();
for (String shard : theShards) {
checkShardConsistency(shard);
}
// now check that the right # are on each shard
- long docs = controlClient.query(new SolrQuery("*:*")).getResults().getNumFound();
theShards = shardToClient.keySet();
int cnt = 0;
for (String s : theShards) {
int times = shardToClient.get(s).size();
for (int i = 0; i < times; i++) {
try {
- cnt += shardToClient.get(s).get(i).query(new SolrQuery("*:*")).getResults().getNumFound();
- break;
+ SolrServer client = shardToClient.get(s).get(i);
+ ZkNodeProps props = clientToInfo.get(new CloudSolrServerClient(client));
+ System.out.println("PROPS:" + props);
+ boolean recovering = props.get(ZkStateReader.STATE_PROP).equals(ZkStateReader.RECOVERING);
+ if (!recovering) {
+ cnt += client.query(new SolrQuery("*:*")).getResults()
+ .getNumFound();
+ break;
+ }
} catch(SolrServerException e) {
// if we have a problem, try the next one
if (i == times - 1) {
@@ -932,10 +1021,13 @@ public class FullSolrCloudTest extends A
class StopableIndexingThread extends Thread {
private volatile boolean stop = false;
private int startI;
- private List<Integer> deletes = new ArrayList<Integer>();
+ private List<Integer> deletes = new ArrayList<Integer>();
+ private int fails;
+ private boolean doDeletes;
- public StopableIndexingThread(int startI) {
+ public StopableIndexingThread(int startI, boolean doDeletes) {
this.startI = startI;
+ this.doDeletes = doDeletes;
setDaemon(true);
}
@@ -943,16 +1035,18 @@ public class FullSolrCloudTest extends A
public void run() {
int i = startI;
int numDeletes = 0;
- int fails = 0;
+ int numAdds = 0;
+
while (true && !stop) {
++i;
- if (random.nextBoolean() && deletes.size() > 0) {
+
+ if (doDeletes && random.nextBoolean() && deletes.size() > 0) {
Integer delete = deletes.remove(0);
try {
+ numDeletes++;
controlClient.deleteById(Integer.toString(delete));
cloudClient.deleteById(Integer.toString(delete));
- numDeletes++;
} catch (Exception e) {
System.err.println("REQUEST FAILED:");
e.printStackTrace();
@@ -962,6 +1056,7 @@ public class FullSolrCloudTest extends A
}
try {
+ numAdds++;
indexr(id, i, i1, 50, tlong, 50, t1,
"to come to the aid of their country.");
} catch (Exception e) {
@@ -970,18 +1065,22 @@ public class FullSolrCloudTest extends A
fails++;
}
- if (random.nextBoolean()) {
+ if (doDeletes && random.nextBoolean()) {
deletes.add(i);
}
}
- System.err.println("added docs:" + i + " with " + fails + " fails" + " deletes:" + numDeletes);
+ System.err.println("added docs:" + numAdds + " with " + fails + " fails" + " deletes:" + numDeletes);
}
public void safeStop() {
stop = true;
}
+
+ public int getFails() {
+ return fails;
+ }
};
Modified: lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/RecoveryZkTest.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/RecoveryZkTest.java?rev=1222409&r1=1222408&r2=1222409&view=diff
==============================================================================
--- lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/RecoveryZkTest.java (original)
+++ lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/RecoveryZkTest.java Thu Dec 22 19:43:08 2011
@@ -30,8 +30,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * TODO: every now and then, this leaves a dir open for some odd reason
- *
* TODO: Why do we see add fails but it still matches control? Because its successful
* locally and gets picked up in distrib search...
*/
@@ -65,10 +63,10 @@ public class RecoveryZkTest extends Full
// start a couple indexing threads
- StopableIndexingThread indexThread = new StopableIndexingThread(0);
+ StopableIndexingThread indexThread = new StopableIndexingThread(0, true);
indexThread.start();
- StopableIndexingThread indexThread2 = new StopableIndexingThread(10000);
+ StopableIndexingThread indexThread2 = new StopableIndexingThread(10000, true);
indexThread2.start();
@@ -120,17 +118,18 @@ public class RecoveryZkTest extends Full
//ureq.setParam("update.chain", DISTRIB_UPDATE_CHAIN);
ureq.process(cloudClient);
}
+
+
+ @Override
+ public void tearDown() throws Exception {
+ super.tearDown();
+ }
+ // skip the randoms - they can deadlock...
protected void indexr(Object... fields) throws Exception {
SolrInputDocument doc = new SolrInputDocument();
addFields(doc, fields);
addFields(doc, "rnd_b", true);
indexDoc(doc);
}
-
- @Override
- public void tearDown() throws Exception {
- super.tearDown();
- }
-
}
Modified: lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/ZkTestServer.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/ZkTestServer.java?rev=1222409&r1=1222408&r2=1222409&view=diff
==============================================================================
--- lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/ZkTestServer.java (original)
+++ lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/ZkTestServer.java Thu Dec 22 19:43:08 2011
@@ -33,7 +33,7 @@ import javax.management.JMException;
import org.apache.solr.SolrTestCaseJ4;
import org.apache.zookeeper.jmx.ManagedUtil;
-import org.apache.zookeeper.server.ServerCnxnFactory;
+import org.apache.zookeeper.server.NIOServerCnxn;
import org.apache.zookeeper.server.ServerConfig;
import org.apache.zookeeper.server.SessionTracker.Session;
import org.apache.zookeeper.server.ZKDatabase;
@@ -56,7 +56,7 @@ public class ZkTestServer {
class ZKServerMain {
- private ServerCnxnFactory cnxnFactory;
+ private NIOServerCnxn.Factory cnxnFactory;
private ZooKeeperServer zooKeeperServer;
protected void initializeAndRun(String[] args) throws ConfigException,
@@ -83,30 +83,26 @@ public class ZkTestServer {
* @throws IOException
*/
public void runFromConfig(ServerConfig config) throws IOException {
- try {
- // Note that this thread isn't going to be doing anything else,
- // so rather than spawning another thread, we will just call
- // run() in this thread.
- // create a file logger url from the command line args
- zooKeeperServer = new ZooKeeperServer();
-
- FileTxnSnapLog ftxn = new FileTxnSnapLog(new
- File(config.getDataLogDir()), new File(config.getDataDir()));
- zooKeeperServer.setTxnLogFactory(ftxn);
- zooKeeperServer.setTickTime(config.getTickTime());
- zooKeeperServer.setMinSessionTimeout(config.getMinSessionTimeout());
- zooKeeperServer.setMaxSessionTimeout(config.getMaxSessionTimeout());
- cnxnFactory = ServerCnxnFactory.createFactory();
- cnxnFactory.configure(config.getClientPortAddress(),
- config.getMaxClientCnxns());
- cnxnFactory.startup(zooKeeperServer);
- cnxnFactory.join();
- if (zooKeeperServer.isRunning()) {
- zkServer.shutdown();
- }
- } catch (InterruptedException e) {
- // should warn, but generally this is ok
+ try {
+ // Note that this thread isn't going to be doing anything else,
+ // so rather than spawning another thread, we will just call
+ // run() in this thread.
+ // create a file logger url from the command line args
+ zooKeeperServer = new ZooKeeperServer();
+
+ FileTxnSnapLog ftxn = new FileTxnSnapLog(new File(config
+ .getDataLogDir()), new File(config.getDataDir()));
+ zooKeeperServer.setTxnLogFactory(ftxn);
+ zooKeeperServer.setTickTime(config.getTickTime());
+ cnxnFactory = new NIOServerCnxn.Factory(config.getClientPortAddress(), config
+ .getMaxClientCnxns());
+ cnxnFactory.startup(zooKeeperServer);
+ cnxnFactory.join();
+ if (zooKeeperServer.isRunning()) {
+ zooKeeperServer.shutdown();
}
+ } catch (InterruptedException e) {
+ }
}
/**
Added: lucene/dev/branches/solrcloud/solr/lib/zookeeper-3.3.4.jar
URL: http://svn.apache.org/viewvc/lucene/dev/branches/solrcloud/solr/lib/zookeeper-3.3.4.jar?rev=1222409&view=auto
==============================================================================
Binary file - no diff available.
Modified: lucene/dev/branches/solrcloud/solr/solrj/src/java/org/apache/solr/client/solrj/impl/LBHttpSolrServer.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/solrcloud/solr/solrj/src/java/org/apache/solr/client/solrj/impl/LBHttpSolrServer.java?rev=1222409&r1=1222408&r2=1222409&view=diff
==============================================================================
--- lucene/dev/branches/solrcloud/solr/solrj/src/java/org/apache/solr/client/solrj/impl/LBHttpSolrServer.java (original)
+++ lucene/dev/branches/solrcloud/solr/solrj/src/java/org/apache/solr/client/solrj/impl/LBHttpSolrServer.java Thu Dec 22 19:43:08 2011
@@ -28,6 +28,7 @@ import org.apache.solr.common.SolrExcept
import java.io.IOException;
import java.lang.ref.WeakReference;
import java.net.MalformedURLException;
+import java.net.SocketException;
import java.net.URL;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
@@ -249,16 +250,21 @@ public class LBHttpSolrServer extends So
rsp.rsp = server.request(req.getRequest());
return rsp; // SUCCESS
} catch (SolrException e) {
- // Server is alive but the request was malformed or invalid
- throw e;
+ System.out
+ .println("root: " + new SolrServerException(e).getRootCause());
+ if (e.code() == 404 || e.code() == 503
+ || e.getMessage().contains("java.net.SocketException")
+ || e.getMessage().contains("java.net.ConnectException")) {
+ ex = addZombie(server, e);
+ } else {
+ // Server is alive but the request was malformed or invalid
+ throw e;
+ }
+ } catch (SocketException e) {
+ ex = addZombie(server, e);
} catch (SolrServerException e) {
if (e.getRootCause() instanceof IOException) {
- ex = e;
- wrapper = new ServerWrapper(server);
- wrapper.lastUsed = System.currentTimeMillis();
- wrapper.standard = false;
- zombieServers.put(wrapper.getKey(), wrapper);
- startAliveCheckExecutor();
+ ex = addZombie(server, e);
} else {
throw e;
}
@@ -296,6 +302,19 @@ public class LBHttpSolrServer extends So
throw new SolrServerException("No live SolrServers available to handle this request", ex);
}
+ }
+
+ private Exception addZombie(CommonsHttpSolrServer server,
+ Exception e) {
+
+ ServerWrapper wrapper;
+
+ wrapper = new ServerWrapper(server);
+ wrapper.lastUsed = System.currentTimeMillis();
+ wrapper.standard = false;
+ zombieServers.put(wrapper.getKey(), wrapper);
+ startAliveCheckExecutor();
+ return e;
}
Modified: lucene/dev/branches/solrcloud/solr/solrj/src/java/org/apache/solr/client/solrj/request/CoreAdminRequest.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/solrcloud/solr/solrj/src/java/org/apache/solr/client/solrj/request/CoreAdminRequest.java?rev=1222409&r1=1222408&r2=1222409&view=diff
==============================================================================
--- lucene/dev/branches/solrcloud/solr/solrj/src/java/org/apache/solr/client/solrj/request/CoreAdminRequest.java (original)
+++ lucene/dev/branches/solrcloud/solr/solrj/src/java/org/apache/solr/client/solrj/request/CoreAdminRequest.java Thu Dec 22 19:43:08 2011
@@ -89,6 +89,58 @@ public class CoreAdminRequest extends So
return params;
}
}
+
+ public static class PrepRecovery extends CoreAdminRequest {
+ protected String nodeName;
+
+
+ public PrepRecovery() {
+ action = CoreAdminAction.PREPRECOVERY;
+ }
+
+ public void setNodeName(String nodeName) { this.nodeName = nodeName; }
+
+ public String getNodeName() { return nodeName; }
+
+
+ @Override
+ public SolrParams getParams() {
+ if( action == null ) {
+ throw new RuntimeException( "no action specified!" );
+ }
+ ModifiableSolrParams params = new ModifiableSolrParams();
+ params.set( CoreAdminParams.ACTION, action.toString() );
+
+ params.set( CoreAdminParams.CORE, core );
+
+ if (nodeName != null) {
+ params.set( "nodeName", nodeName);
+ }
+
+ return params;
+ }
+ }
+
+ public static class RequestRecovery extends CoreAdminRequest {
+
+ public RequestRecovery() {
+ action = CoreAdminAction.REQUESTRECOVERY;
+ }
+
+ @Override
+ public SolrParams getParams() {
+ if( action == null ) {
+ throw new RuntimeException( "no action specified!" );
+ }
+ ModifiableSolrParams params = new ModifiableSolrParams();
+ params.set( CoreAdminParams.ACTION, action.toString() );
+
+ params.set( CoreAdminParams.CORE, core );
+
+ return params;
+ }
+ }
+
//a persist core request
public static class Persist extends CoreAdminRequest {
protected String fileName = null;
Modified: lucene/dev/branches/solrcloud/solr/solrj/src/java/org/apache/solr/common/params/CoreAdminParams.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/solrcloud/solr/solrj/src/java/org/apache/solr/common/params/CoreAdminParams.java?rev=1222409&r1=1222408&r2=1222409&view=diff
==============================================================================
--- lucene/dev/branches/solrcloud/solr/solrj/src/java/org/apache/solr/common/params/CoreAdminParams.java (original)
+++ lucene/dev/branches/solrcloud/solr/solrj/src/java/org/apache/solr/common/params/CoreAdminParams.java Thu Dec 22 19:43:08 2011
@@ -86,7 +86,9 @@ public interface CoreAdminParams
RENAME,
@Deprecated
ALIAS,
- MERGEINDEXES;
+ MERGEINDEXES,
+ PREPRECOVERY,
+ REQUESTRECOVERY;
public static CoreAdminAction get( String p )
{
Modified: lucene/dev/branches/solrcloud/solr/solrj/src/java/org/apache/zookeeper/SolrZooKeeper.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/solrcloud/solr/solrj/src/java/org/apache/zookeeper/SolrZooKeeper.java?rev=1222409&r1=1222408&r2=1222409&view=diff
==============================================================================
--- lucene/dev/branches/solrcloud/solr/solrj/src/java/org/apache/zookeeper/SolrZooKeeper.java (original)
+++ lucene/dev/branches/solrcloud/solr/solrj/src/java/org/apache/zookeeper/SolrZooKeeper.java Thu Dec 22 19:43:08 2011
@@ -1,6 +1,7 @@
package org.apache.zookeeper;
import java.io.IOException;
+import java.nio.channels.SocketChannel;
// nocommit - we use this class to expose nasty stuff for tests
public class SolrZooKeeper extends ZooKeeper {
@@ -20,21 +21,21 @@ public class SolrZooKeeper extends ZooKe
* @param ms the number of milliseconds to pause.
*/
public void pauseCnxn(final long ms) {
- new Thread() {
- public void run() {
- synchronized(cnxn) {
- try {
- try {
- cnxn.sendThread.testableCloseSocket();
- } catch (IOException e) {
- e.printStackTrace();
- }
- Thread.sleep(ms);
- } catch (InterruptedException e) {
- }
- }
- }
- }.start();
+ new Thread() {
+ public void run() {
+ synchronized (cnxn) {
+ try {
+ try {
+ ((SocketChannel) cnxn.sendThread.sockKey.channel()).socket()
+ .close();
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ Thread.sleep(ms);
+ } catch (InterruptedException e) {}
+ }
+ }
+ }.start();
}
-
+
}