Routing a document to the right shard
When a document is indexed, Elasticsearch stores it on one primary shard. The question is: how does it decide which primary shard should hold that document? If an index has multiple shards, why does a document go to shard 1 rather than shard 2?
The choice cannot be random. If it were, Elasticsearch would have no reliable way to know where to look when the document is requested later. Instead, shard selection is determined by a routing calculation:
shard = hash(routing) % number_of_primary_shards
routing is a variable value. By default, it is the document’s _id, though it can also be set explicitly. Elasticsearch passes the routing value through a hash function, then divides the resulting number by number_of_primary_shards. The remainder is always in the range from 0 to number_of_primary_shards-1, and that remainder identifies the shard where the document belongs.
This is why the number of primary shards must be fixed when an index is created. If that number changed later, the same routing values would produce different remainders, invalidating the previous document-to-shard mapping. Elasticsearch would no longer know where existing documents should be found.
All document APIs—get, index, delete, bulk, update, and mget—accept a routing parameter. By supplying a custom routing value, you can control how documents are mapped to shards. A common use case is keeping related documents together, such as storing all documents that belong to the same user on the same shard.
How primary and replica shards work together
A request can be sent to any node in the cluster. Each node is capable of handling requests, and each node knows where documents are located across the cluster, so it can forward a request to the node that holds the required shard.
The node that receives the client request acts as the coordinating node for that request. In practice, it is usually better to distribute client requests across all nodes in the cluster, often by round-robin, so that request handling load is spread more evenly.
Creating, indexing, and deleting documents
Create, index, and delete requests are all write operations. A write must first be completed on the primary shard before the change is copied to the related replica shards.
By the time the client receives a successful response, the document change has been applied on the primary shard and on all replica shards involved in the operation. At that point, the change is considered safe.
Elasticsearch also provides optional request parameters that can influence this process. These options may improve performance in some situations, but they can do so at the cost of data safety. They are not commonly needed because Elasticsearch is already fast, but they are worth understanding.
consistency
consistency controls the shard availability requirement before a write operation is allowed to proceed. By default, even before attempting a write, the primary shard requires a specified number of shard copies to be active and available. This required number is a quorum—in other words, a majority of shard copies. A shard copy can be either the primary shard or a replica shard.
This rule helps prevent writes during a network partition, where parts of the cluster cannot communicate with each other and conflicting changes could lead to inconsistent data.
The quorum is calculated as follows:
int( (primary + number_of_replicas) / 2 ) + 1
The consistency parameter can be set to one, all, or quorum:
one: the write is allowed as long as the primary shard is available.all: the write is allowed only if the primary shard and all replica shards are available.quorum: the write is allowed if a majority of shard copies are available.
The default value is quorum.
One important detail is that number_of_replicas in the quorum formula refers to the configured number of replica shards in the index settings, not the number of replicas that happen to be active at the moment. For example, if an index is configured with three replica shards, the required number is:
int( (primary + 3 replicas) / 2 ) + 1 = 3
If only two nodes are running, the number of active shard copies cannot reach that requirement, so indexing and deleting documents will not be possible under that configuration.
timeout
What happens if there are not enough replica shards available? Elasticsearch waits, hoping that more shards will become available. By default, it waits for up to 1 minute.
If you want the request to fail sooner, you can use the timeout parameter. For example, 100 means 100 milliseconds, and 30s means 30 seconds.
A note on default replica settings
A new index has 1 replica shard by default. In theory, satisfying the quorum requirement would require two active shard copies. That would make a single-node setup unable to do anything useful. To avoid that problem, the quorum requirement is enforced only when number_of_replicas is greater than 1.
Retrieving a document
A document can be retrieved from the primary shard or from any of its replica shards.
For read requests, the coordinating node balances load by round-robining requests across all available shard copies. This allows replicas to help serve reads instead of leaving all read traffic on the primary shard.
There is a brief situation to be aware of: a document may already exist on the primary shard but not yet have been copied to a replica shard. If a read reaches that replica during this window, the replica may report that the document does not exist, while the primary shard could return it successfully. Once the indexing request has successfully returned to the user, however, the document is available on both the primary shard and the replica shards.
Partially updating a document
A partial document update follows a multi-step process:
- The client sends an update request to Node 1.
- Node 1 forwards the request to Node 3, where the primary shard is located.
- Node 3 retrieves the document from the primary shard, modifies the JSON in the
_sourcefield, and attempts to reindex the document on the primary shard. If another process has already modified the document, Node 3 retries this step. If the number of retries exceedsretry_on_conflict, it gives up. - After Node 3 successfully updates the document, it forwards the new version of the document in parallel to the replica shards on Node 1 and Node 2, where it is reindexed. Once all replica shards report success, Node 3 returns success to the coordinating node, and the coordinating node returns success to the client.
The update API also accepts the routing, replication, consistency, and timeout parameters.
Document-based replication
When the primary shard forwards a change to replica shards, it does not forward the update request itself. Instead, it sends the complete new version of the document.
This matters because changes are forwarded to replicas asynchronously, and Elasticsearch cannot guarantee that they will arrive in the same order in which they were sent. If Elasticsearch forwarded only the change operations, those operations might be applied in the wrong order on a replica, producing a corrupted document. Sending the full updated document version avoids that problem.