You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by sh...@apache.org on 2020/01/28 13:47:54 UTC
[lucene-solr] 01/01: SOLR-13996: Refactor
HttpShardHandler.prepDistributed method into smaller pieces
This is an automated email from the ASF dual-hosted git repository.
shalin pushed a commit to branch jira/solr-13996
in repository https://gitbox.apache.org/repos/asf/lucene-solr.git
commit 8381cf14ef62db0fe1d1014e92f7e1f417fcd8fe
Author: Shalin Shekhar Mangar <sh...@apache.org>
AuthorDate: Tue Jan 28 19:17:42 2020 +0530
SOLR-13996: Refactor HttpShardHandler.prepDistributed method into smaller pieces
---
.../solr/handler/component/CloudReplicaSource.java | 246 +++++++++++++++
.../solr/handler/component/HttpShardHandler.java | 350 ++++++---------------
.../handler/component/HttpShardHandlerFactory.java | 4 +-
.../handler/component/LegacyReplicaSource.java | 77 +++++
.../solr/handler/component/ReplicaSource.java | 42 +++
5 files changed, 455 insertions(+), 264 deletions(-)
diff --git a/solr/core/src/java/org/apache/solr/handler/component/CloudReplicaSource.java b/solr/core/src/java/org/apache/solr/handler/component/CloudReplicaSource.java
new file mode 100644
index 0000000..272c00d
--- /dev/null
+++ b/solr/core/src/java/org/apache/solr/handler/component/CloudReplicaSource.java
@@ -0,0 +1,246 @@
+/*
+ * 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.
+ */
+
+package org.apache.solr.handler.component;
+
+import java.lang.invoke.MethodHandles;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+
+import org.apache.solr.client.solrj.routing.ReplicaListTransformer;
+import org.apache.solr.client.solrj.util.ClientUtils;
+import org.apache.solr.common.SolrException;
+import org.apache.solr.common.cloud.ClusterState;
+import org.apache.solr.common.cloud.DocCollection;
+import org.apache.solr.common.cloud.Replica;
+import org.apache.solr.common.cloud.Slice;
+import org.apache.solr.common.cloud.ZkStateReader;
+import org.apache.solr.common.params.ShardParams;
+import org.apache.solr.common.params.SolrParams;
+import org.apache.solr.common.util.StrUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A replica source for solr cloud mode
+ */
+class CloudReplicaSource implements ReplicaSource {
+ private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ private String[] slices;
+ private List<String>[] replicas;
+
+ private CloudReplicaSource(Builder builder) {
+ final String shards = builder.params.get(ShardParams.SHARDS);
+ if (shards != null) {
+ withShardsParam(builder, shards);
+ } else {
+ withClusterState(builder, builder.params);
+ }
+ }
+
+ private void withClusterState(final Builder builder, SolrParams params) {
+ ClusterState clusterState = builder.zkStateReader.getClusterState();
+ String shardKeys = params.get(ShardParams._ROUTE_);
+
+ // This will be the complete list of slices we need to query for this request.
+ Map<String, Slice> sliceMap = new HashMap<>();
+
+ // we need to find out what collections this request is for.
+
+ // A comma-separated list of specified collections.
+ // Eg: "collection1,collection2,collection3"
+ String collections = params.get("collection");
+ if (collections != null) {
+ // If there were one or more collections specified in the query, split
+ // each parameter and store as a separate member of a List.
+ List<String> collectionList = StrUtils.splitSmart(collections, ",",
+ true);
+ // In turn, retrieve the slices that cover each collection from the
+ // cloud state and add them to the Map 'slices'.
+ for (String collectionName : collectionList) {
+ // The original code produced <collection-name>_<shard-name> when the collections
+ // parameter was specified (see ClientUtils.appendMap)
+ // Is this necessary if ony one collection is specified?
+ // i.e. should we change multiCollection to collectionList.size() > 1?
+ addSlices(sliceMap, clusterState, params, collectionName, shardKeys, true);
+ }
+ } else {
+ // just this collection
+ addSlices(sliceMap, clusterState, params, builder.collection, shardKeys, false);
+ }
+
+ this.slices = sliceMap.keySet().toArray(new String[sliceMap.size()]);
+ this.replicas = new List[slices.length];
+ for (int i = 0; i < slices.length; i++) {
+ String sliceName = slices[i];
+ replicas[i] = findReplicas(builder, null, clusterState, sliceMap.get(sliceName));
+ }
+ }
+
+ private void withShardsParam(final Builder builder, String shardsParam) {
+ List<String> sliceOrUrls = StrUtils.splitSmart(shardsParam, ",", true);
+ this.slices = new String[sliceOrUrls.size()];
+ this.replicas = new List[sliceOrUrls.size()];
+
+ ClusterState clusterState = builder.zkStateReader.getClusterState();
+
+ for (int i = 0; i < sliceOrUrls.size(); i++) {
+ String sliceOrUrl = sliceOrUrls.get(i);
+ if (sliceOrUrl.indexOf('/') < 0) {
+ // this is a logical shard
+ this.slices[i] = sliceOrUrl;
+ replicas[i] = findReplicas(builder, shardsParam, clusterState, clusterState.getCollection(builder.collection).getSlice(sliceOrUrl));
+ } else {
+ // this has urls
+ this.replicas[i] = StrUtils.splitSmart(sliceOrUrl, "|", true);
+ builder.replicaListTransformer.transform(replicas[i]);
+ builder.hostChecker.checkWhitelist(clusterState, shardsParam, replicas[i]);
+ }
+ }
+ }
+
+ private List<String> findReplicas(Builder builder, String shardsParam, ClusterState clusterState, Slice slice) {
+ if (slice == null) {
+ // Treat this the same as "all servers down" for a slice, and let things continue
+ // if partial results are acceptable
+ return Collections.emptyList();
+ } else {
+ final Predicate<Replica> isShardLeader = new IsLeaderPredicate(builder.zkStateReader, clusterState, builder.collection, slice.getName());
+ List<Replica> list = slice.getReplicas()
+ .stream()
+ .filter(replica -> replica.isActive(clusterState.getLiveNodes()))
+ .filter(replica -> !builder.onlyNrt || (replica.getType() == Replica.Type.NRT || (replica.getType() == Replica.Type.TLOG && isShardLeader.test(replica))))
+ .collect(Collectors.toList());
+ builder.replicaListTransformer.transform(list);
+ List<String> collect = list.stream().map(Replica::getCoreUrl).collect(Collectors.toList());
+ builder.hostChecker.checkWhitelist(clusterState, shardsParam, collect);
+ return collect;
+ }
+ }
+
+ private void addSlices(Map<String, Slice> target, ClusterState state, SolrParams params, String collectionName, String shardKeys, boolean multiCollection) {
+ DocCollection coll = state.getCollection(collectionName);
+ Collection<Slice> slices = coll.getRouter().getSearchSlices(shardKeys, params, coll);
+ ClientUtils.addSlices(target, collectionName, slices, multiCollection);
+ }
+
+ @Override
+ public List<String> getSliceNames() {
+ return Collections.unmodifiableList(Arrays.asList(slices));
+ }
+
+ @Override
+ public List<String> getReplicasBySlice(int sliceNumber) {
+ assert sliceNumber >= 0 && sliceNumber < replicas.length;
+ return replicas[sliceNumber];
+ }
+
+ @Override
+ public int getSliceCount() {
+ return slices.length;
+ }
+
+ /**
+ * A predicate to test if a replica is the leader according to {@link ZkStateReader#getLeaderRetry(String, String)}.
+ * <p>
+ * The result of getLeaderRetry is cached in the first call so that subsequent tests are faster and do not block.
+ */
+ private static class IsLeaderPredicate implements Predicate<Replica> {
+ private final ZkStateReader zkStateReader;
+ private final ClusterState clusterState;
+ private final String collectionName;
+ private final String sliceName;
+ private Replica shardLeader = null;
+
+ public IsLeaderPredicate(ZkStateReader zkStateReader, ClusterState clusterState, String collectionName, String sliceName) {
+ this.zkStateReader = zkStateReader;
+ this.clusterState = clusterState;
+ this.collectionName = collectionName;
+ this.sliceName = sliceName;
+ }
+
+ @Override
+ public boolean test(Replica replica) {
+ if (shardLeader == null) {
+ try {
+ shardLeader = zkStateReader.getLeaderRetry(collectionName, sliceName);
+ } catch (InterruptedException e) {
+ throw new SolrException(SolrException.ErrorCode.SERVICE_UNAVAILABLE,
+ "Exception finding leader for shard " + sliceName + " in collection "
+ + collectionName, e);
+ } catch (SolrException e) {
+ if (log.isDebugEnabled()) {
+ log.debug("Exception finding leader for shard {} in collection {}. Collection State: {}",
+ sliceName, collectionName, clusterState.getCollectionOrNull(collectionName));
+ }
+ throw e;
+ }
+ }
+ return replica.getName().equals(shardLeader.getName());
+ }
+ }
+
+ static class Builder {
+ private String collection;
+ private ZkStateReader zkStateReader;
+ private SolrParams params;
+ private boolean onlyNrt;
+ private ReplicaListTransformer replicaListTransformer;
+ private HttpShardHandlerFactory.WhitelistHostChecker hostChecker;
+
+ public Builder collection(String collection) {
+ this.collection = collection;
+ return this;
+ }
+
+ public Builder zkStateReader(ZkStateReader stateReader) {
+ this.zkStateReader = stateReader;
+ return this;
+ }
+
+ public Builder params(SolrParams params) {
+ this.params = params;
+ return this;
+ }
+
+ public Builder onlyNrt(boolean onlyNrt) {
+ this.onlyNrt = onlyNrt;
+ return this;
+ }
+
+ public Builder replicaListTransformer(ReplicaListTransformer replicaListTransformer) {
+ this.replicaListTransformer = replicaListTransformer;
+ return this;
+ }
+
+ public Builder whitelistHostChecker(HttpShardHandlerFactory.WhitelistHostChecker hostChecker) {
+ this.hostChecker = hostChecker;
+ return this;
+ }
+
+ public CloudReplicaSource build() {
+ return new CloudReplicaSource(this);
+ }
+ }
+}
diff --git a/solr/core/src/java/org/apache/solr/handler/component/HttpShardHandler.java b/solr/core/src/java/org/apache/solr/handler/component/HttpShardHandler.java
index 8f8c48e..92d96b3 100644
--- a/solr/core/src/java/org/apache/solr/handler/component/HttpShardHandler.java
+++ b/solr/core/src/java/org/apache/solr/handler/component/HttpShardHandler.java
@@ -17,11 +17,8 @@
package org.apache.solr.handler.component;
import java.io.IOException;
-import java.lang.invoke.MethodHandles;
import java.net.ConnectException;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -32,7 +29,6 @@ import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
-import java.util.function.Predicate;
import io.opentracing.Span;
import io.opentracing.Tracer;
@@ -40,43 +36,32 @@ import io.opentracing.propagation.Format;
import org.apache.solr.client.solrj.SolrRequest;
import org.apache.solr.client.solrj.SolrResponse;
import org.apache.solr.client.solrj.SolrServerException;
-import org.apache.solr.client.solrj.impl.BinaryResponseParser;
import org.apache.solr.client.solrj.impl.Http2SolrClient;
import org.apache.solr.client.solrj.impl.LBSolrClient;
import org.apache.solr.client.solrj.request.QueryRequest;
import org.apache.solr.client.solrj.routing.ReplicaListTransformer;
-import org.apache.solr.client.solrj.util.ClientUtils;
import org.apache.solr.cloud.CloudDescriptor;
import org.apache.solr.cloud.ZkController;
import org.apache.solr.common.SolrException;
-import org.apache.solr.common.SolrException.ErrorCode;
-import org.apache.solr.common.cloud.ClusterState;
-import org.apache.solr.common.cloud.DocCollection;
import org.apache.solr.common.cloud.Replica;
-import org.apache.solr.common.cloud.Slice;
import org.apache.solr.common.cloud.ZkCoreNodeProps;
import org.apache.solr.common.params.CommonParams;
import org.apache.solr.common.params.ModifiableSolrParams;
import org.apache.solr.common.params.ShardParams;
import org.apache.solr.common.params.SolrParams;
-import org.apache.solr.common.util.JavaBinCodec;
import org.apache.solr.common.util.NamedList;
-import org.apache.solr.common.util.StrUtils;
import org.apache.solr.core.CoreDescriptor;
import org.apache.solr.request.SolrQueryRequest;
import org.apache.solr.request.SolrRequestInfo;
import org.apache.solr.util.tracing.GlobalTracer;
import org.apache.solr.util.tracing.SolrRequestCarrier;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
public class HttpShardHandler extends ShardHandler {
-
/**
* If the request context map has an entry with this key and Boolean.TRUE as value,
* {@link #prepDistributed(ResponseBuilder)} will only include {@link org.apache.solr.common.cloud.Replica.Type#NRT} replicas as possible
- * destination of the distributed request (or a leader replica of type {@link org.apache.solr.common.cloud.Replica.Type#TLOG}). This is used
+ * destination of the distributed request (or a leader replica of type {@link org.apache.solr.common.cloud.Replica.Type#TLOG}). This is used
* by the RealtimeGet handler, since other types of replicas shouldn't respond to RTG requests
*/
public static String ONLY_NRT_REPLICAS = "distribOnlyRealtime";
@@ -84,11 +69,9 @@ public class HttpShardHandler extends ShardHandler {
private HttpShardHandlerFactory httpShardHandlerFactory;
private CompletionService<ShardResponse> completionService;
private Set<Future<ShardResponse>> pending;
- private Map<String,List<String>> shardToURLs;
+ private Map<String, List<String>> shardToURLs;
private Http2SolrClient httpClient;
- private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
-
public HttpShardHandler(HttpShardHandlerFactory httpShardHandlerFactory, Http2SolrClient httpClient) {
this.httpClient = httpClient;
this.httpShardHandlerFactory = httpShardHandlerFactory;
@@ -141,19 +124,12 @@ public class HttpShardHandler extends ShardHandler {
return urls;
}
- private static final BinaryResponseParser READ_STR_AS_CHARSEQ_PARSER = new BinaryResponseParser() {
- @Override
- protected JavaBinCodec createCodec() {
- return new JavaBinCodec(null, stringCache).setReadStringAsCharSeq(true);
- }
- };
-
@Override
public void submit(final ShardRequest sreq, final String shard, final ModifiableSolrParams params) {
// do this outside of the callable for thread safety reasons
final List<String> urls = getURLs(shard);
final Tracer tracer = GlobalTracer.getTracer();
- final Span span = tracer != null? tracer.activeSpan() : null;
+ final Span span = tracer != null ? tracer.activeSpan() : null;
Callable<ShardResponse> task = () -> {
@@ -183,7 +159,7 @@ public class HttpShardHandler extends ShardHandler {
// req.setResponseParser(new BinaryResponseParser());
// if there are no shards available for a slice, urls.size()==0
- if (urls.size()==0) {
+ if (urls.size() == 0) {
// TODO: what's the right error code here? We should use the same thing when
// all of the servers for a shard are down.
throw new SolrException(SolrException.ErrorCode.SERVICE_UNAVAILABLE, "no servers hosting shard: " + shard);
@@ -198,13 +174,12 @@ public class HttpShardHandler extends ShardHandler {
ssr.nl = rsp.getResponse();
srsp.setShardAddress(rsp.getServer());
}
- }
- catch( ConnectException cex ) {
+ } catch (ConnectException cex) {
srsp.setException(cex); //????
} catch (Exception th) {
srsp.setException(th);
if (th instanceof SolrException) {
- srsp.setResponseCode(((SolrException)th).code());
+ srsp.setResponseCode(((SolrException) th).code());
} else {
srsp.setResponseCode(-1);
}
@@ -216,13 +191,13 @@ public class HttpShardHandler extends ShardHandler {
};
try {
- if (shard != null) {
+ if (shard != null) {
MDC.put("ShardRequest.shards", shard);
}
- if (urls != null && !urls.isEmpty()) {
+ if (urls != null && !urls.isEmpty()) {
MDC.put("ShardRequest.urlList", urls.toString());
}
- pending.add( completionService.submit(task) );
+ pending.add(completionService.submit(task));
} finally {
MDC.remove("ShardRequest.shards");
MDC.remove("ShardRequest.urlList");
@@ -233,26 +208,25 @@ public class HttpShardHandler extends ShardHandler {
req.setBasePath(url);
return httpClient.request(req);
}
-
+
/**
* Subclasses could modify the request based on the shard
*/
- protected QueryRequest makeQueryRequest(final ShardRequest sreq, ModifiableSolrParams params, String shard)
- {
+ protected QueryRequest makeQueryRequest(final ShardRequest sreq, ModifiableSolrParams params, String shard) {
// use generic request to avoid extra processing of queries
return new QueryRequest(params);
}
-
+
/**
* Subclasses could modify the Response based on the the shard
*/
- protected ShardResponse transfomResponse(final ShardRequest sreq, ShardResponse rsp, String shard)
- {
+ protected ShardResponse transfomResponse(final ShardRequest sreq, ShardResponse rsp, String shard) {
return rsp;
}
- /** returns a ShardResponse of the last response correlated with a ShardRequest. This won't
- * return early if it runs into an error.
+ /**
+ * returns a ShardResponse of the last response correlated with a ShardRequest. This won't
+ * return early if it runs into an error.
**/
@Override
public ShardResponse takeCompletedIncludingErrors() {
@@ -260,16 +234,17 @@ public class HttpShardHandler extends ShardHandler {
}
- /** returns a ShardResponse of the last response correlated with a ShardRequest,
+ /**
+ * returns a ShardResponse of the last response correlated with a ShardRequest,
* or immediately returns a ShardResponse if there was an error detected
*/
@Override
public ShardResponse takeCompletedOrError() {
return take(true);
}
-
+
private ShardResponse take(boolean bailOnError) {
-
+
while (pending.size() > 0) {
try {
Future<ShardResponse> future = completionService.take();
@@ -289,7 +264,7 @@ public class HttpShardHandler extends ShardHandler {
} catch (ExecutionException e) {
// should be impossible... the problem with catching the exception
// at this level is we don't know what ShardRequest it applied to
- throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, "Impossible Exception",e);
+ throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, "Impossible Exception", e);
}
}
return null;
@@ -309,237 +284,64 @@ public class HttpShardHandler extends ShardHandler {
final SolrParams params = req.getParams();
final String shards = params.get(ShardParams.SHARDS);
- // since the cost of grabbing cloud state is still up in the air, we grab it only
- // if we need it.
- ClusterState clusterState = null;
- Map<String,Slice> slices = null;
CoreDescriptor coreDescriptor = req.getCore().getCoreDescriptor();
CloudDescriptor cloudDescriptor = coreDescriptor.getCloudDescriptor();
ZkController zkController = req.getCore().getCoreContainer().getZkController();
final ReplicaListTransformer replicaListTransformer = httpShardHandlerFactory.getReplicaListTransformer(req);
- if (shards != null) {
- List<String> lst = StrUtils.splitSmart(shards, ",", true);
- rb.shards = lst.toArray(new String[lst.size()]);
- rb.slices = new String[rb.shards.length];
-
- if (zkController != null) {
- // figure out which shards are slices
- for (int i=0; i<rb.shards.length; i++) {
- if (rb.shards[i].indexOf('/') < 0) {
- // this is a logical shard
- rb.slices[i] = rb.shards[i];
- rb.shards[i] = null;
- }
- }
- }
- } else if (zkController != null) {
- // we weren't provided with an explicit list of slices to query via "shards", so use the cluster state
-
- clusterState = zkController.getClusterState();
- String shardKeys = params.get(ShardParams._ROUTE_);
-
- // This will be the complete list of slices we need to query for this request.
- slices = new HashMap<>();
-
- // we need to find out what collections this request is for.
-
- // A comma-separated list of specified collections.
- // Eg: "collection1,collection2,collection3"
- String collections = params.get("collection");
- if (collections != null) {
- // If there were one or more collections specified in the query, split
- // each parameter and store as a separate member of a List.
- List<String> collectionList = StrUtils.splitSmart(collections, ",",
- true);
- // In turn, retrieve the slices that cover each collection from the
- // cloud state and add them to the Map 'slices'.
- for (String collectionName : collectionList) {
- // The original code produced <collection-name>_<shard-name> when the collections
- // parameter was specified (see ClientUtils.appendMap)
- // Is this necessary if ony one collection is specified?
- // i.e. should we change multiCollection to collectionList.size() > 1?
- addSlices(slices, clusterState, params, collectionName, shardKeys, true);
- }
- } else {
- // just this collection
- String collectionName = cloudDescriptor.getCollectionName();
- addSlices(slices, clusterState, params, collectionName, shardKeys, false);
- }
-
-
- // Store the logical slices in the ResponseBuilder and create a new
- // String array to hold the physical shards (which will be mapped
- // later).
- rb.slices = slices.keySet().toArray(new String[slices.size()]);
- rb.shards = new String[rb.slices.length];
- }
-
HttpShardHandlerFactory.WhitelistHostChecker hostChecker = httpShardHandlerFactory.getWhitelistHostChecker();
if (shards != null && zkController == null && hostChecker.isWhitelistHostCheckingEnabled() && !hostChecker.hasExplicitWhitelist()) {
- throw new SolrException(ErrorCode.FORBIDDEN, "HttpShardHandlerFactory "+HttpShardHandlerFactory.INIT_SHARDS_WHITELIST
- +" not configured but required (in lieu of ZkController and ClusterState) when using the '"+ShardParams.SHARDS+"' parameter."
- +HttpShardHandlerFactory.SET_SOLR_DISABLE_SHARDS_WHITELIST_CLUE);
+ throw new SolrException(SolrException.ErrorCode.FORBIDDEN, "HttpShardHandlerFactory " + HttpShardHandlerFactory.INIT_SHARDS_WHITELIST
+ + " not configured but required (in lieu of ZkController and ClusterState) when using the '" + ShardParams.SHARDS + "' parameter."
+ + HttpShardHandlerFactory.SET_SOLR_DISABLE_SHARDS_WHITELIST_CLUE);
}
- //
- // Map slices to shards
- //
+ ReplicaSource replicaSource;
if (zkController != null) {
-
- // Are we hosting the shard that this request is for, and are we active? If so, then handle it ourselves
- // and make it a non-distributed request.
- String ourSlice = cloudDescriptor.getShardId();
- String ourCollection = cloudDescriptor.getCollectionName();
- // Some requests may only be fulfilled by replicas of type Replica.Type.NRT
- boolean onlyNrtReplicas = Boolean.TRUE == req.getContext().get(ONLY_NRT_REPLICAS);
- if (rb.slices.length == 1 && rb.slices[0] != null
- && ( rb.slices[0].equals(ourSlice) || rb.slices[0].equals(ourCollection + "_" + ourSlice) ) // handle the <collection>_<slice> format
- && cloudDescriptor.getLastPublished() == Replica.State.ACTIVE
- && (!onlyNrtReplicas || cloudDescriptor.getReplicaType() == Replica.Type.NRT)) {
- boolean shortCircuit = params.getBool("shortCircuit", true); // currently just a debugging parameter to check distrib search on a single node
-
- String targetHandler = params.get(ShardParams.SHARDS_QT);
- shortCircuit = shortCircuit && targetHandler == null; // if a different handler is specified, don't short-circuit
-
- if (shortCircuit) {
- rb.isDistrib = false;
- rb.shortCircuitedURL = ZkCoreNodeProps.getCoreUrl(zkController.getBaseUrl(), coreDescriptor.getName());
- if (hostChecker.isWhitelistHostCheckingEnabled() && hostChecker.hasExplicitWhitelist()) {
- /*
- * We only need to check the host whitelist if there is an explicit whitelist (other than all the live nodes)
- * when the "shards" indicate cluster state elements only
- */
- hostChecker.checkWhitelist(clusterState, shards, Arrays.asList(rb.shortCircuitedURL));
- }
- return;
- }
- // We shouldn't need to do anything to handle "shard.rows" since it was previously meant to be an optimization?
- }
-
- if (clusterState == null && zkController != null) {
- clusterState = zkController.getClusterState();
+ replicaSource = new CloudReplicaSource.Builder()
+ .params(params)
+ .zkStateReader(zkController.getZkStateReader())
+ .whitelistHostChecker(hostChecker)
+ .replicaListTransformer(replicaListTransformer)
+ .collection(cloudDescriptor.getCollectionName())
+ .onlyNrt(Boolean.TRUE == req.getContext().get(ONLY_NRT_REPLICAS))
+ .build();
+ rb.slices = replicaSource.getSliceNames().toArray(new String[replicaSource.getSliceCount()]);
+ if (shortCircuitIfPossible(rb, req, params, shards, coreDescriptor, cloudDescriptor, zkController, hostChecker)) {
+ return;
}
-
-
- for (int i=0; i<rb.shards.length; i++) {
- if (rb.shards[i] != null) {
- final List<String> shardUrls = StrUtils.splitSmart(rb.shards[i], "|", true);
- replicaListTransformer.transform(shardUrls);
- hostChecker.checkWhitelist(clusterState, shards, shardUrls);
- // And now recreate the | delimited list of equivalent servers
- rb.shards[i] = createSliceShardsStr(shardUrls);
- } else {
- if (slices == null) {
- slices = clusterState.getCollection(cloudDescriptor.getCollectionName()).getSlicesMap();
- }
- String sliceName = rb.slices[i];
-
- Slice slice = slices.get(sliceName);
-
- if (slice==null) {
- // Treat this the same as "all servers down" for a slice, and let things continue
- // if partial results are acceptable
- rb.shards[i] = "";
- continue;
- // throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "no such shard: " + sliceName);
- }
- final Predicate<Replica> isShardLeader = new Predicate<Replica>() {
- private Replica shardLeader = null;
-
- @Override
- public boolean test(Replica replica) {
- if (shardLeader == null) {
- try {
- shardLeader = zkController.getZkStateReader().getLeaderRetry(cloudDescriptor.getCollectionName(), slice.getName());
- } catch (InterruptedException e) {
- throw new SolrException(SolrException.ErrorCode.SERVICE_UNAVAILABLE, "Exception finding leader for shard " + slice.getName() + " in collection "
- + cloudDescriptor.getCollectionName(), e);
- } catch (SolrException e) {
- if (log.isDebugEnabled()) {
- log.debug("Exception finding leader for shard {} in collection {}. Collection State: {}",
- slice.getName(), cloudDescriptor.getCollectionName(), zkController.getZkStateReader().getClusterState().getCollectionOrNull(cloudDescriptor.getCollectionName()));
- }
- throw e;
- }
- }
- return replica.getName().equals(shardLeader.getName());
- }
- };
-
- final List<Replica> eligibleSliceReplicas = collectEligibleReplicas(slice, clusterState, onlyNrtReplicas, isShardLeader);
-
- final List<String> shardUrls = transformReplicasToShardUrls(replicaListTransformer, eligibleSliceReplicas);
-
- if (hostChecker.isWhitelistHostCheckingEnabled() && hostChecker.hasExplicitWhitelist()) {
- /*
- * We only need to check the host whitelist if there is an explicit whitelist (other than all the live nodes)
- * when the "shards" indicate cluster state elements only
- */
- hostChecker.checkWhitelist(clusterState, shards, shardUrls);
- }
-
- // And now recreate the | delimited list of equivalent servers
- final String sliceShardsStr = createSliceShardsStr(shardUrls);
- if (sliceShardsStr.isEmpty()) {
- boolean tolerant = ShardParams.getShardsTolerantAsBool(rb.req.getParams());
- if (!tolerant) {
- // stop the check when there are no replicas available for a shard
- throw new SolrException(SolrException.ErrorCode.SERVICE_UNAVAILABLE,
- "no servers hosting shard: " + rb.slices[i]);
- }
- }
- rb.shards[i] = sliceShardsStr;
+ for (int i = 0; i < rb.slices.length; i++) {
+ if (!ShardParams.getShardsTolerantAsBool(params) && replicaSource.getReplicasBySlice(i).isEmpty()) {
+ // stop the check when there are no replicas available for a shard
+ // todo fix use of slices[i] which can be null if user specified urls in shards param
+ throw new SolrException(SolrException.ErrorCode.SERVICE_UNAVAILABLE,
+ "no servers hosting shard: " + rb.slices[i]);
}
}
} else {
- if (shards != null) {
- // No cloud, verbatim check of shards
- hostChecker.checkWhitelist(shards, new ArrayList<>(Arrays.asList(shards.split("[,|]"))));
- }
+ replicaSource = new LegacyReplicaSource.Builder()
+ .whitelistHostChecker(hostChecker)
+ .shards(shards)
+ .build();
+ rb.slices = new String[replicaSource.getSliceCount()];
}
+
+ rb.shards = new String[rb.slices.length];
+ for (int i = 0; i < rb.slices.length; i++) {
+ rb.shards[i] = createSliceShardsStr(replicaSource.getReplicasBySlice(i));
+ }
+
String shards_rows = params.get(ShardParams.SHARDS_ROWS);
- if(shards_rows != null) {
+ if (shards_rows != null) {
rb.shards_rows = Integer.parseInt(shards_rows);
}
String shards_start = params.get(ShardParams.SHARDS_START);
- if(shards_start != null) {
+ if (shards_start != null) {
rb.shards_start = Integer.parseInt(shards_start);
}
}
- private static List<Replica> collectEligibleReplicas(Slice slice, ClusterState clusterState, boolean onlyNrtReplicas, Predicate<Replica> isShardLeader) {
- final Collection<Replica> allSliceReplicas = slice.getReplicasMap().values();
- final List<Replica> eligibleSliceReplicas = new ArrayList<>(allSliceReplicas.size());
- for (Replica replica : allSliceReplicas) {
- if (!clusterState.liveNodesContain(replica.getNodeName())
- || replica.getState() != Replica.State.ACTIVE
- || (onlyNrtReplicas && replica.getType() == Replica.Type.PULL)) {
- continue;
- }
-
- if (onlyNrtReplicas && replica.getType() == Replica.Type.TLOG) {
- if (!isShardLeader.test(replica)) {
- continue;
- }
- }
- eligibleSliceReplicas.add(replica);
- }
- return eligibleSliceReplicas;
- }
-
- private static List<String> transformReplicasToShardUrls(final ReplicaListTransformer replicaListTransformer, final List<Replica> eligibleSliceReplicas) {
- replicaListTransformer.transform(eligibleSliceReplicas);
-
- final List<String> shardUrls = new ArrayList<>(eligibleSliceReplicas.size());
- for (Replica replica : eligibleSliceReplicas) {
- String url = ZkCoreNodeProps.getCoreUrl(replica);
- shardUrls.add(url);
- }
- return shardUrls;
- }
-
private static String createSliceShardsStr(final List<String> shardUrls) {
final StringBuilder sliceShardsStr = new StringBuilder();
boolean first = true;
@@ -554,17 +356,41 @@ public class HttpShardHandler extends ShardHandler {
return sliceShardsStr.toString();
}
-
- private void addSlices(Map<String,Slice> target, ClusterState state, SolrParams params, String collectionName, String shardKeys, boolean multiCollection) {
- DocCollection coll = state.getCollection(collectionName);
- Collection<Slice> slices = coll.getRouter().getSearchSlices(shardKeys, params , coll);
- ClientUtils.addSlices(target, collectionName, slices, multiCollection);
+ private boolean shortCircuitIfPossible(ResponseBuilder rb, SolrQueryRequest req, SolrParams params, String shards, CoreDescriptor coreDescriptor, CloudDescriptor cloudDescriptor, ZkController zkController, HttpShardHandlerFactory.WhitelistHostChecker hostChecker) {
+ // Are we hosting the shard that this request is for, and are we active? If so, then handle it ourselves
+ // and make it a non-distributed request.
+ String ourSlice = cloudDescriptor.getShardId();
+ String ourCollection = cloudDescriptor.getCollectionName();
+ // Some requests may only be fulfilled by replicas of type Replica.Type.NRT
+ boolean onlyNrtReplicas = Boolean.TRUE == req.getContext().get(ONLY_NRT_REPLICAS);
+ if (rb.slices.length == 1 && rb.slices[0] != null
+ && (rb.slices[0].equals(ourSlice) || rb.slices[0].equals(ourCollection + "_" + ourSlice)) // handle the <collection>_<slice> format
+ && cloudDescriptor.getLastPublished() == Replica.State.ACTIVE
+ && (!onlyNrtReplicas || cloudDescriptor.getReplicaType() == Replica.Type.NRT)) {
+ boolean shortCircuit = params.getBool("shortCircuit", true); // currently just a debugging parameter to check distrib search on a single node
+
+ String targetHandler = params.get(ShardParams.SHARDS_QT);
+ shortCircuit = shortCircuit && targetHandler == null; // if a different handler is specified, don't short-circuit
+
+ if (shortCircuit) {
+ rb.isDistrib = false;
+ rb.shortCircuitedURL = ZkCoreNodeProps.getCoreUrl(zkController.getBaseUrl(), coreDescriptor.getName());
+ if (hostChecker.isWhitelistHostCheckingEnabled() && hostChecker.hasExplicitWhitelist()) {
+ /*
+ * We only need to check the host whitelist if there is an explicit whitelist (other than all the live nodes)
+ * when the "shards" indicate cluster state elements only
+ */
+ hostChecker.checkWhitelist(zkController.getClusterState(), shards, Collections.singletonList(rb.shortCircuitedURL));
+ }
+ return true;
+ }
+ // We shouldn't need to do anything to handle "shard.rows" since it was previously meant to be an optimization?
+ }
+ return false;
}
- public ShardHandlerFactory getShardHandlerFactory(){
+ public ShardHandlerFactory getShardHandlerFactory() {
return httpShardHandlerFactory;
}
-
-
}
diff --git a/solr/core/src/java/org/apache/solr/handler/component/HttpShardHandlerFactory.java b/solr/core/src/java/org/apache/solr/handler/component/HttpShardHandlerFactory.java
index a091874..3e7e39d 100644
--- a/solr/core/src/java/org/apache/solr/handler/component/HttpShardHandlerFactory.java
+++ b/solr/core/src/java/org/apache/solr/handler/component/HttpShardHandlerFactory.java
@@ -431,8 +431,8 @@ public class HttpShardHandlerFactory extends ShardHandlerFactory implements org.
/**
* Creates a new completion service for use by a single set of distributed requests.
*/
- public CompletionService newCompletionService() {
- return new ExecutorCompletionService<ShardResponse>(commExecutor);
+ public CompletionService<ShardResponse> newCompletionService() {
+ return new ExecutorCompletionService<>(commExecutor);
}
/**
diff --git a/solr/core/src/java/org/apache/solr/handler/component/LegacyReplicaSource.java b/solr/core/src/java/org/apache/solr/handler/component/LegacyReplicaSource.java
new file mode 100644
index 0000000..a67d3a5
--- /dev/null
+++ b/solr/core/src/java/org/apache/solr/handler/component/LegacyReplicaSource.java
@@ -0,0 +1,77 @@
+/*
+ * 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.
+ */
+
+package org.apache.solr.handler.component;
+
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.solr.common.util.StrUtils;
+
+/**
+ * A replica source for solr stand alone mode
+ */
+class LegacyReplicaSource implements ReplicaSource {
+ private List<String>[] replicas;
+
+ public LegacyReplicaSource(Builder builder) {
+ List<String> list = StrUtils.splitSmart(builder.shardsParam, ",", true);
+ replicas = new List[list.size()];
+ for (int i = 0; i < list.size(); i++) {
+ replicas[i] = StrUtils.splitSmart(list.get(i), "|", true);
+ // todo do we really not need to transform in non-cloud mode?!
+ // builder.replicaListTransformer.transform(replicas[i]);
+ builder.hostChecker.checkWhitelist(builder.shardsParam, replicas[i]);
+ }
+ }
+
+ @Override
+ public List<String> getSliceNames() {
+ // there are no logical slice names in non-cloud
+ return Collections.emptyList();
+ }
+
+ @Override
+ public int getSliceCount() {
+ return replicas.length;
+ }
+
+ @Override
+ public List<String> getReplicasBySlice(int sliceNumber) {
+ assert sliceNumber >= 0 && sliceNumber < replicas.length;
+ return replicas[sliceNumber];
+ }
+
+ static class Builder {
+ private String shardsParam;
+ private HttpShardHandlerFactory.WhitelistHostChecker hostChecker;
+
+ public Builder shards(String shardsParam) {
+ this.shardsParam = shardsParam;
+ return this;
+ }
+
+ public Builder whitelistHostChecker(HttpShardHandlerFactory.WhitelistHostChecker hostChecker) {
+ this.hostChecker = hostChecker;
+ return this;
+ }
+
+ public LegacyReplicaSource build() {
+ return new LegacyReplicaSource(this);
+ }
+ }
+}
diff --git a/solr/core/src/java/org/apache/solr/handler/component/ReplicaSource.java b/solr/core/src/java/org/apache/solr/handler/component/ReplicaSource.java
new file mode 100644
index 0000000..979e69b
--- /dev/null
+++ b/solr/core/src/java/org/apache/solr/handler/component/ReplicaSource.java
@@ -0,0 +1,42 @@
+/*
+ * 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.
+ */
+
+package org.apache.solr.handler.component;
+
+import java.util.List;
+
+/**
+ * A source of slices and corresponding replicas required to execute a request.
+ *
+ * @lucene.experimental
+ */
+interface ReplicaSource {
+ /**
+ * @return the list of slice names
+ */
+ List<String> getSliceNames();
+
+ /**
+ * Get the list of replica urls for a 0-indexed slice number.
+ */
+ List<String> getReplicasBySlice(int sliceNumber);
+
+ /**
+ * @return the count of slices
+ */
+ int getSliceCount();
+}