SolrCloud Distributed Requests

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 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

{
  "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 to requireZkConnected. This will cause requests to fail rather than setting a zkConnected header to false.

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.

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

{
  "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 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:

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:

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:

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:

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:

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:

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.

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 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:
```
shards.preference=replica.base:stable:hash:sessionId&sessionId=abc123
```
Prefer PULL replicas:
```
shards.preference=replica.type:PULL
```
Prefer PULL replicas, or TLOG replicas if PULL replicas are not available:
```
shards.preference=replica.type:PULL,replica.type:TLOG
```

Prefer any local replicas:

shards.preference=replica.location:local

Prefer any replicas on a host called "server1" with "server2" as the secondary option:
```
shards.preference=replica.location:http://server1,replica.location:http://server2
```
Prefer PULL replicas if available, otherwise TLOG replicas, and local replicas among those:
```
shards.preference=replica.type:PULL,replica.type:TLOG,replica.location:local
```
Prefer local replicas, and among them PULL replicas when available, otherwise TLOG replicas:
```
shards.preference=replica.location:local,replica.type:PULL,replica.type:TLOG
```
Prefer any replica that is not a leader:
```
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.

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 Document Routing for more information

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:

<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 allowUrls for details.

The HttpShardHandlerFactory accepts the following parameters:

socketTimeout

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

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

Optional

Default: 100000

The maximum number of concurrent connections that is made to each individual shard in a distributed search.

corePoolSize

Optional

Default: 0

The retained lowest limit on the number of threads used in coordinating distributed search.

maximumPoolSize

Optional

Default: Integer.MAX_VALUE

The maximum number of threads used for coordinating distributed search.

maxThreadIdleTime

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

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

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.

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:

<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.

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.