You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by er...@apache.org on 2017/10/31 23:49:14 UTC
lucene-solr:branch_7x: SOLR-11438: Solr should return rf when min_rf
is specified for deletes as well as adds
Repository: lucene-solr
Updated Branches:
refs/heads/branch_7x ff6bf6283 -> 6a2be9163
SOLR-11438: Solr should return rf when min_rf is specified for deletes as well as adds
(cherry picked from commit 148d81c)
Project: http://git-wip-us.apache.org/repos/asf/lucene-solr/repo
Commit: http://git-wip-us.apache.org/repos/asf/lucene-solr/commit/6a2be916
Tree: http://git-wip-us.apache.org/repos/asf/lucene-solr/tree/6a2be916
Diff: http://git-wip-us.apache.org/repos/asf/lucene-solr/diff/6a2be916
Branch: refs/heads/branch_7x
Commit: 6a2be9163bef25fb0c504c4f2aad2e60df4c5abe
Parents: ff6bf62
Author: Erick Erickson <er...@apache.org>
Authored: Tue Oct 31 16:21:58 2017 -0700
Committer: Erick Erickson <er...@apache.org>
Committed: Tue Oct 31 16:42:37 2017 -0700
----------------------------------------------------------------------
solr/CHANGES.txt | 2 +
.../apache/solr/update/SolrCmdDistributor.java | 119 +++--
.../processor/DistributedUpdateProcessor.java | 454 +++++++++++--------
.../solr/cloud/ReplicationFactorTest.java | 259 +++++++++--
4 files changed, 560 insertions(+), 274 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/6a2be916/solr/CHANGES.txt
----------------------------------------------------------------------
diff --git a/solr/CHANGES.txt b/solr/CHANGES.txt
index 8139246..a40c7d7 100644
--- a/solr/CHANGES.txt
+++ b/solr/CHANGES.txt
@@ -46,6 +46,8 @@ New Features
* SOLR-11520: Implement autoscaling suggestions for cores count violations (noble)
+* SOLR-11438: Solr should return rf when min_rf is specified for deletes as well as adds (Erick Erickson)
+
Bug Fixes
----------------------
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/6a2be916/solr/core/src/java/org/apache/solr/update/SolrCmdDistributor.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/update/SolrCmdDistributor.java b/solr/core/src/java/org/apache/solr/update/SolrCmdDistributor.java
index 5772b2e..1851f3d 100644
--- a/solr/core/src/java/org/apache/solr/update/SolrCmdDistributor.java
+++ b/solr/core/src/java/org/apache/solr/update/SolrCmdDistributor.java
@@ -16,6 +16,7 @@
*/
package org.apache.solr.update;
+
import org.apache.http.HttpResponse;
import org.apache.solr.client.solrj.SolrClient;
import org.apache.solr.client.solrj.SolrServerException;
@@ -32,7 +33,8 @@ import org.apache.solr.common.params.ModifiableSolrParams;
import org.apache.solr.common.util.NamedList;
import org.apache.solr.core.Diagnostics;
import org.apache.solr.update.processor.DistributedUpdateProcessor;
-import org.apache.solr.update.processor.DistributedUpdateProcessor.RequestReplicationTracker;
+import org.apache.solr.update.processor.DistributedUpdateProcessor.RollupRequestReplicationTracker;
+import org.apache.solr.update.processor.DistributedUpdateProcessor.LeaderRequestReplicationTracker;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -180,10 +182,12 @@ public class SolrCmdDistributor implements Closeable {
}
public void distribDelete(DeleteUpdateCommand cmd, List<Node> nodes, ModifiableSolrParams params) throws IOException {
- distribDelete(cmd, nodes, params, false);
+ distribDelete(cmd, nodes, params, false, null, null);
}
-
- public void distribDelete(DeleteUpdateCommand cmd, List<Node> nodes, ModifiableSolrParams params, boolean sync) throws IOException {
+
+ public void distribDelete(DeleteUpdateCommand cmd, List<Node> nodes, ModifiableSolrParams params, boolean sync,
+ RollupRequestReplicationTracker rollupTracker,
+ LeaderRequestReplicationTracker leaderTracker) throws IOException {
for (Node node : nodes) {
UpdateRequest uReq = new UpdateRequest();
@@ -194,20 +198,22 @@ public class SolrCmdDistributor implements Closeable {
} else {
uReq.deleteByQuery(cmd.query);
}
-
- submit(new Req(cmd, node, uReq, sync), false);
+
+ submit(new Req(cmd, node, uReq, sync, rollupTracker, leaderTracker), false);
}
}
public void distribAdd(AddUpdateCommand cmd, List<Node> nodes, ModifiableSolrParams params) throws IOException {
- distribAdd(cmd, nodes, params, false, null);
+ distribAdd(cmd, nodes, params, false, null, null);
}
public void distribAdd(AddUpdateCommand cmd, List<Node> nodes, ModifiableSolrParams params, boolean synchronous) throws IOException {
- distribAdd(cmd, nodes, params, synchronous, null);
+ distribAdd(cmd, nodes, params, synchronous, null, null);
}
-
- public void distribAdd(AddUpdateCommand cmd, List<Node> nodes, ModifiableSolrParams params, boolean synchronous, RequestReplicationTracker rrt) throws IOException {
+
+ public void distribAdd(AddUpdateCommand cmd, List<Node> nodes, ModifiableSolrParams params, boolean synchronous,
+ RollupRequestReplicationTracker rollupTracker,
+ LeaderRequestReplicationTracker leaderTracker) throws IOException {
for (Node node : nodes) {
UpdateRequest uReq = new UpdateRequest();
if (cmd.isLastDocInBatch)
@@ -217,7 +223,7 @@ public class SolrCmdDistributor implements Closeable {
if (cmd.isInPlaceUpdate()) {
params.set(DistributedUpdateProcessor.DISTRIB_INPLACE_PREVVERSION, String.valueOf(cmd.prevVersion));
}
- submit(new Req(cmd, node, uReq, synchronous, rrt), false);
+ submit(new Req(cmd, node, uReq, synchronous, rollupTracker, leaderTracker), false);
}
}
@@ -320,18 +326,22 @@ public class SolrCmdDistributor implements Closeable {
public int retries;
public boolean synchronous;
public UpdateCommand cmd;
- public RequestReplicationTracker rfTracker;
+ final private RollupRequestReplicationTracker rollupTracker;
+ final private LeaderRequestReplicationTracker leaderTracker;
public Req(UpdateCommand cmd, Node node, UpdateRequest uReq, boolean synchronous) {
- this(cmd, node, uReq, synchronous, null);
+ this(cmd, node, uReq, synchronous, null, null);
}
-
- public Req(UpdateCommand cmd, Node node, UpdateRequest uReq, boolean synchronous, RequestReplicationTracker rfTracker) {
+
+ public Req(UpdateCommand cmd, Node node, UpdateRequest uReq, boolean synchronous,
+ RollupRequestReplicationTracker rollupTracker,
+ LeaderRequestReplicationTracker leaderTracker) {
this.node = node;
this.uReq = uReq;
this.synchronous = synchronous;
this.cmd = cmd;
- this.rfTracker = rfTracker;
+ this.rollupTracker = rollupTracker;
+ this.leaderTracker = leaderTracker;
}
public String toString() {
@@ -340,40 +350,63 @@ public class SolrCmdDistributor implements Closeable {
sb.append("; node=").append(String.valueOf(node));
return sb.toString();
}
-
- public void trackRequestResult(HttpResponse resp, boolean success) {
- if (rfTracker != null) {
- Integer rf = null;
- if (resp != null) {
- // need to parse out the rf from requests that were forwards to another leader
- InputStream inputStream = null;
- try {
- inputStream = resp.getEntity().getContent();
- BinaryResponseParser brp = new BinaryResponseParser();
- NamedList<Object> nl= brp.processResponse(inputStream, null);
- Object hdr = nl.get("responseHeader");
- if (hdr != null && hdr instanceof NamedList) {
- NamedList<Object> hdrList = (NamedList<Object>)hdr;
- Object rfObj = hdrList.get(UpdateRequest.REPFACT);
- if (rfObj != null && rfObj instanceof Integer) {
- rf = (Integer)rfObj;
- }
+
+ // Called whenever we get results back from a sub-request.
+ // The only ambiguity is if I have _both_ a rollup tracker and a leader tracker. In that case we need to handle
+ // both requests returning from leaders of other shards _and_ from my followers. This happens if a leader happens
+ // to be the aggregator too.
+ //
+ // This isn't really a problem because only responses _from_ some leader will have the "rf" parameter, in which case
+ // we need to add the data to the rollup tracker.
+ //
+ // In the case of a leaderTracker and rollupTracker both being present, then we need to take care when assembling
+ // the final response to check both the rollup and leader trackers on the aggrator node.
+ public void trackRequestResult(HttpResponse resp, boolean success) {
+
+ // Returing Integer.MAX_VALUE here means there was no "rf" on the response, therefore we just need to increment
+ // our achieved rf if we are a leader, i.e. have a leaderTracker.
+ int rfFromResp = getRfFromResponse(resp);
+
+ if (leaderTracker != null && rfFromResp == Integer.MAX_VALUE) {
+ leaderTracker.trackRequestResult(node, success);
+ }
+
+ if (rollupTracker != null) {
+ rollupTracker.testAndSetAchievedRf(rfFromResp);
+ }
+ }
+
+ private int getRfFromResponse(HttpResponse resp) {
+ if (resp != null) {
+
+ InputStream inputStream = null;
+
+ try {
+ inputStream = resp.getEntity().getContent();
+ BinaryResponseParser brp = new BinaryResponseParser();
+ NamedList<Object> nl = brp.processResponse(inputStream, null);
+ Object hdr = nl.get("responseHeader");
+ if (hdr != null && hdr instanceof NamedList) {
+ NamedList<Object> hdrList = (NamedList<Object>) hdr;
+ Object rfObj = hdrList.get(UpdateRequest.REPFACT);
+ if (rfObj != null && rfObj instanceof Integer) {
+ return (Integer) rfObj;
}
- } catch (Exception e) {
- log.warn("Failed to parse response from "+node+" during replication factor accounting due to: "+e);
- } finally {
- if (inputStream != null) {
- try {
- inputStream.close();
- } catch (Exception ignore){}
+ }
+ } catch (Exception e) {
+ log.warn("Failed to parse response from " + node + " during replication factor accounting due to: " + e);
+ } finally {
+ if (inputStream != null) {
+ try {
+ inputStream.close();
+ } catch (Exception ignore) {
}
}
}
- rfTracker.trackRequestResult(node, success, rf);
}
+ return Integer.MAX_VALUE;
}
}
-
public static Diagnostics.Callable testing_errorHook; // called on error when forwarding request. Currently data=[this, Request]
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/6a2be916/solr/core/src/java/org/apache/solr/update/processor/DistributedUpdateProcessor.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/update/processor/DistributedUpdateProcessor.java b/solr/core/src/java/org/apache/solr/update/processor/DistributedUpdateProcessor.java
index ca5fcaa..5f4fcdd 100644
--- a/solr/core/src/java/org/apache/solr/update/processor/DistributedUpdateProcessor.java
+++ b/solr/core/src/java/org/apache/solr/update/processor/DistributedUpdateProcessor.java
@@ -23,15 +23,12 @@ import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.EnumSet;
-import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
-import java.util.Objects;
import java.util.Set;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.lucene.util.BytesRef;
@@ -135,103 +132,7 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
}
}
}
-
- /**
- * Keeps track of the replication factor achieved for a distributed update request
- * originated in this distributed update processor.
- */
- public static class RequestReplicationTracker {
- final int minRf;
- // if a leader is driving the update request, then this will be non-null
- // however a replica may also be driving the update request (forwards to leaders)
- // in which case we leave this as null so we only count the rf back from the leaders
- final String onLeaderShardId;
- // track number of nodes we sent requests to and how many resulted in errors
- // there may be multiple requests per node when processing a batch
- final Map<String,AtomicInteger> nodeErrorTracker;
- // if not using DirectUpdates, a leader may end up forwarding to other
- // leaders, so we need to keep the achieved rf for each of those too
- final Map<String,Integer> otherLeaderRf;
-
- private RequestReplicationTracker(String shardId, int minRf) {
- this.minRf = minRf;
- this.onLeaderShardId = shardId;
- this.nodeErrorTracker = new HashMap<>(5);
- this.otherLeaderRf = new HashMap<>();
- }
-
- // gives the replication factor that was achieved for this request
- public int getAchievedRf() {
- // look across all shards to find the minimum achieved replication
- // factor ... unless the client is using direct updates from CloudSolrServer
- // there may be multiple shards at play here
- int achievedRf = 1;
- if (onLeaderShardId != null) {
- synchronized (nodeErrorTracker) {
- for (AtomicInteger nodeErrors : nodeErrorTracker.values()) {
- if (nodeErrors.get() == 0)
- ++achievedRf;
- }
- }
- } else {
- // the node driving this updateRequest is not a leader and so
- // it only forwards to other leaders, so its local result doesn't count
- achievedRf = Integer.MAX_VALUE;
- }
-
- // min achieved may come from a request to another leader
- synchronized (otherLeaderRf) {
- for (Integer otherRf : otherLeaderRf.values()) {
- if (otherRf < achievedRf)
- achievedRf = otherRf;
- }
- }
-
- return (achievedRf == Integer.MAX_VALUE) ? 1 : achievedRf;
- }
-
- public void trackRequestResult(Node node, boolean success, Integer rf) {
- String shardId = node.getShardId();
- if (log.isDebugEnabled())
- log.debug("trackRequestResult("+node+"): success? "+success+" rf="+rf+
- ", shardId="+shardId+" onLeaderShardId="+onLeaderShardId);
-
- if (!Objects.equals(onLeaderShardId, shardId)) {
- // result from another leader that we forwarded to
- synchronized (otherLeaderRf) {
- otherLeaderRf.put(shardId, rf != null ? rf : new Integer(1));
- }
- return;
- }
-
- if (onLeaderShardId != null) {
- // track result for this leader
- String nodeUrl = node.getUrl();
- AtomicInteger nodeErrors = null;
- // potentially many results flooding into this method from multiple nodes concurrently
- synchronized (nodeErrorTracker) {
- nodeErrors = nodeErrorTracker.get(nodeUrl);
- if (nodeErrors == null) {
- nodeErrors = new AtomicInteger(0);
- nodeErrorTracker.put(nodeUrl, nodeErrors);
- }
- }
-
- if (!success)
- nodeErrors.incrementAndGet();
- }
- }
-
- public String toString() {
- StringBuilder sb = new StringBuilder("RequestReplicationTracker");
- sb.append(": onLeaderShardId=").append(String.valueOf(onLeaderShardId));
- sb.append(", minRf=").append(minRf);
- sb.append(", achievedRf=").append(getAchievedRf());
- return sb.toString();
- }
- }
-
public static final String COMMIT_END_POINT = "commit_end_point";
public static final String LOG_REPLAY = "log_replay";
@@ -273,7 +174,8 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
private UpdateCommand updateCommand; // the current command this processor is working on.
//used for keeping track of replicas that have processed an add/update from the leader
- private RequestReplicationTracker replicationTracker = null;
+ private RollupRequestReplicationTracker rollupReplicationTracker = null;
+ private LeaderRequestReplicationTracker leaderReplicationTracker = null;
// should we clone the document before sending it to replicas?
// this is set to true in the constructor if the next processors in the chain
@@ -454,7 +356,7 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
return nodes;
} else {
- // I need to forward onto the leader...
+ // I need to forward on to the leader...
forwardToLeader = true;
return Collections.singletonList(
new RetryNode(new ZkCoreNodeProps(leaderReplica), zkController.getZkStateReader(), collection, shardId));
@@ -681,9 +583,9 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
@Override
public void processAdd(AddUpdateCommand cmd) throws IOException {
-
+
assert TestInjection.injectFailUpdateRequests();
-
+
updateCommand = cmd;
if (zkEnabled) {
@@ -692,36 +594,10 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
} else {
isLeader = getNonZkLeaderAssumption(req);
}
-
- // check if client has requested minimum replication factor information
- int minRf = -1; // disabled by default
- if (replicationTracker != null) {
- minRf = replicationTracker.minRf; // for subsequent requests in the same batch
- } else {
- SolrParams rp = cmd.getReq().getParams();
- String distribUpdate = rp.get(DISTRIB_UPDATE_PARAM);
- // somewhat tricky logic here: we only activate the replication tracker if we're on
- // a leader or this is the top-level request processor
- if (distribUpdate == null || distribUpdate.equals(DistribPhase.TOLEADER.toString())) {
- String minRepFact = rp.get(UpdateRequest.MIN_REPFACT);
- if (minRepFact != null) {
- try {
- minRf = Integer.parseInt(minRepFact);
- } catch (NumberFormatException nfe) {
- minRf = -1;
- }
-
- if (minRf <= 0)
- throw new SolrException(ErrorCode.BAD_REQUEST, "Invalid value "+minRepFact+" for "+UpdateRequest.MIN_REPFACT+
- "; must be >0 and less than or equal to the collection replication factor.");
- }
-
- if (minRf > 1) {
- String myShardId = forwardToLeader ? null : cloudDesc.getShardId();
- replicationTracker = new RequestReplicationTracker(myShardId, minRf);
- }
- }
- }
+
+ // check if client has requested minimum replication factor information. will set replicationTracker to null if
+ // we aren't the leader or subShardLeader
+ checkReplicationTracker(cmd);
// If we were sent a previous version, set this to the AddUpdateCommand (if not already set)
if (!cmd.isInPlaceUpdate()) {
@@ -730,9 +606,9 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
// TODO: if minRf > 1 and we know the leader is the only active replica, we could fail
// the request right here but for now I think it is better to just return the status
// to the client that the minRf wasn't reached and let them handle it
-
+
boolean dropCmd = false;
- if (!forwardToLeader) {
+ if (!forwardToLeader) {
dropCmd = versionAdd(cmd);
}
@@ -764,19 +640,19 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
cmdDistrib.distribAdd(cmd, nodesByRoutingRules, params, true);
}
}
-
- ModifiableSolrParams params = null;
+
if (nodes != null) {
- params = new ModifiableSolrParams(filterParams(req.getParams()));
+ ModifiableSolrParams params = new ModifiableSolrParams(filterParams(req.getParams()));
params.set(DISTRIB_UPDATE_PARAM,
- (isLeader || isSubShardLeader ?
- DistribPhase.FROMLEADER.toString() :
- DistribPhase.TOLEADER.toString()));
+ (isLeader || isSubShardLeader ?
+ DistribPhase.FROMLEADER.toString() :
+ DistribPhase.TOLEADER.toString()));
params.set(DISTRIB_FROM, ZkCoreNodeProps.getCoreUrl(
zkController.getBaseUrl(), req.getCore().getName()));
-
- if (replicationTracker != null && minRf > 1)
- params.set(UpdateRequest.MIN_REPFACT, String.valueOf(minRf));
+
+ if (req.getParams().get(UpdateRequest.MIN_REPFACT) != null) {
+ params.set(UpdateRequest.MIN_REPFACT, req.getParams().get(UpdateRequest.MIN_REPFACT));
+ }
if (cmd.isInPlaceUpdate()) {
params.set(DISTRIB_INPLACE_PREVVERSION, String.valueOf(cmd.prevVersion));
@@ -787,12 +663,12 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
// and the current in-place update (that depends on the previous update), if reordered
// in the stream, can result in the current update being bottled up behind the previous
// update in the stream and can lead to degraded performance.
- cmdDistrib.distribAdd(cmd, nodes, params, true, replicationTracker);
+ cmdDistrib.distribAdd(cmd, nodes, params, true, rollupReplicationTracker, leaderReplicationTracker);
} else {
- cmdDistrib.distribAdd(cmd, nodes, params, false, replicationTracker);
+ cmdDistrib.distribAdd(cmd, nodes, params, false, rollupReplicationTracker, leaderReplicationTracker);
}
}
-
+
// TODO: what to do when no idField?
if (returnVersions && rsp != null && idField != null) {
if (addsResponse == null) {
@@ -803,13 +679,56 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
idField.getType().indexedToReadable(cmd.getIndexedId(), scratch);
addsResponse.add(scratch.toString(), cmd.getVersion());
}
-
+
// TODO: keep track of errors? needs to be done at a higher level though since
// an id may fail before it gets to this processor.
// Given that, it may also make sense to move the version reporting out of this
// processor too.
+
}
-
+
+ // helper method, processAdd was getting a bit large.
+ // Sets replicationTracker = null if we aren't the leader or don't care abeout minRf
+ // We have three possibilities here:
+ //
+ // 1> there is no min_rf specified: Just return
+ // 2> we are a leader: Allocate a LeaderTracker and, if we're getting the original request, a RollupTracker
+ // 3> we're a follower: allocat a RollupTracker
+ //
+ private void checkReplicationTracker(UpdateCommand cmd) {
+ String repFact = req.getParams().get(UpdateRequest.MIN_REPFACT);
+
+ if (zkEnabled == false || repFact == null) {
+ rollupReplicationTracker = null; // never need one of these in stand-alone
+ leaderReplicationTracker = null;
+ return;
+ }
+
+ int requestedReplicationFactor;
+
+ try {
+ requestedReplicationFactor = Integer.parseInt(repFact);
+ } catch (NumberFormatException nfe) {
+ throw new SolrException(ErrorCode.SERVER_ERROR, "MinRF must be an integer, was " + repFact);
+ }
+
+ SolrParams rp = cmd.getReq().getParams();
+ String distribUpdate = rp.get(DISTRIB_UPDATE_PARAM);
+ // Ok,we're receiving the original request, we need a rollup tracker, but only one so we accumulate over the
+ // course of a batch.
+ if ((distribUpdate == null || DistribPhase.NONE.toString().equals(distribUpdate)) &&
+ rollupReplicationTracker == null) {
+ rollupReplicationTracker = new RollupRequestReplicationTracker(repFact);
+ }
+ // If we're a leader, we need a leader replication tracker, so let's do that. If there are multiple docs in
+ // a batch we need to use the _same_ leader replication tracker.
+ if (isLeader && leaderReplicationTracker == null) {
+ leaderReplicationTracker = new LeaderRequestReplicationTracker(
+ req.getCore().getCoreDescriptor().getCloudDescriptor().getShardId(), requestedReplicationFactor);
+ }
+ }
+
+
@Override
protected void doClose() {
if (cmdDistrib != null) {
@@ -820,8 +739,8 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
// TODO: optionally fail if n replicas are not reached...
private void doFinish() {
// TODO: if not a forward and replication req is not specified, we could
- // send in a background thread
-
+ // send in a background thread
+
cmdDistrib.finish();
List<Error> errors = cmdDistrib.getErrors();
// TODO - we may need to tell about more than one error...
@@ -849,9 +768,8 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
// Since it is not a forward request, for each fail, try to tell them to
// recover - the doc was already added locally, so it should have been
// legit
-
- DistribPhase phase =
- DistribPhase.parseParam(error.req.uReq.getParams().get(DISTRIB_UPDATE_PARAM));
+
+ DistribPhase phase = DistribPhase.parseParam(error.req.uReq.getParams().get(DISTRIB_UPDATE_PARAM));
if (phase != DistribPhase.FROMLEADER)
continue; // don't have non-leaders try to recovery other nodes
@@ -911,7 +829,8 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
}
// If the client specified minRf and we didn't achieve the minRf, don't send recovery and let client retry
- if (replicationTracker != null && replicationTracker.getAchievedRf() < replicationTracker.minRf) {
+ if (leaderReplicationTracker != null &&
+ leaderReplicationTracker.getAchievedRf() < leaderReplicationTracker.getRequestedRf()) {
continue;
}
@@ -949,14 +868,33 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
}
}
}
+ // in either case, we need to attach the achieved and min rf to the response.
+ if (leaderReplicationTracker != null || rollupReplicationTracker != null) {
+ int achievedRf = Integer.MAX_VALUE;
+ int requestedRf = Integer.MAX_VALUE;
- if (replicationTracker != null) {
- rsp.getResponseHeader().add(UpdateRequest.REPFACT, replicationTracker.getAchievedRf());
- rsp.getResponseHeader().add(UpdateRequest.MIN_REPFACT, replicationTracker.minRf);
- replicationTracker = null;
- }
+ if (leaderReplicationTracker != null) {
-
+ achievedRf = leaderReplicationTracker.getAchievedRf();
+ requestedRf = leaderReplicationTracker.getRequestedRf();
+
+ // Transfer this to the rollup tracker if it exists
+ if (rollupReplicationTracker != null) {
+ rollupReplicationTracker.testAndSetAchievedRf(achievedRf);
+ }
+ }
+
+ // Rollup tracker has accumulated stats.
+ if (rollupReplicationTracker != null) {
+ achievedRf = rollupReplicationTracker.getAchievedRf();
+ requestedRf = rollupReplicationTracker.getRequestedRf();
+ }
+ rsp.getResponseHeader().add(UpdateRequest.MIN_REPFACT, requestedRf);
+ rsp.getResponseHeader().add(UpdateRequest.REPFACT, achievedRf);
+ rollupReplicationTracker = null;
+ leaderReplicationTracker = null;
+
+ }
if (0 < errorsForClient.size()) {
throw new DistributedUpdatesAsyncException(errorsForClient);
}
@@ -1048,8 +986,6 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
// TODO: if versions aren't stored, do we need to set on the cmd anyway for some reason?
// there may be other reasons in the future for a version on the commands
- boolean checkDeleteByQueries = false;
-
if (versionsStored) {
long bucketVersion = bucket.highest;
@@ -1162,9 +1098,6 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
log.debug("Dropping add update due to version {}", idBytes.utf8ToString());
return true;
}
-
- // also need to re-apply newer deleteByQuery commands
- checkDeleteByQueries = true;
}
}
if (replicaType == Replica.Type.TLOG && (cmd.getFlags() & UpdateCommand.REPLAY) == 0) {
@@ -1376,21 +1309,32 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
if (!cmd.isDeleteById()) {
doDeleteByQuery(cmd);
- return;
+ } else {
+ doDeleteById(cmd);
}
+ }
+ // Implementing min_rf here was a bit tricky. When a request comes in for a delete by id to a replica that does _not_
+ // have any documents specified by those IDs, the request is not forwarded to any other replicas on that shard. Thus
+ // we have to spoof the replicationTracker and set the achieved rf to the number of active replicas.
+ //
+ private void doDeleteById(DeleteUpdateCommand cmd) throws IOException {
if (zkEnabled) {
zkCheck();
nodes = setupRequest(cmd.getId(), null, cmd.getRoute());
} else {
isLeader = getNonZkLeaderAssumption(req);
}
-
+
+ // check if client has requested minimum replication factor information. will set replicationTracker to null if
+ // we aren't the leader or subShardLeader
+ checkReplicationTracker(cmd);
+
boolean dropCmd = false;
if (!forwardToLeader) {
dropCmd = versionDelete(cmd);
}
-
+
if (dropCmd) {
// TODO: do we need to add anything to the response?
return;
@@ -1406,7 +1350,7 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
params.set(DISTRIB_FROM, ZkCoreNodeProps.getCoreUrl(
zkController.getBaseUrl(), req.getCore().getName()));
params.set(DISTRIB_FROM_PARENT, cloudDesc.getShardId());
- cmdDistrib.distribDelete(cmd, subShardLeaders, params, true);
+ cmdDistrib.distribDelete(cmd, subShardLeaders, params, true, null, null);
}
final List<Node> nodesByRoutingRules = getNodesByRoutingRules(zkController.getClusterState(), coll, cmd.getId(), null);
@@ -1417,22 +1361,22 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
zkController.getBaseUrl(), req.getCore().getName()));
params.set(DISTRIB_FROM_COLLECTION, cloudDesc.getCollectionName());
params.set(DISTRIB_FROM_SHARD, cloudDesc.getShardId());
- cmdDistrib.distribDelete(cmd, nodesByRoutingRules, params, true);
+ cmdDistrib.distribDelete(cmd, nodesByRoutingRules, params, true, null, null);
}
}
-
- ModifiableSolrParams params = null;
if (nodes != null) {
-
- params = new ModifiableSolrParams(filterParams(req.getParams()));
+ ModifiableSolrParams params = new ModifiableSolrParams(filterParams(req.getParams()));
params.set(DISTRIB_UPDATE_PARAM,
(isLeader || isSubShardLeader ? DistribPhase.FROMLEADER.toString()
: DistribPhase.TOLEADER.toString()));
params.set(DISTRIB_FROM, ZkCoreNodeProps.getCoreUrl(
zkController.getBaseUrl(), req.getCore().getName()));
- cmdDistrib.distribDelete(cmd, nodes, params);
+ if (req.getParams().get(UpdateRequest.MIN_REPFACT) != null) {
+ params.add(UpdateRequest.MIN_REPFACT, req.getParams().get(UpdateRequest.MIN_REPFACT));
+ }
+ cmdDistrib.distribDelete(cmd, nodes, params, false, rollupReplicationTracker, leaderReplicationTracker);
}
// cmd.getIndexId == null when delete by query
@@ -1472,6 +1416,7 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
}
public void doDeleteByQuery(DeleteUpdateCommand cmd) throws IOException {
+
// even in non zk mode, tests simulate updates from a leader
if(!zkEnabled) {
isLeader = getNonZkLeaderAssumption(req);
@@ -1488,13 +1433,15 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
// - log + execute the local DBQ
// FROM: we are a replica receiving a DBQ from our leader
// - log + execute the local DBQ
- DistribPhase phase =
- DistribPhase.parseParam(req.getParams().get(DISTRIB_UPDATE_PARAM));
+ DistribPhase phase = DistribPhase.parseParam(req.getParams().get(DISTRIB_UPDATE_PARAM));
- DocCollection coll = zkEnabled
+ DocCollection coll = zkEnabled
? zkController.getClusterState().getCollection(collection) : null;
if (zkEnabled && DistribPhase.NONE == phase) {
+ if (req.getParams().get(UpdateRequest.MIN_REPFACT) != null && rollupReplicationTracker == null) {
+ rollupReplicationTracker = new RollupRequestReplicationTracker(req.getParams().get(UpdateRequest.MIN_REPFACT));
+ }
boolean leaderForAnyShard = false; // start off by assuming we are not a leader for any shard
ModifiableSolrParams outParams = new ModifiableSolrParams(filterParams(req.getParams()));
@@ -1534,7 +1481,15 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
}
outParams.remove("commit"); // this will be distributed from the local commit
- cmdDistrib.distribDelete(cmd, leaders, outParams);
+
+
+ if (req.getParams().get(UpdateRequest.MIN_REPFACT) != null) {
+ // If we've determined that there are no docs on this shard that need to be deleted, then we don't send
+ // sub-requests to any other replicas for this shard. In this case, min_rf is meaningless for this shard
+ // so flag that in replicationTracker
+ outParams.add(UpdateRequest.MIN_REPFACT, req.getParams().get(UpdateRequest.MIN_REPFACT));
+ }
+ cmdDistrib.distribDelete(cmd, leaders, outParams, false, rollupReplicationTracker, null);
if (!leaderForAnyShard) {
return;
@@ -1544,6 +1499,10 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
phase = DistribPhase.TOLEADER;
}
+ // check if client has requested minimum replication factor information. will set replicationTracker to null if
+ // we aren't the leader or subShardLeader
+ checkReplicationTracker(cmd);
+
List<Node> replicas = null;
if (zkEnabled && DistribPhase.TOLEADER == phase) {
@@ -1564,9 +1523,7 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
boolean isReplayOrPeersync = (cmd.getFlags() & (UpdateCommand.REPLAY | UpdateCommand.PEER_SYNC)) != 0;
boolean leaderLogic = isLeader && !isReplayOrPeersync;
-
versionDeleteByQuery(cmd);
-
if (zkEnabled) {
// forward to all replicas
ModifiableSolrParams params = new ModifiableSolrParams(filterParams(req.getParams()));
@@ -1591,7 +1548,7 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
for (ZkCoreNodeProps replicaProp : replicaProps) {
myReplicas.add(new StdNode(replicaProp, collection, myShardId));
}
- cmdDistrib.distribDelete(cmd, myReplicas, params);
+ cmdDistrib.distribDelete(cmd, myReplicas, params, false, rollupReplicationTracker, leaderReplicationTracker);
someReplicas = true;
}
}
@@ -1599,11 +1556,10 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
Thread.currentThread().interrupt();
throw new ZooKeeperException(ErrorCode.SERVER_ERROR, "", e);
}
-
if (leaderLogic) {
List<Node> subShardLeaders = getSubShardLeaders(coll, cloudDesc.getShardId(), null, null);
if (subShardLeaders != null) {
- cmdDistrib.distribDelete(cmd, subShardLeaders, params, true);
+ cmdDistrib.distribDelete(cmd, subShardLeaders, params, true, rollupReplicationTracker, leaderReplicationTracker);
}
final List<Node> nodesByRoutingRules = getNodesByRoutingRules(zkController.getClusterState(), coll, null, null);
if (nodesByRoutingRules != null && !nodesByRoutingRules.isEmpty()) {
@@ -1613,10 +1569,11 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
zkController.getBaseUrl(), req.getCore().getName()));
params.set(DISTRIB_FROM_COLLECTION, cloudDesc.getCollectionName());
params.set(DISTRIB_FROM_SHARD, cloudDesc.getShardId());
- cmdDistrib.distribDelete(cmd, nodesByRoutingRules, params, true);
+
+ cmdDistrib.distribDelete(cmd, nodesByRoutingRules, params, true, rollupReplicationTracker, leaderReplicationTracker);
}
if (replicas != null) {
- cmdDistrib.distribDelete(cmd, replicas, params);
+ cmdDistrib.distribDelete(cmd, replicas, params, false, rollupReplicationTracker, leaderReplicationTracker);
someReplicas = true;
}
}
@@ -1691,6 +1648,7 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
}
// internal helper method to tell if we are the leader for an add or deleteById update
+ // NOTE: not called by this class!
boolean isLeader(UpdateCommand cmd) {
updateCommand = cmd;
@@ -1938,9 +1896,7 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
if (next != null && nodes == null) next.finish();
}
-
-
private List<Node> getCollectionUrls(SolrQueryRequest req, String collection, EnumSet<Replica.Type> types) {
ClusterState clusterState = req.getCore()
.getCoreContainer().getZkController().getClusterState();
@@ -2048,4 +2004,124 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
}
}
}
+
+
+ // Keeps track of the replication factor achieved for a distributed update request
+ // originated in this distributed update processor. A RollupReplicationTracker is the only tracker that will
+ // persist across sub-requests.
+ //
+ // Note that the replica that receives the original request has the only RollupReplicationTracker that exists for the
+ // lifetime of the batch. The leader for each shard keeps track of it's own achieved replicaiton for its shard
+ // and attaches that to the response to the originating node (i.e. the one with the RollupReplicationTracker).
+ // Followers in general do not need a tracker of any sort with the sole exception of the RollupReplicationTracker
+ // allocated on the original node that recieves the top-level request.
+ //
+ // DeleteById is tricky. Since the docs are sent one at a time, there has to be some fancy dancing. In the
+ // deleteById case, here are the rules:
+ //
+ // If I'm leader, there are two possibilities:
+ // 1> I got the original request. This is the hard one. There are two sub-cases:
+ // a> Some document in the request is deleted from the shard I lead. In this case my computed replication
+ // factor counts.
+ // b> No document in the packet is deleted from my shard. In that case I have nothing to say about the
+ // achieved replication factor.
+ //
+ // 2> I'm a leader and I got the request from some other replica. In this case I can be certain of a couple of things:
+ // a> The document in the request will be deleted from my shard
+ // b> my replication factor counts.
+ //
+ // Even the DeleteById case follows the rules for whether a RollupReplicaitonTracker is allocated.
+ // This doesn't matter when it comes to delete-by-query since all leaders get the sub-request.
+
+
+ public static class RollupRequestReplicationTracker {
+
+ private int achievedRf = Integer.MAX_VALUE;
+ private final int requestedRf;
+
+ RollupRequestReplicationTracker(String minRepFact) {
+ try {
+ this.requestedRf = Integer.parseInt(minRepFact);
+ } catch (NumberFormatException nfe) {
+ throw new SolrException(ErrorCode.SERVER_ERROR, "MinRF must be an integer, was " + minRepFact);
+ }
+ }
+
+ public int getAchievedRf() {
+ return achievedRf;
+ }
+
+ public int getRequestedRf() {
+ return requestedRf;
+ }
+
+ // We want to report only the minimun _ever_ achieved...
+ public void testAndSetAchievedRf(int rf) {
+ this.achievedRf = Math.min(this.achievedRf, rf);
+ }
+
+ public String toString() {
+ StringBuilder sb = new StringBuilder("RollupRequestReplicationTracker")
+ .append(", requestedRf: ")
+ .append(requestedRf)
+ .append(" achievedRf: ")
+ .append(achievedRf);
+ return sb.toString();
+ }
+ }
+
+
+ // Allocate a LeaderRequestReplicatinTracker if (and only if) we're a leader. If the request comes in to the leader
+ // at first, allocate both one of these and a RollupRequestReplicationTracker.
+ //
+ // Since these are leader-only, all they really have to do is track the individual update request for this shard
+ // and return it to be added to the rollup tracker. Which is kind of simple since we get an onSuccess method in
+ // SolrCmdDistributor
+
+ public static class LeaderRequestReplicationTracker {
+ private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ // Since we only allocate one of these on the leader and, by definition, the leader has been found and is running,
+ // we have a replication factor of one by default.
+ private int achievedRf = 1;
+ private final int requestedRf;
+
+ private final String myShardId;
+
+ int getRequestedRf() {
+ return requestedRf;
+ }
+
+ LeaderRequestReplicationTracker(String shardId, int requestedRf) {
+ this.requestedRf = requestedRf;
+ this.myShardId = shardId;
+ }
+
+ // gives the replication factor that was achieved for this request
+ public int getAchievedRf() {
+ return achievedRf;
+ }
+
+ public void trackRequestResult(Node node, boolean success) {
+ if (log.isDebugEnabled()) {
+ log.debug("trackRequestResult(" + node + "): success? " + success +
+ ", shardId=" + myShardId);
+ }
+
+ if (success) {
+ ++achievedRf;
+ }
+ }
+
+ public String toString() {
+ StringBuilder sb = new StringBuilder("LeaderRequestReplicationTracker");
+ sb.append(", achievedRf=")
+ .append(getAchievedRf())
+ .append(" for shard ")
+ .append(myShardId)
+ .append(" requested replication factor: ")
+ .append(requestedRf);
+ return sb.toString();
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/6a2be916/solr/core/src/test/org/apache/solr/cloud/ReplicationFactorTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/ReplicationFactorTest.java b/solr/core/src/test/org/apache/solr/cloud/ReplicationFactorTest.java
index 8fa1607..1c483f5 100644
--- a/solr/core/src/test/org/apache/solr/cloud/ReplicationFactorTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/ReplicationFactorTest.java
@@ -17,20 +17,27 @@
package org.apache.solr.cloud;
import java.io.File;
+import java.io.IOException;
import java.lang.invoke.MethodHandles;
import java.util.ArrayList;
+import java.util.HashSet;
import java.util.List;
import java.util.Locale;
+import java.util.Set;
+import org.apache.commons.lang3.StringUtils;
import org.apache.lucene.util.LuceneTestCase;
import org.apache.lucene.util.LuceneTestCase.Slow;
import org.apache.solr.SolrTestCaseJ4.SuppressSSL;
+import org.apache.solr.client.solrj.SolrServerException;
import org.apache.solr.client.solrj.embedded.JettySolrRunner;
import org.apache.solr.client.solrj.impl.HttpSolrClient;
import org.apache.solr.client.solrj.request.CollectionAdminRequest;
import org.apache.solr.client.solrj.request.UpdateRequest;
import org.apache.solr.client.solrj.response.CollectionAdminResponse;
import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.cloud.DocCollection;
+import org.apache.solr.common.cloud.DocRouter;
import org.apache.solr.common.cloud.Replica;
import org.apache.solr.common.cloud.ZkCoreNodeProps;
import org.apache.solr.common.util.NamedList;
@@ -38,7 +45,6 @@ import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-
/**
* Tests a client application's ability to get replication factor
* information back from the cluster after an add or update.
@@ -74,7 +80,7 @@ public class ReplicationFactorTest extends AbstractFullDistribZkTestBase {
public void test() throws Exception {
log.info("replication factor test running");
waitForThingsToLevelOut(30000);
-
+
// test a 1x3 collection
log.info("Testing replication factor handling for repfacttest_c8n_1x3");
testRf3();
@@ -97,19 +103,9 @@ public class ReplicationFactorTest extends AbstractFullDistribZkTestBase {
String testCollectionName = "repfacttest_c8n_2x2";
String shardId = "shard1";
int minRf = 2;
-
- CollectionAdminResponse resp = createCollection(testCollectionName, "conf1", numShards, replicationFactor, maxShardsPerNode);
-
- if (resp.getResponse().get("failure") != null) {
- CollectionAdminRequest.deleteCollection(testCollectionName).process(cloudClient);
-
- resp = createCollection(testCollectionName, "conf1", numShards, replicationFactor, maxShardsPerNode);
-
- if (resp.getResponse().get("failure") != null) {
- fail("Could not create " + testCollectionName);
- }
- }
-
+
+ createCollectionWithRetry(testCollectionName, "conf1", numShards, replicationFactor, maxShardsPerNode);
+
cloudClient.setDefaultCollection(testCollectionName);
List<Replica> replicas =
@@ -123,39 +119,132 @@ public class ReplicationFactorTest extends AbstractFullDistribZkTestBase {
doc.addField("a_t", "hello" + i);
batch.add(doc);
}
-
+
// send directly to the leader using HttpSolrServer instead of CloudSolrServer (to test support for non-direct updates)
UpdateRequest up = new UpdateRequest();
up.setParam(UpdateRequest.MIN_REPFACT, String.valueOf(minRf));
up.add(batch);
Replica leader = cloudClient.getZkStateReader().getLeaderRetry(testCollectionName, shardId);
- sendNonDirectUpdateRequestReplicaWithRetry(leader, up, 2, testCollectionName);
- sendNonDirectUpdateRequestReplicaWithRetry(replicas.get(0), up, 2, testCollectionName);
-
+
+ sendNonDirectUpdateRequestReplicaWithRetry(leader, up, 2, testCollectionName);
+ sendNonDirectUpdateRequestReplicaWithRetry(replicas.get(0), up, 2, testCollectionName);
+
+ // Insure nothing is tricky about a delete where only one shard needs to delete anything.
+ sendNonDirectDeletesRequestReplicaWithRetry(leader, getSomeIds(1), 2,
+ getSomeIds(1), 2, testCollectionName);
+ sendNonDirectDeletesRequestReplicaWithRetry(replicas.get(0), getSomeIds(1), 2,
+ getSomeIds(1), 2, testCollectionName);
+
+ sendNonDirectDeletesRequestReplicaWithRetry(leader, getSomeIds(2), 2,
+ getSomeIds(2), 2, testCollectionName);
+ sendNonDirectDeletesRequestReplicaWithRetry(replicas.get(0), getSomeIds(2), 2,
+ getSomeIds(2), 2, testCollectionName);
+
// so now kill the replica of shard2 and verify the achieved rf is only 1
- List<Replica> shard2Replicas =
+ List<Replica> shard2Replicas =
ensureAllReplicasAreActive(testCollectionName, "shard2", numShards, replicationFactor, 30);
assertTrue("Expected active 1 replicas for "+testCollectionName, replicas.size() == 1);
-
+
getProxyForReplica(shard2Replicas.get(0)).close();
Thread.sleep(2000);
-
+
// shard1 will have rf=2 but shard2 will only have rf=1
- sendNonDirectUpdateRequestReplicaWithRetry(leader, up, 1, testCollectionName);
+ sendNonDirectUpdateRequestReplicaWithRetry(leader, up, 1, testCollectionName);
sendNonDirectUpdateRequestReplicaWithRetry(replicas.get(0), up, 1, testCollectionName);
-
+
+ // Whether the replication factor is 1 or 2 in the delete-by-id case depends on whether the doc IDs happen to fall
+ // on a single shard or not.
+
+ Set<Integer> byIDs;
+ byIDs = getSomeIds(2);
+ sendNonDirectDeletesRequestReplicaWithRetry(leader,
+ byIDs, calcByIdRf(byIDs, testCollectionName, "shard2"),
+ getSomeIds(2), 1, testCollectionName);
+ byIDs = getSomeIds(2);
+ sendNonDirectDeletesRequestReplicaWithRetry(replicas.get(0), byIDs,
+ calcByIdRf(byIDs, testCollectionName, "shard2"),
+ getSomeIds(2), 1, testCollectionName);
// heal the partition
getProxyForReplica(shard2Replicas.get(0)).reopen();
-
Thread.sleep(2000);
}
-
+
+ // When doing a delete by id, it's tricky, very tricky. If any document we're deleting by ID goes to shardWithOne,
+ // then the replication factor we return will be 1.
+ //
+ private int calcByIdRf(Set<Integer> byIDs, String testCollectionName, String shardWithOne) {
+ ZkController zkController = jettys.get(0).getCoreContainer().getZkController();
+ DocCollection coll = zkController.getClusterState().getCollection(testCollectionName);
+ int retval = 2;
+ for (int id : byIDs) {
+ DocRouter router = coll.getRouter();
+ if (shardWithOne.equals(router.getTargetSlice(Integer.toString(id), null, null, null, coll).getName())) {
+ retval = 1;
+ }
+ }
+ return retval;
+ }
+
+ int idFloor = random().nextInt(100) + 1000; // Get the delete tests to use disjoint documents although
+
+ // Randomize documents so we exercise requests landing on replicas that have (or don't) particular documents
+ // Yeah, this will go on forever if you ask for more than 100, but it suffices.
+ private Set<Integer> getSomeIds(int count) {
+ Set<Integer> ids = new HashSet<>();
+ while (ids.size() < count) {
+ ids.add(idFloor + random().nextInt(100));
+ }
+ idFloor += 100 + count;
+ return ids;
+ }
+
+
+ protected void sendNonDirectDeletesRequestReplicaWithRetry(Replica rep,
+ Set<Integer> byIdsSet, int expectedRfByIds,
+ Set<Integer> byQueriesSet, int expectedRfDBQ,
+ String coll) throws Exception {
+ // First add the docs indicated
+ List<String> byIdsList = new ArrayList<>();
+ List<String> byQueryList = new ArrayList<>();
+ List<SolrInputDocument> batch = new ArrayList<SolrInputDocument>(10);
+ for (int myId : byIdsSet) {
+ SolrInputDocument doc = new SolrInputDocument();
+ doc.addField(id, myId);
+ doc.addField("a_t", "hello" + id);
+ batch.add(doc);
+ byIdsList.add(Integer.toString(myId));
+ }
+ for (int myId : byQueriesSet) {
+ SolrInputDocument doc = new SolrInputDocument();
+ doc.addField(id, myId);
+ doc.addField("a_t", "hello" + id);
+ batch.add(doc);
+ byQueryList.add(Integer.toString(myId));
+ }
+
+ // Add the docs.
+ sendDocsWithRetry(batch, expectedRfDBQ, 5, 1);
+
+ // Delete the docs by ID indicated
+ UpdateRequest req = new UpdateRequest();
+ req.deleteById(byIdsList);
+ req.setParam(UpdateRequest.MIN_REPFACT, String.valueOf(expectedRfByIds));
+ sendNonDirectUpdateRequestReplicaWithRetry(rep, req, expectedRfByIds, coll);
+
+ //Delete the docs by query indicated.
+ req = new UpdateRequest();
+ req.deleteByQuery("id:(" + StringUtils.join(byQueriesSet, " OR ") + ")");
+ req.setParam(UpdateRequest.MIN_REPFACT, String.valueOf(expectedRfDBQ));
+ sendNonDirectUpdateRequestReplicaWithRetry(rep, req, expectedRfDBQ, coll);
+
+ }
protected void sendNonDirectUpdateRequestReplicaWithRetry(Replica replica, UpdateRequest up, int expectedRf, String collection) throws Exception {
try {
sendNonDirectUpdateRequestReplica(replica, up, expectedRf, collection);
+ Thread.sleep(100); // Let the system settle down before retrying
} catch (Exception e) {
sendNonDirectUpdateRequestReplica(replica, up, expectedRf, collection);
}
@@ -163,19 +252,18 @@ public class ReplicationFactorTest extends AbstractFullDistribZkTestBase {
@SuppressWarnings("rawtypes")
protected void sendNonDirectUpdateRequestReplica(Replica replica, UpdateRequest up, int expectedRf, String collection) throws Exception {
-
ZkCoreNodeProps zkProps = new ZkCoreNodeProps(replica);
String url = zkProps.getBaseUrl() + "/" + collection;
-
try (HttpSolrClient solrServer = getHttpSolrClient(url)) {
NamedList resp = solrServer.request(up);
NamedList hdr = (NamedList) resp.get("responseHeader");
Integer batchRf = (Integer)hdr.get(UpdateRequest.REPFACT);
+ // Note that this also tests if we're wonky and return an achieved rf greater than the number of live replicas.
assertTrue("Expected rf="+expectedRf+" for batch but got "+
- batchRf+"; clusterState: "+printClusterStateInfo(), batchRf == expectedRf);
+ batchRf + "; clusterState: " + printClusterStateInfo(), batchRf == expectedRf);
}
}
-
+
protected void testRf3() throws Exception {
int numShards = 1;
int replicationFactor = 3;
@@ -183,8 +271,8 @@ public class ReplicationFactorTest extends AbstractFullDistribZkTestBase {
String testCollectionName = "repfacttest_c8n_1x3";
String shardId = "shard1";
int minRf = 2;
-
- createCollection(testCollectionName, "conf1", numShards, replicationFactor, maxShardsPerNode);
+
+ createCollectionWithRetry(testCollectionName, "conf1", numShards, replicationFactor, maxShardsPerNode);
cloudClient.setDefaultCollection(testCollectionName);
List<Replica> replicas =
@@ -193,17 +281,30 @@ public class ReplicationFactorTest extends AbstractFullDistribZkTestBase {
int rf = sendDoc(1, minRf);
assertRf(3, "all replicas should be active", rf);
-
+
+ // Uses cloudClient do do it's work
+ doDBIdWithRetry(3, 5, "deletes should have propagated to all 3 replicas", 1);
+ doDBQWithRetry(3, 5, "deletes should have propagated to all 3 replicas", 1);
+
+
getProxyForReplica(replicas.get(0)).close();
rf = sendDoc(2, minRf);
assertRf(2, "one replica should be down", rf);
+ // Uses cloudClient do do it's work
+ doDBQWithRetry(2, 5, "deletes should have propagated to 2 replicas", 1);
+ doDBIdWithRetry(2, 5, "deletes should have propagated to 2 replicas", 1);
+
+
getProxyForReplica(replicas.get(1)).close();
rf = sendDoc(3, minRf);
assertRf(1, "both replicas should be down", rf);
-
+
+ doDBQWithRetry(1, 5, "deletes should have propagated to only 1 replica", 1);
+ doDBIdWithRetry(1, 5, "deletes should have propagated to only 1 replica", 1);
+
// heal the partitions
getProxyForReplica(replicas.get(0)).reopen();
getProxyForReplica(replicas.get(1)).reopen();
@@ -214,7 +315,10 @@ public class ReplicationFactorTest extends AbstractFullDistribZkTestBase {
rf = sendDoc(4, minRf);
assertRf(3, "partitions to replicas have been healed", rf);
-
+
+ doDBQWithRetry(3, 5, "deletes should have propagated to all 3 replicas", 1);
+ doDBIdWithRetry(3, 5, "deletes should have propagated to all 3 replicas", 1);
+
// now send a batch
List<SolrInputDocument> batch = new ArrayList<SolrInputDocument>(10);
for (int i=5; i < 15; i++) {
@@ -226,7 +330,10 @@ public class ReplicationFactorTest extends AbstractFullDistribZkTestBase {
int batchRf = sendDocsWithRetry(batch, minRf, 5, 1);
assertRf(3, "batch should have succeeded on all replicas", batchRf);
-
+
+ doDBQWithRetry(3, 5, "deletes should have propagated to only 1 replica", 15);
+ doDBIdWithRetry(3, 5, "deletes should have propagated to only 1 replica", 15);
+
// add some chaos to the batch
getProxyForReplica(replicas.get(0)).close();
@@ -242,6 +349,9 @@ public class ReplicationFactorTest extends AbstractFullDistribZkTestBase {
batchRf = sendDocsWithRetry(batch, minRf, 5, 1);
assertRf(2, "batch should have succeeded on 2 replicas (only one replica should be down)", batchRf);
+ doDBQWithRetry(2, 5, "deletes should have propagated to only 1 replica", 15);
+ doDBIdWithRetry(2, 5, "deletes should have propagated to only 1 replica", 15);
+
// close the 2nd replica, and send a 3rd batch with expected achieved rf=1
getProxyForReplica(replicas.get(1)).close();
@@ -256,12 +366,60 @@ public class ReplicationFactorTest extends AbstractFullDistribZkTestBase {
batchRf = sendDocsWithRetry(batch, minRf, 5, 1);
assertRf(1, "batch should have succeeded on the leader only (both replicas should be down)", batchRf);
+ doDBQWithRetry(1, 5, "deletes should have propagated to only 1 replica", 15);
+ doDBIdWithRetry(1, 5, "deletes should have propagated to only 1 replica", 15);
+
getProxyForReplica(replicas.get(0)).reopen();
getProxyForReplica(replicas.get(1)).reopen();
-
- Thread.sleep(2000);
- ensureAllReplicasAreActive(testCollectionName, shardId, numShards, replicationFactor, 30);
- }
+
+ Thread.sleep(2000);
+ ensureAllReplicasAreActive(testCollectionName, shardId, numShards, replicationFactor, 30);
+ }
+
+ protected void addDocs(Set<Integer> docIds, int expectedRf, int retries) throws Exception {
+
+ Integer[] idList = docIds.toArray(new Integer[docIds.size()]);
+ if (idList.length == 1) {
+ sendDoc(idList[0], expectedRf);
+ return;
+ }
+ List<SolrInputDocument> batch = new ArrayList<SolrInputDocument>(10);
+ for (int docId : idList) {
+ SolrInputDocument doc = new SolrInputDocument();
+ doc.addField(id, docId);
+ doc.addField("a_t", "hello" + docId);
+ batch.add(doc);
+ }
+ sendDocsWithRetry(batch, expectedRf, retries, 1);
+ }
+
+ protected void doDBQWithRetry(int expectedRf, int retries, String msg, int docsToAdd) throws Exception {
+ Set<Integer> docIds = getSomeIds(docsToAdd);
+ addDocs(docIds, expectedRf, retries);
+ UpdateRequest req = new UpdateRequest();
+ req.deleteByQuery("id:(" + StringUtils.join(docIds, " OR ") + ")");
+ req.setParam(UpdateRequest.MIN_REPFACT, String.valueOf(expectedRf));
+ doDelete(req, msg, expectedRf, retries);
+ }
+
+ protected void doDBIdWithRetry(int expectedRf, int retries, String msg, int docsToAdd) throws Exception {
+ Set<Integer> docIds = getSomeIds(docsToAdd);
+ addDocs(docIds, expectedRf, retries);
+ UpdateRequest req = new UpdateRequest();
+ req.deleteById(StringUtils.join(docIds, ","));
+ req.setParam(UpdateRequest.MIN_REPFACT, String.valueOf(expectedRf));
+ doDelete(req, msg, expectedRf, retries);
+ }
+
+ protected void doDelete(UpdateRequest req, String msg, int expectedRf, int retries) throws IOException, SolrServerException, InterruptedException {
+ int achievedRf = -1;
+ for (int idx = 0; idx < retries; ++idx) {
+ achievedRf = cloudClient.getMinAchievedReplicationFactor(cloudClient.getDefaultCollection(), cloudClient.request(req));
+ if (achievedRf == expectedRf) return;
+ Thread.sleep(1000);
+ }
+ assertEquals(msg, expectedRf, achievedRf);
+ }
protected int sendDoc(int docId, int minRf) throws Exception {
UpdateRequest up = new UpdateRequest();
@@ -278,6 +436,23 @@ public class ReplicationFactorTest extends AbstractFullDistribZkTestBase {
String assertionFailedMessage =
String.format(Locale.ENGLISH, "Expected rf=%d because %s but got %d", expected, explain, actual);
fail(assertionFailedMessage+"; clusterState: "+printClusterStateInfo());
- }
- }
+ }
+ }
+
+ void createCollectionWithRetry(String testCollectionName, String config, int numShards, int replicationFactor, int maxShardsPerNode) throws IOException, SolrServerException, InterruptedException {
+ CollectionAdminResponse resp = createCollection(testCollectionName, "conf1", numShards, replicationFactor, maxShardsPerNode);
+
+ if (resp.getResponse().get("failure") != null) {
+ Thread.sleep(5000); // let system settle down. This should be very rare.
+
+ CollectionAdminRequest.deleteCollection(testCollectionName).process(cloudClient);
+
+ resp = createCollection(testCollectionName, "conf1", numShards, replicationFactor, maxShardsPerNode);
+
+ if (resp.getResponse().get("failure") != null) {
+ fail("Could not create " + testCollectionName);
+ }
+ }
+ }
+
}