solr/solr-ref-guide/modules/deployment-guide/pages/solrcloud-distributed-requests.adoc

= SolrCloud Distributed Requests
// 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.

When a Solr node receives a search request, the request is automatically routed to a replica of a shard that is part of the collection being searched.

The chosen replica acts as an aggregator: it creates internal requests to randomly chosen replicas of every shard in the collection, coordinates the responses, issues any subsequent internal requests as needed (for example, to refine facets values, or request additional stored fields), and constructs the final response for the client.

== Query Fault Tolerance

In a SolrCloud cluster each individual node load balances read requests across all the replicas in a collection.
You may still need a load balancer on the 'outside' that talks to the cluster.
Or you need a client which understands how to read and interact with Solr's metadata in ZooKeeper and only requests the ZooKeeper ensemble's address to discover which nodes should receive requests.
Solr provides a smart Java SolrJ client called {solr-javadocs}/solrj/org/apache/solr/client/solrj/impl/CloudSolrClient.html[CloudSolrClient] which is capable of this.

Even if some nodes in the cluster are offline or unreachable, a Solr node will be able to correctly respond to a search request as long as it can communicate with at least one replica of every shard, or one replica of every _relevant_ shard if the user limited the search via the `shards` or `\_route_` parameters.
The more replicas there are of every shard, the more likely that the Solr cluster will be able to handle search results in the event of node failures.

=== zkConnected Parameter

A Solr node will return the results of a search request as long as it can communicate with at least one replica of every shard that it knows about, even if it can _not_ communicate with ZooKeeper at the time it receives the request.
This is normally the preferred behavior from a fault tolerance standpoint, but may result in stale or incorrect results if there have been major changes to the collection structure that the node has not been informed of via ZooKeeper (i.e., shards may have been added or removed, or split into sub-shards).

A `zkConnected` header is included in every search response indicating if the node that processed the request was connected with ZooKeeper at the time:

.Solr Response with zkConnected
[source,json]
----
{
  "responseHeader": {
    "status": 0,
    "zkConnected": true,
    "QTime": 20,
    "params": {
      "q": "*:*"
    }
  },
  "response": {
    "numFound": 107,
    "start": 0,
    "docs": [ "..." ]
  }
}
----

To prevent stale or incorrect results in the event that the request-serving node can't communicate with ZooKeeper, set the <<shards-tolerant-parameter,`shards.tolerant`>> parameter to `requireZkConnected`.
This will cause requests to fail rather than setting a `zkConnected` header to `false`.

[#shards-tolerant-parameter]
=== shards.tolerant Parameter

In the event that one or more shards queried are unavailable, then Solr's default behavior is to fail the request.
However, there are many use-cases where partial results are acceptable and so Solr provides a boolean `shards.tolerant` parameter (default `false`).
In addition to `true` and `false`, `shards.tolerant` may also be set to `requireZkConnected` - see below.

If `shards.tolerant=true` then partial results may be returned.
If the returned response does not contain results from all the appropriate shards then the response header contains a special flag called `partialResults`.

If `shards.tolerant=requireZkConnected` and the node serving the search request cannot communicate with ZooKeeper, the request will fail, rather than returning potentially stale or incorrect results.
This will also cause requests to fail when one or more queried shards are completely unavailable, just like when `shards.tolerant=false`.

// TODO: add section on shards.info with examples
The client can specify `shards.info` along with the `shards.tolerant` parameter to retrieve more fine-grained details.

Example response with `partialResults` flag set to `true`:

.Solr Response with partialResults
[source,json]
----
{
  "responseHeader": {
    "status": 0,
    "zkConnected": true,
    "partialResults": true,
    "QTime": 20,
    "params": {
      "q": "*:*"
    }
  },
  "response": {
    "numFound": 77,
    "start": 0,
    "docs": [ "..." ]
  }
}
----

=== distrib.singlePass Parameter

If set to `true`, the `distrib.singlePass` parameter changes the distributed search algorithm to fetch all requested stored fields from each shard in the first phase itself.
This eliminates the need for making a second request to fetch the stored fields.

This can be faster when requesting a very small number of fields containing small values.
However, if large fields are requested or if a lot of fields are requested then the overhead of fetching them over the network from all shards can make the request slower as compared to the normal distributed search path.

Note that this optimization only applies to distributed search.
Certain features such as faceting may make additional network requests for refinements, etc.

== Routing Queries

There are several ways to control how queries are routed.

=== Limiting Which Shards are Queried

While one of the advantages of using SolrCloud is the ability to query very large collections distributed across various shards, in some cases you may have configured Solr with specific xref:solrcloud-shards-indexing.adoc#document-routing[document routing].
You have the option of searching over all of your data or just parts of it.

Because SolrCloud automatically load balances queries, a query across all shards for a collection is simply a query that does not define a `shards` parameter:

[source,text]
----
http://localhost:8983/solr/gettingstarted/select?q=*:*
----

This is in contrast to user-managed clusters, where the `shards` parameter is required in order to distribute the query.

To limit the query to just one shard, use the `shards` parameter to specify the shard by its logical ID, as in:

[source,text]
----
http://localhost:8983/solr/gettingstarted/select?q=*:*&shards=shard1
----

If you want to search a group of shards, you can specify each shard separated by a comma in one request:

[source,text]
----
http://localhost:8983/solr/gettingstarted/select?q=*:*&shards=shard1,shard2
----

In both of the above examples, while only the specific shards are queried, any random replica of the shard will get the request.

You could instead specify a list of replicas you wish to use in place of a shard IDs by separating the replica IDs with commas:

[source,text]
----
http://localhost:8983/solr/gettingstarted/select?q=*:*&shards=localhost:7574/solr/gettingstarted,localhost:8983/solr/gettingstarted
----

Or you can specify a list of replicas to choose from for a single shard (for load balancing purposes) by using the pipe symbol (|) between different replica IDs:

[source,text]
----
http://localhost:8983/solr/gettingstarted/select?q=*:*&shards=localhost:7574/solr/gettingstarted|localhost:7500/solr/gettingstarted
----

Finally, you can specify a list of shards (separated by commas) each defined by a list of replicas (separated by pipes).

In the following example, 2 shards are queried, the first being a random replica from shard1, the second being a random replica from the explicit pipe delimited list:

[source,text]
----
http://localhost:8983/solr/gettingstarted/select?q=*:*&shards=shard1,localhost:7574/solr/gettingstarted|localhost:7500/solr/gettingstarted
----

=== shards.preference Parameter

Solr allows you to pass an optional string parameter named `shards.preference` to indicate that a distributed query should sort the available replicas in the given order of precedence within each shard.

The syntax is: `shards.preference=_property_:__value__`.
The order of the properties and the values are significant: the first one is the primary sort, the second is secondary, etc.

IMPORTANT: `shards.preference` is supported for single shard scenarios only when using the SolrJ clients.
Queries that do not use the SolrJ client cannot use `shards.preference` in single shard collections.

The properties that can be specified are as follows:

`replica.type`::
One or more replica types that are preferred.
Any combination of `PULL`, `TLOG` and `NRT` is allowed.

`replica.location`::
One or more replica locations that are preferred.
+
A location starts with `http://hostname:port`.
Matching is done for the given string as a prefix, so it's possible to e.g., leave out the port.
+
A special value `local` may be used to denote any local replica running on the same Solr instance as the one handling the query.
This is useful when a query requests many fields or large fields to be returned per document because it avoids moving large amounts of data over the network when it is available locally.
In addition, this feature can be useful for minimizing the impact of a problematic replica with degraded performance, as it reduces the likelihood that the degraded replica will be hit by other healthy replicas.
+
The value of `replica.location:local` diminishes as the number of shards (that have no locally-available replicas) in a collection increases because the query controller will have to direct the query to non-local replicas for most of the shards.
+
In other words, this feature is mostly useful for optimizing queries directed towards collections with a small number of shards and many replicas.
+
Also, this option should only be used if you are load balancing requests across all nodes that host replicas for the collection you are querying, as Solr's `CloudSolrClient` will do.
If not load-balancing, this feature can introduce a hotspot in the cluster since queries won't be evenly distributed across the cluster.

`replica.base`::
Applied after sorting by inherent replica attributes, this property defines a fallback ordering among sets of preference-equivalent replicas; if specified, only one value may be specified for this property, and it must be specified last.
+
`random`, the default, randomly shuffles replicas for each request.
This distributes requests evenly, but can result in sub-optimal cache usage for shards with replication factor > 1.
+
`stable:dividend:_paramName_` parses an integer from the value associated with the given parameter name; this integer is used as the dividend (mod equivalent replica count) to determine (via list rotation) order of preference among equivalent replicas.
+
`stable[:hash[:_paramName_]]` the string value associated with the given parameter name is hashed to a dividend that is used to determine replica preference order (analogous to the explicit `dividend` property above); `_paramName_` defaults to `q` if not specified, providing stable routing keyed to the string value of the "main query".
Note that this may be inappropriate for some use cases (e.g., static main queries that leverage parameter substitution)

`replica.leader`::
Prefer replicas based on their leader status, set to either `true` or `false`.
+
Consider a shard with two `TLOG` replicas and four `PULL` replicas (six replicas in total, one of which is the leader).
With `shards.preference=replica.leader:false`, 5 out of 6 replicas will be preferred.
Contrast this with `shards.preference=replica.type:PULL` where only 4 of 6 replicas will be preferred.
+
Note that the non-leader `TLOG` replica behaves like a `PULL` replica from a search perspective; it pulls index updates from the leader just like a `PULL` replica and does not perform soft-commits.
The difference is that the non-leader `TLOG` replica also captures updates in its TLOG, so that it is a candidate to replace the current leader if it is lost.

`node.sysprop`::
Query will be routed to nodes with same defined system properties as the current one.
For example, if you start Solr nodes on different racks, you'll want to identify those nodes by a xref:configuration-guide:property-substitution.adoc#jvm-system-properties[system property] (e.g., `-Drack=rack1`).
Then, queries can contain `shards.preference=node.sysprop:sysprop.rack`, to make sure you always hit shards with the same value of `rack`.

*Examples*:

* Prefer stable routing (keyed to client "sessionId" parameter) among otherwise equivalent replicas:
+
[source,text]
shards.preference=replica.base:stable:hash:sessionId&sessionId=abc123

* Prefer PULL replicas:
+
[source,text]
shards.preference=replica.type:PULL

* Prefer PULL replicas, or TLOG replicas if PULL replicas are not available:
+
[source,text]
shards.preference=replica.type:PULL,replica.type:TLOG

* Prefer any local replicas:
+
[source,text]
shards.preference=replica.location:local

* Prefer any replicas on a host called "server1" with "server2" as the secondary option:
+
[source,text]
shards.preference=replica.location:http://server1,replica.location:http://server2

* Prefer PULL replicas if available, otherwise TLOG replicas, and local replicas among those:
+
[source,text]
shards.preference=replica.type:PULL,replica.type:TLOG,replica.location:local

* Prefer local replicas, and among them PULL replicas when available, otherwise TLOG replicas:
+
[source,text]
shards.preference=replica.location:local,replica.type:PULL,replica.type:TLOG

* Prefer any replica that is not a leader:
+
[source,text]
shards.preference=replica.leader:false

Note that if you provide these parameters in a query string, they need to be properly URL-encoded.

=== collection Parameter

The `collection` parameter allows you to specify a collection or a number of collections on which the query should be executed.
This allows you to query multiple collections at once and the features of Solr which work in a distributed manner will work across collections.

[source,plain]
----
http://localhost:8983/solr/collection1/select?collection=collection1,collection2,collection3
----

=== \_route_ Parameter

The `\_route_` parameter can be used to specify a route key which is used to figure out the corresponding shards.
For example, if you have a document with a unique key "user1!123", then specifying the route key as "_route_=user1!" (notice the trailing '!' character) will route the request to the shard which hosts that user.
You can specify multiple route keys separated by comma.
This parameter can be leveraged when we have shard data by users.
See xref:solrcloud-shards-indexing.adoc#document-routing[Document Routing] for more information

[source,plain]
----
http://localhost:8983/solr/collection1/select?q=*:*&_route_=user1!
http://localhost:8983/solr/collection1/select?q=*:*&_route_=user1!,user2!
----

== Near Real Time (NRT) Use Cases

Near Real Time (NRT) search means that documents are available for search soon after being indexed.
NRT searching is one of the main features of SolrCloud and is rarely attempted in user-managed clusters or single-node installations.

Document durability and searchability are controlled by `commits`.
The "Near" in "Near Real Time" is configurable to meet the needs of your application.
Commits are either "hard" or "soft" and can be issued by a client (say SolrJ), via a REST call or configured to occur automatically in `solrconfig.xml`.
The recommendation usually gives is to configure your commit strategy in `solrconfig.xml` (see below) and avoid issuing commits externally.

Typically in NRT applications, hard commits are configured with `openSearcher=false`, and soft commits are configured to make documents visible for search.

When a commit occurs, various background tasks are initiated, segment merging for example.
These background tasks do not block additional updates to the index nor do they delay the availability of the documents for search.

When configuring for NRT, pay special attention to cache and autowarm settings as they can have a significant impact on NRT performance.
For extremely short autoCommit intervals, consider disabling caching and autowarming completely.

== Configuring the ShardHandlerFactory

For finer-grained control, you can directly configure and tune aspects of the concurrency and thread-pooling used within distributed search in Solr.
The default configuration favors throughput over latency.

This is done by defining a `shardHandlerFactory` in the configuration for your search handler.

To add a `shardHandlerFactory` to the standard search handler, provide a configuration in `solrconfig.xml`, as in this example:

[source,xml]
----
<requestHandler name="/select" class="solr.SearchHandler">
  <!-- other params go here -->
  <shardHandlerFactory class="HttpShardHandlerFactory">
    <int name="socketTimeout">1000</int>
    <int name="connTimeout">5000</int>
  </shardHandlerFactory>
</requestHandler>
----

`HttpShardHandlerFactory` is the only `ShardHandlerFactory` implementation included out of the box with Solr.

NOTE:: The `shardHandlerFactory` is reliant on the `allowUrls` parameter configured in `solr.xml`, which controls which nodes are allowed to talk to each other.
This means that the configuration of hosts is global instead of per-core or per-collection.
See the section xref:configuration-guide:configuring-solr-xml.adoc#allow-urls [allowUrls] for details.

The `HttpShardHandlerFactory` accepts the following parameters:

`socketTimeout`::
+
[%autowidth,frame=none]
|===
|Optional |Default: `0`
|===
+
The amount of time in milliseconds that a socket is allowed to wait.
The default is `0`, where the operating system's default will be used.

`connTimeout`::
+
[%autowidth,frame=none]
|===
|Optional |Default: `0`
|===
+
The amount of time in milliseconds that is accepted for binding / connecting a socket.
The default is `0`, where the operating system's default will be used.

`maxConnectionsPerHost`::
+
[%autowidth,frame=none]
|===
|Optional |Default: `100000`
|===
+
The maximum number of concurrent connections that is made to each individual shard in a distributed search.

`corePoolSize`::
+
[%autowidth,frame=none]
|===
|Optional |Default: `0`
|===
+
The retained lowest limit on the number of threads used in coordinating distributed search.

`maximumPoolSize`::
+
[%autowidth,frame=none]
|===
|Optional |Default: `Integer.MAX_VALUE`
|===
+
The maximum number of threads used for coordinating distributed search.

`maxThreadIdleTime`::
+
[%autowidth,frame=none]
|===
|Optional |Default: `5`
|===
+
The amount of time in seconds to wait for before threads are scaled back in response to a reduction in load.

`sizeOfQueue`::
+
[%autowidth,frame=none]
|===
|Optional |Default: `-1`
|===
+
If specified, the thread pool will use a backing queue instead of a direct handoff buffer.
High throughput systems will want to configure this to be a direct hand off (with `-1`).
Systems that desire better latency will want to configure a reasonable size of queue to handle variations in requests.

`fairnessPolicy`::
+
[%autowidth,frame=none]
|===
|Optional |Default: `false`
|===
+
Chooses the JVM specifics dealing with fair policy queuing.
If enabled distributed searches will be handled in a first-in-first-out fashion at a cost to throughput.
If disabled throughput will be favored over latency.


[[distributedidf]]
== Distributed Inverse Document Frequency (IDF)

Document and term statistics are needed in order to calculate relevancy.
In a distributed system, these statistics can vary from node to node, introducing bias or inaccuracies into scoring calculations.

Solr stores the document and term statistics in a cache called the `statsCache`.
There are four implementations out of the box when it comes to document statistics calculation:

* `LocalStatsCache`: This uses only local term and document statistics to compute relevance.
In cases with uniform term distribution across shards, this works reasonably well.
This option is the default if no `<statsCache>` is configured.

* `ExactStatsCache`: This implementation uses global values (across the collection) for document frequency.
It's recommended to choose this option if precise scoring across nodes is important for your implementation.

* `ExactSharedStatsCache`: This is like the `ExactStatsCache` in its functionality but the global stats are reused for subsequent requests with the same terms.

* `LRUStatsCache`: This implementation uses a least-recently-used cache to hold global stats, which are shared between requests.

The implementation can be selected by setting `<statsCache>` in `solrconfig.xml`.
For example, the following line makes Solr use the `ExactStatsCache` implementation:

[source,xml]
----
<statsCache class="org.apache.solr.search.stats.ExactStatsCache"/>
----

=== distrib.statsCache Parameter

The query param distrib.statsCache defaults to `true`. If set to `false`, distributed calls to fetch global term stats is turned off for this query. This can reduce overhead for queries that do not utilize distributed IDF for score calculation.

[source,xml]
----
http://localhost:8987/solr/collection1/select?q=*%3A*&wt=json&fq={!terms f=id}id1,id2&distrib.statsCache=false
----



== Avoiding Distributed Deadlock

Each shard serves top-level query requests and then makes sub-requests to all of the other shards.
Care should be taken to ensure that the max number of threads serving HTTP requests is greater than the possible number of requests from both top-level clients and other shards.
If this is not the case, the configuration may result in a distributed deadlock.

For example, a deadlock might occur in the case of two shards, each with just a single thread to service HTTP requests.
Both threads could receive a top-level request concurrently, and make sub-requests to each other.
Because there are no more remaining threads to service requests, the incoming requests will be blocked until the other pending requests are finished, but they will not finish since they are waiting for the sub-requests.
By ensuring that Solr is configured to handle a sufficient number of threads, you can avoid deadlock situations like this.

== Distributed Tracing and Debugging

The `debug` parameter with a value of `track` can be used to trace the request as well as find timing information for each phase of a distributed request.