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 2013/11/25 23:32:06 UTC
svn commit: r1545428 - in /lucene/dev/trunk/solr/core/src:
java/org/apache/solr/update/SolrCmdDistributor.java
test/org/apache/solr/update/MockStreamingSolrServers.java
test/org/apache/solr/update/SolrCmdDistributorTest.java
Author: markrmiller
Date: Mon Nov 25 22:32:05 2013
New Revision: 1545428
URL: http://svn.apache.org/r1545428
Log:
We need better testing for SolrCmdDistributor retry logic.
Added:
lucene/dev/trunk/solr/core/src/test/org/apache/solr/update/MockStreamingSolrServers.java (with props)
Modified:
lucene/dev/trunk/solr/core/src/java/org/apache/solr/update/SolrCmdDistributor.java
lucene/dev/trunk/solr/core/src/test/org/apache/solr/update/SolrCmdDistributorTest.java
Modified: lucene/dev/trunk/solr/core/src/java/org/apache/solr/update/SolrCmdDistributor.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/core/src/java/org/apache/solr/update/SolrCmdDistributor.java?rev=1545428&r1=1545427&r2=1545428&view=diff
==============================================================================
--- lucene/dev/trunk/solr/core/src/java/org/apache/solr/update/SolrCmdDistributor.java (original)
+++ lucene/dev/trunk/solr/core/src/java/org/apache/solr/update/SolrCmdDistributor.java Mon Nov 25 22:32:05 2013
@@ -43,6 +43,9 @@ public class SolrCmdDistributor {
private StreamingSolrServers servers;
+ private int retryPause = 500;
+ private int maxRetriesOnForward = MAX_RETRIES_ON_FORWARD;
+
private List<Error> allErrors = new ArrayList<Error>();
private List<Error> errors = new ArrayList<Error>();
@@ -54,6 +57,12 @@ public class SolrCmdDistributor {
servers = new StreamingSolrServers(updateShardHandler);
}
+ public SolrCmdDistributor(StreamingSolrServers servers, int maxRetriesOnForward, int retryPause) {
+ this.servers = servers;
+ this.maxRetriesOnForward = maxRetriesOnForward;
+ this.retryPause = retryPause;
+ }
+
public void finish() {
try {
servers.blockUntilFinished();
@@ -85,8 +94,7 @@ public class SolrCmdDistributor {
// this can happen in certain situations such as shutdown
if (isRetry) {
- if (rspCode == 404 || rspCode == 403 || rspCode == 503
- || rspCode == 500) {
+ if (rspCode == 404 || rspCode == 403 || rspCode == 503) {
doRetry = true;
}
@@ -98,14 +106,14 @@ public class SolrCmdDistributor {
doRetry = true;
}
}
- if (err.req.retries < MAX_RETRIES_ON_FORWARD && doRetry) {
+ if (err.req.retries < maxRetriesOnForward && doRetry) {
err.req.retries++;
SolrException.log(SolrCmdDistributor.log, "forwarding update to "
+ oldNodeUrl + " failed - retrying ... retries: "
+ err.req.retries);
try {
- Thread.sleep(500);
+ Thread.sleep(retryPause);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
log.warn(null, e);
@@ -252,7 +260,7 @@ public class SolrCmdDistributor {
public static class Error {
public Exception e;
- public int statusCode;
+ public int statusCode = -1;
public Req req;
}
Added: lucene/dev/trunk/solr/core/src/test/org/apache/solr/update/MockStreamingSolrServers.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/core/src/test/org/apache/solr/update/MockStreamingSolrServers.java?rev=1545428&view=auto
==============================================================================
--- lucene/dev/trunk/solr/core/src/test/org/apache/solr/update/MockStreamingSolrServers.java (added)
+++ lucene/dev/trunk/solr/core/src/test/org/apache/solr/update/MockStreamingSolrServers.java Mon Nov 25 22:32:05 2013
@@ -0,0 +1,86 @@
+package org.apache.solr.update;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import java.io.IOException;
+import java.net.ConnectException;
+
+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.common.util.NamedList;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MockStreamingSolrServers extends StreamingSolrServers {
+ public static Logger log = LoggerFactory
+ .getLogger(MockStreamingSolrServers.class);
+
+ public enum Exp {CONNECT_EXCEPTION};
+
+ private volatile Exp exp = null;
+
+ public MockStreamingSolrServers(UpdateShardHandler updateShardHandler) {
+ super(updateShardHandler);
+ }
+
+ @Override
+ public synchronized SolrServer getSolrServer(final SolrCmdDistributor.Req req) {
+ SolrServer server = super.getSolrServer(req);
+ return new MockSolrServer(server);
+ }
+
+ public void setExp(Exp exp) {
+ this.exp = exp;
+ }
+
+ private IOException exception() {
+ switch (exp) {
+ case CONNECT_EXCEPTION:
+ return new ConnectException();
+
+ default:
+ break;
+ }
+ return null;
+ }
+
+ class MockSolrServer extends SolrServer {
+
+ private SolrServer solrServer;
+
+ public MockSolrServer(SolrServer solrServer) {
+ this.solrServer = solrServer;
+ }
+
+ @Override
+ public NamedList<Object> request(SolrRequest request)
+ throws SolrServerException, IOException {
+ if (exp != null) {
+ throw exception();
+ }
+
+ return solrServer.request(request);
+ }
+
+
+ @Override
+ public void shutdown() {}
+
+ }
+}
Modified: lucene/dev/trunk/solr/core/src/test/org/apache/solr/update/SolrCmdDistributorTest.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/core/src/test/org/apache/solr/update/SolrCmdDistributorTest.java?rev=1545428&r1=1545427&r2=1545428&view=diff
==============================================================================
--- lucene/dev/trunk/solr/core/src/test/org/apache/solr/update/SolrCmdDistributorTest.java (original)
+++ lucene/dev/trunk/solr/core/src/test/org/apache/solr/update/SolrCmdDistributorTest.java Mon Nov 25 22:32:05 2013
@@ -30,6 +30,7 @@ import org.apache.lucene.index.LogDocMer
import org.apache.solr.BaseDistributedSearchTestCase;
import org.apache.solr.client.solrj.SolrQuery;
import org.apache.solr.client.solrj.SolrServer;
+import org.apache.solr.client.solrj.SolrServerException;
import org.apache.solr.client.solrj.embedded.JettySolrRunner;
import org.apache.solr.client.solrj.impl.HttpSolrServer;
import org.apache.solr.client.solrj.request.LukeRequest;
@@ -46,6 +47,7 @@ import org.apache.solr.core.SolrCore;
import org.apache.solr.core.SolrEventListener;
import org.apache.solr.search.SolrIndexSearcher;
import org.apache.solr.servlet.SolrDispatchFilter;
+import org.apache.solr.update.MockStreamingSolrServers.Exp;
import org.apache.solr.update.SolrCmdDistributor.Error;
import org.apache.solr.update.SolrCmdDistributor.Node;
import org.apache.solr.update.SolrCmdDistributor.RetryNode;
@@ -55,6 +57,9 @@ import org.junit.BeforeClass;
import org.xml.sax.SAXException;
public class SolrCmdDistributorTest extends BaseDistributedSearchTestCase {
+
+ private AtomicInteger id = new AtomicInteger();
+
@BeforeClass
public static void beforeClass() throws Exception {
@@ -139,7 +144,7 @@ public class SolrCmdDistributorTest exte
// add one doc to controlClient
AddUpdateCommand cmd = new AddUpdateCommand(null);
- cmd.solrDoc = sdoc("id", 1);
+ cmd.solrDoc = sdoc("id", id.incrementAndGet());
params = new ModifiableSolrParams();
cmdDistrib.distribAdd(cmd, nodes, params);
@@ -166,20 +171,21 @@ public class SolrCmdDistributorTest exte
// add another 2 docs to control and 3 to client
cmdDistrib = new SolrCmdDistributor(updateShardHandler);
- cmd.solrDoc = sdoc("id", 2);
+ cmd.solrDoc = sdoc("id", id.incrementAndGet());
params = new ModifiableSolrParams();
params.set(DistributedUpdateProcessor.COMMIT_END_POINT, true);
cmdDistrib.distribAdd(cmd, nodes, params);
+ int id2 = id.incrementAndGet();
AddUpdateCommand cmd2 = new AddUpdateCommand(null);
- cmd2.solrDoc = sdoc("id", 3);
+ cmd2.solrDoc = sdoc("id", id2);
params = new ModifiableSolrParams();
params.set(DistributedUpdateProcessor.COMMIT_END_POINT, true);
cmdDistrib.distribAdd(cmd2, nodes, params);
AddUpdateCommand cmd3 = new AddUpdateCommand(null);
- cmd3.solrDoc = sdoc("id", 4);
+ cmd3.solrDoc = sdoc("id", id.incrementAndGet());
params = new ModifiableSolrParams();
params.set(DistributedUpdateProcessor.COMMIT_END_POINT, true);
@@ -204,8 +210,7 @@ public class SolrCmdDistributorTest exte
// now delete doc 2 which is on both control and client1
DeleteUpdateCommand dcmd = new DeleteUpdateCommand(null);
- dcmd.id = "2";
-
+ dcmd.id = Integer.toString(id2);
cmdDistrib = new SolrCmdDistributor(updateShardHandler);
@@ -239,8 +244,6 @@ public class SolrCmdDistributorTest exte
//System.out.println(clients.get(0).request(new LukeRequest()));
}
- int id = 5;
-
cmdDistrib = new SolrCmdDistributor(updateShardHandler);
int cnt = atLeast(303);
@@ -257,7 +260,7 @@ public class SolrCmdDistributorTest exte
}
AddUpdateCommand c = new AddUpdateCommand(null);
- c.solrDoc = sdoc("id", id++);
+ c.solrDoc = sdoc("id", id.incrementAndGet());
if (nodes.size() > 0) {
params = new ModifiableSolrParams();
cmdDistrib.distribAdd(c, nodes, params);
@@ -312,15 +315,100 @@ public class SolrCmdDistributorTest exte
((NamedList<Object>) resp.get("index")).get("maxDoc"));
}
+
+ testMaxRetries();
+ testOneRetry();
+ testRetryNode();
+ }
+
+ private void testMaxRetries() throws IOException {
+ final MockStreamingSolrServers ss = new MockStreamingSolrServers(updateShardHandler);
+ SolrCmdDistributor cmdDistrib = new SolrCmdDistributor(ss, 5, 0);
+ ss.setExp(Exp.CONNECT_EXCEPTION);
+ ArrayList<Node> nodes = new ArrayList<Node>();
+ final HttpSolrServer solrclient1 = (HttpSolrServer) clients.get(0);
+
+ final AtomicInteger retries = new AtomicInteger();
+ ZkNodeProps nodeProps = new ZkNodeProps(ZkStateReader.BASE_URL_PROP, solrclient1.getBaseURL(), ZkStateReader.CORE_NAME_PROP, "");
+ RetryNode retryNode = new RetryNode(new ZkCoreNodeProps(nodeProps), null, "collection1", "shard1") {
+ @Override
+ public boolean checkRetry() {
+ retries.incrementAndGet();
+ return true;
+ }
+ };
+
+ nodes.add(retryNode);
+
+ AddUpdateCommand cmd = new AddUpdateCommand(null);
+ cmd.solrDoc = sdoc("id", id.incrementAndGet());
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ cmdDistrib.distribAdd(cmd, nodes, params);
+ cmdDistrib.finish();
+
+ assertEquals(6, retries.get());
+
+ assertEquals(1, cmdDistrib.getErrors().size());
+ }
+
+ private void testOneRetry() throws Exception {
+ final HttpSolrServer solrclient = (HttpSolrServer) clients.get(0);
+ long numFoundBefore = solrclient.query(new SolrQuery("*:*")).getResults()
+ .getNumFound();
+ final MockStreamingSolrServers ss = new MockStreamingSolrServers(updateShardHandler);
+ SolrCmdDistributor cmdDistrib = new SolrCmdDistributor(ss, 5, 0);
+ ss.setExp(Exp.CONNECT_EXCEPTION);
+ ArrayList<Node> nodes = new ArrayList<Node>();
+
+ ZkNodeProps nodeProps = new ZkNodeProps(ZkStateReader.BASE_URL_PROP, solrclient.getBaseURL(),
+ ZkStateReader.CORE_NAME_PROP, "");
+
+ final AtomicInteger retries = new AtomicInteger();
+ nodeProps = new ZkNodeProps(ZkStateReader.BASE_URL_PROP, solrclient.getBaseURL(), ZkStateReader.CORE_NAME_PROP, "");
+ RetryNode retryNode = new RetryNode(new ZkCoreNodeProps(nodeProps), null, "collection1", "shard1") {
+ @Override
+ public boolean checkRetry() {
+ ss.setExp(null);
+ retries.incrementAndGet();
+ return true;
+ }
+ };
+
+
+ nodes.add(retryNode);
+
+ AddUpdateCommand cmd = new AddUpdateCommand(null);
+ cmd.solrDoc = sdoc("id", id.incrementAndGet());
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ CommitUpdateCommand ccmd = new CommitUpdateCommand(null, false);
+ cmdDistrib.distribAdd(cmd, nodes, params);
+ cmdDistrib.distribCommit(ccmd, nodes, params);
+ cmdDistrib.finish();
+
+ assertEquals(1, retries.get());
+
+
+ long numFoundAfter = solrclient.query(new SolrQuery("*:*")).getResults()
+ .getNumFound();
+
+ // we will get java.net.SocketException: Network is unreachable, which we don't retry on
+ assertEquals(numFoundBefore + 1, numFoundAfter);
+ assertEquals(0, cmdDistrib.getErrors().size());
+ }
+
+
+ private void testRetryNode() throws SolrServerException, IOException {
// Test RetryNode
- cmdDistrib = new SolrCmdDistributor(updateShardHandler);
+ SolrCmdDistributor cmdDistrib = new SolrCmdDistributor(updateShardHandler);
final HttpSolrServer solrclient = (HttpSolrServer) clients.get(0);
long numFoundBefore = solrclient.query(new SolrQuery("*:*")).getResults()
.getNumFound();
- nodes = new ArrayList<Node>();
+ ArrayList<Node> nodes = new ArrayList<Node>();
- nodeProps = new ZkNodeProps(ZkStateReader.BASE_URL_PROP, "[ff01::114]:33332" + context, ZkStateReader.CORE_NAME_PROP, "");
+ ZkNodeProps nodeProps = new ZkNodeProps(ZkStateReader.BASE_URL_PROP, "[ff01::114]:33332" + context, ZkStateReader.CORE_NAME_PROP, "");
RetryNode retryNode = new RetryNode(new ZkCoreNodeProps(nodeProps), null, "collection1", "shard1") {
@Override
public boolean checkRetry() {
@@ -336,13 +424,13 @@ public class SolrCmdDistributorTest exte
nodes.add(retryNode);
- cmd = new AddUpdateCommand(null);
- cmd.solrDoc = sdoc("id", 1111111);
- params = new ModifiableSolrParams();
+ AddUpdateCommand cmd = new AddUpdateCommand(null);
+ cmd.solrDoc = sdoc("id", id.incrementAndGet());
+ ModifiableSolrParams params = new ModifiableSolrParams();
cmdDistrib.distribAdd(cmd, nodes, params);
- ccmd = new CommitUpdateCommand(null, false);
+ CommitUpdateCommand ccmd = new CommitUpdateCommand(null, false);
params = new ModifiableSolrParams();
params.set(DistributedUpdateProcessor.COMMIT_END_POINT, true);
cmdDistrib.distribCommit(ccmd, nodes, params);
@@ -351,7 +439,10 @@ public class SolrCmdDistributorTest exte
long numFoundAfter = solrclient.query(new SolrQuery("*:*")).getResults()
.getNumFound();
+ // we will get java.net.SocketException: Network is unreachable and then retry
assertEquals(numFoundBefore + 1, numFoundAfter);
+
+ assertEquals(0, cmdDistrib.getErrors().size());
}
@Override