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/16 19:43:59 UTC
svn commit: r1215243 - in /lucene/dev/branches/solrcloud/solr/core/src:
java/org/apache/solr/update/ java/org/apache/solr/update/processor/
test/org/apache/solr/cloud/
Author: markrmiller
Date: Fri Dec 16 18:43:59 2011
New Revision: 1215243
URL: http://svn.apache.org/viewvc?rev=1215243&view=rev
Log:
take forwarding into account when deciding if a response failed or not
Modified:
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/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/ChaosMonkeyDistributedZkTest.java
lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/FullDistributedZkTest.java
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=1215243&r1=1215242&r2=1215243&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 Fri Dec 16 18:43:59 2011
@@ -79,32 +79,32 @@ public class SolrCmdDistributor {
this.rsp = rsp;
}
- public void finish(List<String> shards, ModifiableSolrParams params) {
+ public void finish(List<String> shards, ModifiableSolrParams params, boolean forwarding) {
// piggyback on any outstanding adds or deletes if possible.
- flushAdds(1, null, shards, params);
- flushDeletes(1, null, shards, params);
+ flushAdds(1, null, shards, params, forwarding);
+ flushDeletes(1, null, shards, params, forwarding);
- checkResponses(true);
+ checkResponses(true, forwarding);
}
- public void distribDelete(DeleteUpdateCommand cmd, List<String> shards, ModifiableSolrParams params) throws IOException {
- checkResponses(false);
+ public void distribDelete(DeleteUpdateCommand cmd, List<String> shards, ModifiableSolrParams params, boolean forwarding) throws IOException {
+ checkResponses(false, forwarding);
if (cmd.isDeleteById()) {
- doDelete(cmd, shards, params);
+ doDelete(cmd, shards, params, forwarding);
} else {
// TODO: query must be broadcast to all ??
- doDelete(cmd, shards, params);
+ doDelete(cmd, shards, params, forwarding);
}
}
- public void distribAdd(AddUpdateCommand cmd, List<String> shards, ModifiableSolrParams params) throws IOException {
+ public void distribAdd(AddUpdateCommand cmd, List<String> shards, ModifiableSolrParams params, boolean forwarding) throws IOException {
- checkResponses(false);
+ checkResponses(false, forwarding);
// make sure any pending deletes are flushed
- flushDeletes(1, null, shards, params);
+ flushDeletes(1, null, shards, params, forwarding);
// TODO: this is brittle
// need to make a clone since these commands may be reused
@@ -125,7 +125,7 @@ public class SolrCmdDistributor {
}
alist.add(clone);
- flushAdds(maxBufferedAddsPerServer, null, shards, params);
+ flushAdds(maxBufferedAddsPerServer, null, shards, params, forwarding);
}
public void distribCommit(CommitUpdateCommand cmd, List<String> shards, ModifiableSolrParams params)
@@ -135,13 +135,13 @@ public class SolrCmdDistributor {
// 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);
+ checkResponses(true, false);
// piggyback on any outstanding adds or deletes if possible.
// TODO: review this
- flushAdds(1, cmd, shards, params);
+ flushAdds(1, cmd, shards, params, false);
- flushDeletes(1, cmd, shards, params);
+ flushDeletes(1, cmd, shards, params, false);
UpdateRequestExt ureq = new UpdateRequestExt();
ureq.setParams(params);
@@ -153,20 +153,20 @@ public class SolrCmdDistributor {
// then do that here.
// nocommit
if (/* cmd.waitFlush || */cmd.waitSearcher) {
- checkResponses(true);
+ checkResponses(true, false);
}
}
- private void doDelete(DeleteUpdateCommand cmd, List<String> shards, ModifiableSolrParams params) throws IOException {
+ private void doDelete(DeleteUpdateCommand cmd, List<String> shards, ModifiableSolrParams params, boolean forwarding) throws IOException {
- flushAdds(1, null, shards, params);
+ flushAdds(1, null, shards, params, forwarding);
if (dlist == null) {
dlist = new ArrayList<DeleteUpdateCommand>(2);
}
dlist.add(clone(cmd));
- flushDeletes(maxBufferedDeletesPerServer, null, shards, params);
+ flushDeletes(maxBufferedDeletesPerServer, null, shards, params, forwarding);
}
void addCommit(UpdateRequestExt ureq, CommitUpdateCommand cmd) {
@@ -176,7 +176,7 @@ public class SolrCmdDistributor {
: AbstractUpdateRequest.ACTION.COMMIT, false, cmd.waitSearcher);
}
- boolean flushAdds(int limit, CommitUpdateCommand ccmd, List<String> urls, ModifiableSolrParams params) {
+ boolean flushAdds(int limit, CommitUpdateCommand ccmd, List<String> urls, ModifiableSolrParams params, boolean forwarding) {
// check for pending deletes
if (alist == null || alist.size() < limit) return false;
@@ -194,7 +194,7 @@ public class SolrCmdDistributor {
return true;
}
- boolean flushDeletes(int limit, CommitUpdateCommand ccmd, List<String> shards, ModifiableSolrParams params) {
+ boolean flushDeletes(int limit, CommitUpdateCommand ccmd, List<String> shards, ModifiableSolrParams params, boolean forwarding) {
// check for pending deletes
if (dlist == null || dlist.size() < limit) return false;
@@ -289,7 +289,7 @@ public class SolrCmdDistributor {
}
}
- void checkResponses(boolean block) {
+ void checkResponses(boolean block, boolean forwarding) {
int expectedResponses = pending == null ? 0 : pending.size();
int nonConnectionErrors = 0;
@@ -352,17 +352,24 @@ public class SolrCmdDistributor {
"interrupted waiting for shard update response", e);
}
}
- if (failed > 0) {
- System.out.println("expected:" + expectedResponses + " failed:" + failed + " failedAfterConnect:" + nonConnectionErrors);
- }
+// 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) {
- System.out.println("clear exception");
- rsp.setException(null);
+ if (failed == expectedResponses && forwarding) {
+ // this is a pure forwarding request and it fully failed -
+ // don't reset the exception - TODO: most likely there is now a new
+ // leader - we really should retry the request...
+ } else {
+ // System.out.println("clear exception");
+ rsp.setException(null);
+ }
}
}
}
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=1215243&r1=1215242&r2=1215243&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 Fri Dec 16 18:43:59 2011
@@ -215,7 +215,7 @@ public class DistributedUpdateProcessor
if (isLeader) {
params.set(SEEN_LEADER, true);
}
- cmdDistrib.distribAdd(cmd, shards, params);
+ cmdDistrib.distribAdd(cmd, shards, params, forwardToLeader);
} else {
// nocommit: At a minimum, local updates must be protected by synchronization
// right now we count on versionAdd to do the local add
@@ -234,7 +234,7 @@ public class DistributedUpdateProcessor
}
if (shards != null) {
- cmdDistrib.finish(shards, params);
+ cmdDistrib.finish(shards, params, forwardToLeader);
}
// TODO: keep track of errors? needs to be done at a higher level though since
@@ -390,7 +390,7 @@ public class DistributedUpdateProcessor
if (isLeader) {
params.set(SEEN_LEADER, true);
}
- cmdDistrib.distribDelete(cmd, shards, params);
+ cmdDistrib.distribDelete(cmd, shards, params, forwardToLeader);
} else {
// super.processDelete(cmd);
}
@@ -407,7 +407,7 @@ public class DistributedUpdateProcessor
}
if (shards != null) {
- cmdDistrib.finish(shards, params);
+ cmdDistrib.finish(shards, params, forwardToLeader);
}
}
@@ -575,7 +575,7 @@ public class DistributedUpdateProcessor
if (shards != null) {
cmdDistrib.distribCommit(cmd, shards, params);
- cmdDistrib.finish(shards, params);
+ cmdDistrib.finish(shards, params, forwardToLeader);
}
}
}
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=1215243&r1=1215242&r2=1215243&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 Fri Dec 16 18:43:59 2011
@@ -17,6 +17,7 @@ package org.apache.solr.cloud;
* limitations under the License.
*/
+import java.net.BindException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@@ -186,7 +187,13 @@ public class ChaosMonkey {
if (!deadPool.isEmpty()) {
System.out.println("start jetty");
JettySolrRunner jetty = deadPool.remove(random.nextInt(deadPool.size()));
- jetty.start();
+ try {
+ jetty.start();
+ } catch (BindException e) {
+ jetty.stop();
+ sleep(2000);
+ jetty.start();
+ }
starts.incrementAndGet();
continue;
}
Modified: lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/ChaosMonkeyDistributedZkTest.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/ChaosMonkeyDistributedZkTest.java?rev=1215243&r1=1215242&r2=1215243&view=diff
==============================================================================
--- lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/ChaosMonkeyDistributedZkTest.java (original)
+++ lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/ChaosMonkeyDistributedZkTest.java Fri Dec 16 18:43:59 2011
@@ -58,7 +58,8 @@ public class ChaosMonkeyDistributedZkTes
indexThread.safeStop();
// try and wait for any replications and what not to finish...
- Thread.sleep(5000);
+ // TODO: I suppose we should poll zk here about state
+ Thread.sleep(8000);
commit();
Modified: lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/FullDistributedZkTest.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/FullDistributedZkTest.java?rev=1215243&r1=1215242&r2=1215243&view=diff
==============================================================================
--- lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/FullDistributedZkTest.java (original)
+++ lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/cloud/FullDistributedZkTest.java Fri Dec 16 18:43:59 2011
@@ -770,7 +770,7 @@ public class FullDistributedZkTest exten
long num = -1;
long lastNum = -1;
String failMessage = null;
- System.out.println("\n\ncheck const");
+ System.out.println("check const of " + shard);
for (SolrServer client : solrClients) {
try {
num = client.query(new SolrQuery("*:*")).getResults().getNumFound();
@@ -788,10 +788,18 @@ public class FullDistributedZkTest exten
if (failMessage != null) {
fail(failMessage);
}
+
+ }
+
+ protected void checkShardConsistency() throws SolrServerException {
+ 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();
- Set<String> theShards = shardToClient.keySet();
+ theShards = shardToClient.keySet();
int cnt = 0;
for (String s : theShards) {
int times = shardToClient.get(s).size();
@@ -809,13 +817,6 @@ public class FullDistributedZkTest exten
}
assertEquals(docs, cnt);
}
-
- protected void checkShardConsistency() throws SolrServerException {
- Set<String> theShards = shardToClient.keySet();
- for (String shard : theShards) {
- checkShardConsistency(shard);
- }
- }
private SolrServer getClient(String nodeName) {
for (CloudSolrServerClient client : clientToInfo.keySet()) {