Three plain ideas, put together: copies, order, and trust. Once each is clear, the rest follows naturally.

Elasticsearch stores documents in indexes, and every index is cut into pieces called shards. Each shard is a small, complete search engine of its own. To protect the data, and to spread out reading work, Elasticsearch keeps more than one copy of each shard. The main copy is called the primary. The extra copies are called replicas. All the copies of one shard together are called its replication group, and they live on different nodes, so that the loss of one machine does not lose the data.

Copies are only useful if they agree. If the primary has a document and the replica does not, then a search that happens to hit the replica will miss it, and if the primary then fails, the document is gone. Synchronization is the set of rules and mechanisms that keep the copies identical: how a change is sent from the primary to the replicas, in what order, how the replicas confirm it, how a replica that missed changes catches up, and how the cluster decides which copies can be trusted.

The third idea is trust. It is not enough to copy changes. The cluster must also know which copies are fully up to date, because that knowledge decides what is safe to do when something breaks. A copy that is known to hold every acknowledged change may be promoted to primary. A copy that missed changes must catch up first. Elasticsearch keeps this knowledge in the cluster state, in a list called the in-sync set.

What it is not

Three things sound similar but are different. First, a snapshot is a backup stored in an outside repository at a point in time. It is not a live copy. Second, cross-cluster replication is a separate feature that copies indexes from one cluster to another, asynchronously, and is normally a paid feature. Third, reindexing copies documents from one index into another by reading and writing them again. Primary-replica synchronization is none of these. It is the continuous, automatic, per-shard copying that happens inside one index, within one cluster, on every single write.

The one-sentence version

If you remember one line from this chapter, make it this: primary-replica synchronization is the process by which the primary shard assigns every write a sequence number, forwards it to all in-sync replicas, waits for their confirmation, and tells the master to drop any copy that fails, so that every acknowledged write is present on every trusted copy.

Every design has a problem behind it. Here the problem is that hardware breaks and users still expect their data to be there.

In a cluster with dozens or hundreds of machines, something is always failing somewhere: a disk, a power supply, a network cable, a software bug, a planned restart. If each shard existed in only one place, every failure would make part of the data unavailable, and some failures would destroy it. Replication answers this with three benefits at once.

• Availability. If the machine holding a primary fails, a replica on another machine can take over within seconds, and users keep reading and writing.
• Durability. An acknowledged write exists on more than one machine. Losing any single machine does not lose it.
• Read capacity. Replicas can answer searches and lookups, so that read traffic is spread across more machines, and popular data does not overload a single node.

But copying has its own difficulties, and these are what synchronization is about. Copies can receive changes at different times. Network messages can arrive out of order. A replica can be slow, or disconnected, or restarted in the middle of a change. A primary can crash halfway through sending a change, so that some replicas have it and others do not. Each difficulty needs a clear rule, or the copies would slowly drift apart, and nobody would know which was right.

A short history

The exact release for each step is worth double-checking in the release notes of your version. The direction of travel is what matters: from “copy and hope” toward “number every operation and prove every copy.”

Why not simply copy files?

One could imagine replicas that just copy the primary’s files every few seconds. Some systems work that way, and it has real advantages, such as saving the replica from repeating the indexing work. But file copying makes it harder to acknowledge a write as durable on several machines at once, and it delays the moment a write appears on replicas. The standard Elasticsearch model instead sends each operation to the replicas, which repeat the indexing work and keep their own transaction logs. That choice gives strong per-write guarantees, at the price of extra CPU on every copy.

A dozen words explain nearly everything about synchronization. Learn them here, and later chapters will read like common sense.

Who decides what

Three parties share the work. The primary decides the order of operations and their outcome, for example whether a conflicting update is accepted. The replicas obey: they repeat what the primary decided. The master node keeps the official list of which copies exist, where they are, and which of them are in sync. The primary may propose changes to that list, such as “drop this replica,” but only the master can make them official.

Why order matters so much

Suppose two operations touch the same document: first “set price to 10,” then “set price to 12.” If a replica applied them in the wrong order, it would end with price 10, while the primary has 12. The copies would disagree for as long as nobody touched that document again. Network messages can arrive out of order, so the arrival order cannot be trusted. Sequence numbers solve this. Each operation carries its number, and the replica uses it to decide which of two conflicting operations is newer, no matter which arrived first.

Why “in sync” is stronger than “alive”

A replica can be alive and yet not in sync. It might have been offline for ten minutes, missing thousands of operations. It is running, and it holds old data, but it is not trustworthy. Elasticsearch therefore separates the two ideas. A copy joins the in-sync set only when it has caught up, and it is removed from that set the moment it is known to have missed something. Reads may be served only by started copies, and promotions use only the in-sync ones.

Quick glossary

Where do the copies live, who talks to whom, and which parts of the cluster take part in keeping them together?

Every shard has a replication group spread over different nodes. The rules of placement guarantee that two copies of the same shard never live on the same node, because one crash would then destroy both. When zone awareness is configured, copies are also spread across zones or racks. The master node keeps the list of copies and marks which are in sync. The primary talks directly to its replicas over the network to forward operations.

flowchart TB
    Client["Client sends a write request"] --> Coord["Coordinating node: finds the shard using the document id or routing value"]
    Coord --> P["Primary shard copy on Node A"]
    subgraph Group["Replication group of one shard"]
        P
        R1["Replica copy on Node B"]
        R2["Replica copy on Node C"]
    end
    P -->|forwards each operation| R1
    P -->|forwards each operation| R2
    M["Master node: cluster state with the in-sync copy list"] -.->|tracks which copies are in sync| Group
    

Read the picture from top to bottom. A client sends a write to any node, which acts as coordinating node. It works out which shard owns the document and sends the operation to the node that holds that shard’s primary. The primary executes it and forwards it to the replicas on other nodes. The dotted line shows that the master does not carry the data. It only keeps the official record of who is in the group and who is in sync.

The main components

One synchronization per shard

An index with five primary shards and one replica each has five separate replication groups. Each is synchronized on its own. A slow replica in one group does not slow the others. This is why the cluster can scale writes by adding shards: each new group adds capacity, and no group needs to talk to any other.

Walking one write through the machine

Now we open the box and follow a single write from the moment it arrives to the moment it is acknowledged.

flowchart TD
    A["Index or bulk request arrives at the coordinating node"] --> B["Routed to the primary shard"]
    B --> C["Primary validates the operation and checks versions"]
    C --> D["Primary assigns sequence number and primary term"]
    D --> E["Primary indexes locally and writes its translog"]
    E --> F["Primary sends the operation to all in-sync replicas in parallel"]
    F --> G["Each replica indexes and writes its own translog"]
    G --> H{"Did every in-sync replica confirm?"}
    H -- Yes --> I["Primary acknowledges success to the client"]
    H -- No --> J["Primary asks the master to drop the failed copy from the in-sync set"]
    J --> I
    

Step one: routing to the right shard

The coordinating node decides which shard owns a document by taking a hash of the routing value, which is by default the document’s identifier, and reducing it to a shard number using the count of primary shards. Because this depends on the number of primary shards, that number cannot be changed on an existing index without rebuilding it. The same identifier always leads to the same shard, so updates and deletes find the right home.

Step two: the primary validates

The primary is the only copy that makes decisions. It checks that the operation is well formed. If the operation carries a condition, such as “only update if the document is still at version 7,” it checks that condition against the current state of the document. If the check fails, the operation is rejected here, and no replica ever sees it. If it passes, the primary decides the outcome, for example whether the operation creates a new document or replaces an old one.

Step three: numbering

The primary assigns the operation a sequence number, the next number in an unbroken series, and stamps it with the current primary term. From this moment, the operation is a fixed piece of history: replicas will apply it as it is, with this number, and no replica may renumber it.

Step four: local execution

The primary indexes the document into its in-memory buffer and appends the operation to its translog. By default, the translog is synced to disk before the request finishes, so that a crash cannot lose an operation that has been acknowledged. The primary does this before, or while, it forwards the operation, and in the standard flow it does not wait to hear from replicas before doing its own work.

Step five: forwarding in parallel

The primary sends the operation to every copy in the group that is either in sync or currently recovering, all at the same time, not one after the other. For a bulk request, it sends the whole batch destined for that shard as one message, already carrying the sequence numbers. Sending in parallel means that the time cost of having three replicas is not three times the time cost of having one. It is closer to the time of the slowest one.

Step six: replicas execute

Each replica receives the operation with its sequence number and primary term. It first checks the term: if the term is older than the newest term the replica knows about, the operation comes from an old primary and is refused. Otherwise it applies the operation to its own index and translog. If operations arrive out of order, the replica uses the sequence numbers to make sure that an older operation never overwrites a newer one.

Step seven: acknowledgment

When all in-sync replicas have confirmed, the primary responds to the client. The response contains a small summary of the shards involved: how many copies were involved, how many succeeded, and how many failed. If a replica failed, the primary has already reported this to the master, which removes that copy from the in-sync set before the acknowledgment is sent.

What the write does not wait for

Acknowledgment does not mean that the document is yet visible to searches. Visibility depends on the refresh, which happens separately on every copy, on its own schedule. A document can be safely stored on all copies and still not appear in search results for up to a refresh interval. The acknowledgment is a promise about safety, not about search visibility. Chapter 11 explores the consequences.

These three ideas are the bookkeeping that makes everything else safe. Once you understand them, failures stop feeling mysterious.

Sequence numbers: a fixed order for history

Every operation that changes a shard, whether it adds, replaces, or deletes a document, receives a sequence number from the primary. The numbers form an unbroken series: 0, 1, 2, 3, and so on. Once assigned, a number is never changed and never reused within the same shard. This gives each shard a single, agreed history, like a numbered list of everything that ever happened.

Primary terms: which leader gave the order

Each time a new primary is chosen for a shard, a counter called the primary term goes up by one. Every operation carries the term of the primary that issued it. Replicas remember the highest term they have seen, and refuse any command carrying a lower one. This is how the group protects itself from a zombie primary: an old primary that was cut off by a network problem, has been replaced, and does not yet know it. Its commands carry an old term, and every replica that has heard about the new primary ignores them.

Local checkpoints: how far each copy has got

Each copy tracks its own local checkpoint. This is the highest sequence number such that the copy has processed every operation from the beginning up to that number, with no gaps. If a copy has processed operations 0 to 97 and 99 but not 98, its local checkpoint is 97, not 99, because a gap is still open. The local checkpoint moves forward only as gaps are filled.

The global checkpoint: what is safe everywhere

The primary collects the local checkpoints of all in-sync copies. The lowest of them is the global checkpoint. It is the number up to which every in-sync copy is known to have processed every operation. Everything at or below the global checkpoint is safe: it exists on all trusted copies, and it will never be rolled back.

flowchart LR
    P["Primary: local checkpoint 105"] --> G(("Lowest local checkpoint of all in-sync copies"))
    R1["Replica 1: local checkpoint 103"] --> G
    R2["Replica 2: local checkpoint 104"] --> G
    G --> V["Global checkpoint 103: all operations up to 103 are safe on every in-sync copy"]
    

In the picture, the primary has processed everything up to operation 105, one replica up to 103, and another up to 104. The global checkpoint is 103. Operations 104 and 105 are probably on most copies, but not yet proven to be on all of them, so they are not yet counted as globally safe.

How checkpoints travel

Replicas tell the primary their local checkpoint in every confirmation. The primary recomputes the global checkpoint and sends it back to the replicas, attached to later replication messages. When no writes are flowing, a background task sends the latest global checkpoint to replicas at a regular interval, thirty seconds by default, so that quiet shards still learn the news. The global checkpoint is therefore always known to be safe on a copy, but it might lag slightly behind the primary’s own view.

What checkpoints are used for

Handling out-of-order arrival

Networks do not promise order. A replica might receive operation 51 before operation 50. If both touch the same document, the replica must not let the older operation overwrite the newer one. It compares sequence numbers: when an older operation arrives after a newer one for the same document, the replica recognizes it as outdated. It records it in its history so that its checkpoint can advance, but it does not let it replace the newer state of the document. This gives every replica the same final result, whatever order the messages arrive in.

Gaps and no-ops

Sometimes a sequence number is assigned to an operation that then fails on the primary, for example because of a problem discovered after numbering. If nothing were done, that number would remain a permanent gap, and checkpoints could never move past it. To avoid this, the primary sends a special operation called a no-op for that number. It changes nothing, but it fills the gap, so that the series of numbers stays unbroken and checkpoints can advance.

A replica can fail in many ways: it can crash, stall, lose its network, or hit an error. The rule that governs them all is short and strict.

The rule

A write may be acknowledged only if every copy in the in-sync set has confirmed it, or has been removed from the in-sync set. There is no third option. A copy that could not confirm is never silently left in the set. This single rule means that when the client hears “success,” any copy still in the set has the write, and any copy without the write has been officially marked as untrustworthy.

flowchart TD
    A["Replica fails to apply an operation or stops responding"] --> B["Primary reports the failed copy to the master"]
    B --> C{"Does the master accept the report for this primary term?"}
    C -- Yes --> D["Master removes the copy from the in-sync set in the cluster state"]
    D --> E["Primary may now acknowledge the write"]
    E --> F["Master arranges for a new replica to be built"]
    C -- No --> G["Primary is stale: it steps down and does not acknowledge"]
    

Step by step through a replica failure

Why the master must be involved

The primary could simply stop sending operations to a failed replica. But the cluster, and above all the master, would still think that copy was current, and might later promote it, or route reads to it, as if nothing had happened. Making the removal official through the master closes this gap. It also serves as a safety check on the primary itself: if the master rejects the primary’s report because the term is outdated, the primary knows that it is a zombie and must not acknowledge anything.

What if the primary cannot reach the master?

Suppose a network problem cuts the primary off from the master, while some replicas are on the other side. The primary cannot make its report, so it cannot complete any write that involves a failed replica, and it will not acknowledge. Meanwhile, the master, on the healthy side, may already have chosen a new primary from the replicas it can reach, with a higher term. When the network heals, the old primary learns of the higher term and steps down. No acknowledged write was lost, because none was acknowledged in isolation.

flowchart LR
    subgraph Isolated["Cut-off side"]
        OP["Old primary: cannot reach the master"]
    end
    subgraph Healthy["Connected side"]
        M["Master"]
        NR["Replica promoted to new primary with a higher term"]
    end
    OP -. "network partition" .- M
    M --> NR
    OP --> X["Old primary cannot acknowledge writes and later steps down"]
    

What the client sees

The write response carries a small summary: the total number of shard copies involved, the number that succeeded, and the number that failed. Normally, the successful count equals the total. If a replica was dropped, the numbers differ, and the client can see that the write, though acknowledged, was made with reduced protection. Well-built monitoring watches for such answers, since they are an early sign of trouble.

How many copies must be available to start a write

By default, a write proceeds as long as the primary is active. A setting, which can be given per request or per index, lets you insist that a certain number of copies be active before the write even begins. For instance, requiring two active copies prevents writes from being accepted when only the primary is alive. This is a check made before the write starts, not a guarantee made after it. A copy that was active at the start can still fail during the write, and the in-sync rule then handles it as described.

When all replicas are gone

If every replica has been dropped, the primary continues alone, and the write is safe only on one machine. The cluster health shows yellow. The data is available but exposed to the failure of that one node. This is the moment when speed of repair matters, and why alerts on yellow status that lasts too long are so important.

The hardest day for a replication group is the day the primary dies. Here is exactly how the group stays correct.

flowchart LR
    F1["Primary node fails or is cut off"] --> F2["Master detects the loss"]
    F2 --> F3["Master picks an in-sync replica as the new primary"]
    F3 --> F4["Primary term increases by one"]
    F4 --> F5["New primary resyncs the other replicas above the global checkpoint"]
    F5 --> F6["Master rebuilds the missing replica copies"]
    F6 --> F7["The group is fully in sync again"]
    

Step one: detection

The master finds out that the node holding the primary is gone, through the regular health checks between nodes. It removes the node from the cluster state. Until a new primary is chosen, that shard cannot accept writes, though replicas can still serve reads.

Step two: choosing the successor

The master looks at the in-sync set and picks one of its members as the new primary. It must be in the set, because only those copies are known to hold every acknowledged write. The master also raises the primary term by one, and records both in the cluster state. Then it publishes the new state, and the chosen replica learns that it is now the primary.

Step three: the new primary starts work

The new primary knows the global checkpoint, which says that everything up to that number is safe on all in-sync copies. But above that number, things are uncertain. Operations that were in flight when the old primary died might have reached some replicas and not others. Some of them were never acknowledged, so the client may have received an error or a timeout. The other replicas might hold a different set of these uncertain operations than the new primary does.

Step four: the resync

To align everybody, the new primary performs a primary-replica resync. It takes the operations it holds above the global checkpoint, from its translog, and sends them to all the other replicas, stamped with the new term. It also sends no-ops to fill any gaps in the numbering. The replicas apply what they lack, and they discard any unconfirmed operations from the old term that do not match the new primary’s history. After the resync, all in-sync copies share exactly the same history, up to the new primary’s latest operation.

What happens to writes that were in flight

A write that was in flight when the primary died has an unknown outcome. It may have been applied on the new primary, in which case the resync will spread it to every copy, and it will survive. It may not have reached any surviving copy, in which case it is lost, and the client received no acknowledgment. What can never happen is that a write is acknowledged and then lost, because acknowledgment required all in-sync copies. This is why clients that retry after a timeout should expect that a write might be applied twice, and should use identifiers that make repeating them harmless.

Step five: rebuilding the missing copy

With the old primary gone, the group has one copy fewer than planned. The allocation service schedules a new replica. If the failed node is expected to return soon, the master waits, by default about a minute, before starting to build a replacement elsewhere. This avoids wasting effort on copying large amounts of data when the node was only restarting. If the node returns in time, its data is reused, and only the missing operations are replayed.

When no in-sync copy survives

If every copy in the in-sync set is lost, the cluster does not promote a stale copy. It leaves the shard unassigned, and the health turns red, because a stale copy might be missing acknowledged writes. An operator can, as an explicit and dangerous decision, tell the cluster to use a stale copy and accept the data loss. That decision should be made with full understanding, and ideally after trying to bring the lost nodes back, because they may hold the missing writes.

The whole sequence in a table

A replica that was away, or is brand new, must catch up before it can rejoin the trusted set. The process is called peer recovery.

When recovery happens

• A node restarts and its replicas must catch up on what they missed.
• A new replica is created, for example when you increase the number of replicas.
• A shard is moved to another node for balance, maintenance, or disk space.
• A replica was dropped from the in-sync set after a failure and now returns.

flowchart TD
    S["Replica needs recovery"] --> Q{"Does the primary still hold history from the replica's last safe point?"}
    Q -- Yes --> O["Operation-based recovery: replay only the missing operations"]
    Q -- No --> F["File-based recovery: copy missing segment files first"]
    F --> O2["Then replay the operations that arrived during the copy"]
    O --> D["Replica catches up and is marked in sync"]
    O2 --> D
    

Operation-based recovery: the fast path

If the replica was only away briefly, its data is still on disk, and it knows its own local checkpoint. The primary still holds the history of operations after that point. Then the recovery only needs to send those missing operations, in order, and the replica replays them. This is usually fast, because the amount of data is small: only what changed during the absence.

File-based recovery: the slow path

If the primary no longer holds the needed history, or the replica has no usable data at all, the recovery must first copy files. The primary compares the segment files that the target already has with its own, using checksums. Files that are identical are reused. Missing ones are sent over the network. Once the files are in place, the primary replays operations that arrived during the copy, so that the replica ends up fully current. File-based recovery can take a long time for large shards, and it uses a lot of network and disk.

The phases in order

Writes keep flowing during recovery

Recovery does not pause writes. While the target catches up, the primary keeps accepting new operations and also forwards them to the recovering copy, which applies them in the right places by sequence number. The recovering copy is not yet in the in-sync set, so acknowledgments do not wait for its confirmation to be counted in the rules. Only once it has caught up is it added, and from then on writes wait for it like any other member.

Recovery from the local disk

When a node restarts, its own shards first recover from local files. Each replays its translog up to the point that is known to be safe, using the global checkpoint as a guide, and opens from a commit that is not beyond it. Only then does it ask the primary for any operations that it missed while it was down. This two-step sequence, local first and peer second, is why a rolling restart of a healthy cluster is usually quick.

Limits and pacing

Recovery competes with normal traffic for disk, network, and CPU. To protect the cluster, there are limits: the number of recoveries a node may run at once, and the bandwidth that recoveries may use. Modern versions can also adapt the bandwidth limit to the hardware. Raising the limits makes recovery faster but can slow searching and indexing during the process. Lowering them protects live traffic but leaves the cluster exposed for longer. Balance is a decision to make deliberately, considering how long you can live with reduced protection.

Fast recovery depends on one thing: keeping just enough recent history. Here is how it is kept and how much it costs.

Every copy has its own translog

Each copy, primary or replica, appends every operation it applies to its own translog before acknowledging it. The translog is a simple, ordered file on disk. If the node crashes, it can replay the translog to recover operations that were not yet saved into committed segments. Because each copy keeps its own, a replica can recover on its own after a crash, and does not need the primary for that.

Durability of the translog

By default, the translog is synced to disk before each request finishes, on the primary and on every replica. That is why an acknowledged write survives even a simultaneous power failure of all nodes: it has been written to disk on every in-sync copy. A setting can relax this, syncing every few seconds instead, which speeds up writes but risks losing the last few seconds of operations on a copy that crashes. For data that cannot be recreated, keep the default.

The problem of history

The translog alone is not enough for long absences. It is trimmed regularly, once operations are safely saved in committed segments and are below the global checkpoint. If a replica is away for an hour, the primary might already have discarded the old translog parts. Something else must preserve history for the replica. That something is a pair of features working together.

Soft deletes

Normally, when a document is updated or deleted, Lucene marks the old version as deleted and it disappears from searches, until a merge removes it for good. With soft deletes, the old version is not fully removed. It is kept in the index, marked in a way that hides it from searches but preserves it as a record of what happened. This turns the index itself into a store of history. A recovering replica can read from this history to find the operations it missed.

Retention leases

But history costs disk space, so it cannot be kept forever. A retention lease is a promise made by the primary: “I will keep history at least back to sequence number N, for this copy, for a limited time.” Each copy has a lease, which moves forward as the copy catches up. The primary keeps the history required by the oldest lease. If a lease expires, because its copy has been away too long, the history it protected can be dropped, and that copy must later recover by copying files instead. The default lease period is twelve hours.

A numbers walkthrough

Let us use invented numbers. Suppose a shard receives 2,000 operations per second, and a node hosting one of its replicas is down for a ten-minute restart. That is 1,200,000 operations missed. With history retained, the recovery replays these operations, which may amount to a few hundred megabytes and takes a short time. Without history, the recovery would copy the shard’s files, which might be tens of gigabytes, and take many times longer. The retained history acts as a cheap insurance against long copy operations.

Now suppose the same node is down for two days, longer than the lease period. The lease has expired, and the history has been released. The replica returns, finds that the primary cannot supply the missing operations, and goes through file-based recovery. Planned maintenance that takes longer than the lease period will therefore incur the higher cost, and it is worth knowing that before starting.

Older versions

Before soft deletes became the standard, Elasticsearch kept extra translog files for a while, controlled by retention settings on the translog. Those settings were later deprecated in favor of soft deletes and leases, which use disk space more efficiently and are tied to what copies actually need. If you upgrade an old cluster, check the release notes for the changes in these settings.

Copies that agree about history can still show slightly different pictures to a searcher. Knowing exactly what is and is not guaranteed prevents many confusing bugs.

Who answers a read

Read requests can be served by any started copy of a shard, primary or replica. By default, the coordinating node spreads requests among the copies, and modern versions use adaptive replica selection: the node prefers copies that have recently answered quickly and have short queues. This spreads load well, and it also routes around a struggling copy.

Refresh is per copy

Here is the key subtlety. Synchronization guarantees that every in-sync copy has processed every acknowledged operation. It does not guarantee that every copy has refreshed at the same moment. Each copy refreshes on its own timer, by default once a second, and only after a refresh does a document become visible to searches on that copy. So right after a write, the primary may have refreshed and the replica not yet, and the same search may return different results depending on which copy answers.

The bouncing results problem

A user runs the same search twice, and the second time an item that was present disappears, or one that was missing appears. This happens when the two requests are served by different copies that are at different points of their refresh cycle. It is harmless for most applications, but confusing for a user who sees the result list change without doing anything. The cure is to make repeated requests from the same user go to the same copies, by giving the search a stable preference value, such as a session identifier. The same copies then answer each time, and results change only when the data does.

Real-time lookups

Fetching a document by its identifier is different. This lookup is real-time: it can see a write even before the next refresh, by consulting recent operations directly. Because an acknowledged write is present on all in-sync copies, a real-time lookup returns the newest version from any in-sync copy. That makes lookups by identifier the right tool when you must read your own write right away.

Waiting for visibility

Write requests offer options to deal with search visibility. One option forces a refresh as part of the request, which makes the change visible at once but is expensive, especially in bulk. Another makes the request wait until the normal refresh has made the change visible. Both have costs in speed, so use them sparingly, for example in tests or in user actions that truly need instant visibility.

Where this sits in the bigger picture

People sometimes ask which of the classic trade-offs Elasticsearch chooses. For writes inside one replication group, it favors consistency: it will not acknowledge a write that is not safe on all trusted copies, and it will pause writes rather than let copies diverge when a partition prevents the primary from confirming with the master. For searches, it favors availability and speed: any started copy can answer, and small differences in visibility are accepted. Understanding this split explains most of the behavior you will observe.

Synchronization has a small number of knobs. Each one trades speed, safety, or cost.

The defaults above are documented values for recent versions and may differ in yours, so check the documentation of the version you run.

Choosing the number of replicas

Changing the number of replicas

Raising the count creates new copies, which must be built by recovery from the primary. This uses network and disk, so do it during quiet periods and one index at a time for large indexes. Lowering the count simply removes copies, which is immediate. Neither change requires downtime. The change is a cluster state update, and takes effect as the allocation service carries it out.

Relaxing durability

Setting the translog to sync only every few seconds makes writes faster, because they no longer wait for a disk sync on every copy. The price is that a copy that crashes may lose the last few seconds of operations that it had acknowledged. The other copies may still hold them, and the group can often repair itself, but a simultaneous failure of several copies could lose data. Use it only where losing a few seconds is acceptable, or where the source data can be replayed.

Zone awareness

Telling the cluster which zone each node lives in lets the allocation rules keep copies of a shard in different zones. Then the loss of a whole zone leaves at least one copy alive. To make this work, you need enough nodes in each zone to hold the copies, and the number of copies should match the number of zones or divide evenly across them.

Delayed rebuilding

The short delay before rebuilding replicas of a departed node avoids wasteful copying when a node just restarts. For fleets where nodes commonly take longer to return, such as during a rolling upgrade of large machines, raising the delay may save effort. For fleets where nodes rarely return, lowering it restores protection sooner. The best value depends on how long your typical restart takes.

Replication multiplies work. Understanding the multiplication tells you where speed is won and lost.

Write amplification

In the standard model, every replica repeats the whole indexing job. With one replica, the cluster does the analysis, indexing, and translog work twice for each document. With two replicas, three times. Storage grows in the same way. The network carries each operation from the primary to each replica. So the total resource cost of a write is roughly the cost of one copy multiplied by the number of copies.

A numbers walkthrough

Let us use invented numbers. Suppose an application sends 100 megabytes per second of documents into an index with one replica. The cluster receives 100 megabytes per second at the front, but the primaries then send another 100 megabytes per second to the replicas, so 200 megabytes per second are indexed in total, and 100 megabytes per second cross the internal network between nodes. With two replicas, the total indexed becomes 300 megabytes per second, and the internal network carries 200. A cluster sized for the incoming rate alone would be short of processor and network capacity by a large factor.

What sets the write speed

Replicas do not speed up writes

A common misunderstanding is that more replicas make the cluster faster. They make reads faster, because more copies can answer, but they make writes slower and more expensive. Write capacity scales with the number of primary shards and the machines behind them, not with the number of replicas. If writes are your bottleneck, adding replicas will only make it worse.

Bulk loading tricks

Because replication multiplies cost, loading a large amount of data into a new index is much faster with zero replicas, followed by adding them at the end. The replicas then are built by copying finished segment files, which is cheaper than indexing each document again. The risk is that, during the load, there is no second copy, so the load must be repeatable. Restore the replicas as soon as the load finishes.

Memory guards for in-flight writes

Recent versions track the memory used by operations in flight at three stages: at the coordinating node, at the primary, and at the replica. When the limit at a stage is exceeded, new work at that stage is refused with a “too many requests” answer. The limit for the replica stage is somewhat higher than for the primary stage, because refusing at the replica would waste work already done on the primary. The result is that the cluster protects itself, and clients should be ready to slow down and retry.

Read scaling

Replicas do add read capacity. If searches are the bottleneck and the index is small enough to hold several copies, adding replicas can help. Adaptive replica selection sends requests to the healthiest copies, and requests are spread among all started copies. For very large indexes, adding replicas is expensive in storage, so consider whether caching, better queries, or more shards would serve better.

Recovery cost and shard size

The time to recover a copy grows with shard size when file copying is needed. Very large shards, hundreds of gigabytes, can take hours to rebuild and leave the group exposed for that time. Moderately sized shards, in the tens of gigabytes, rebuild faster and rebalance more easily. This is one more reason to avoid oversized shards.

Every design carries a price. Here is the honest balance sheet of synchronizing copies this way.

How strict should you be?

A design is only proven by how it behaves when things go wrong. Here is what happens in each failure, and how to prepare.

Designing for resilience

• Enough copies. At least one replica for anything you cannot afford to lose, and two where multiple failures could overlap.
• Copies in different zones. Configure awareness so that one zone never holds all copies of a shard.
• Capacity to absorb a failure. Keep enough free disk and headroom on the remaining nodes to take over the shards of a lost node.
• Healthy masters. Three or more dedicated master-eligible nodes, since promotions need the master.
• Snapshots. Regular, tested snapshots protect against losing every copy, or against mistakes that replicate to all copies.
• Practice. Test failover in a non-production environment, and measure how long recovery takes.

Rolling restarts

To restart nodes without losing availability, limit shard allocation before you start so that the cluster does not rebuild replicas for a node that will be back within minutes, restart one node at a time, wait for it to rejoin and for its shards to be recovered, then restore normal allocation and wait for green health before moving on. Because history is retained by leases, a node that returns within the lease period recovers by replaying operations, and the whole process is quick. Never restart several nodes at once if they hold copies of the same shard, since that could leave no in-sync copy running.

Dealing with unknown outcomes

When a write times out, or the connection breaks, the client cannot know whether it was applied. The correct response is to design writes to be safe to repeat. Supplying your own document identifiers achieves this, since repeating the operation overwrites the same document. With generated identifiers, a repeated write after an uncertain timeout can create a duplicate, so deduplicate later or avoid this pattern for critical data.

Replication moves your complete data between machines all day long. That traffic and those copies deserve protection.

The traffic between nodes carries full documents

Every operation the primary forwards to a replica contains the complete document. On an unprotected network, anyone who can watch traffic between nodes can read your data. Enable encryption for node-to-node communication. This also authenticates nodes to each other, if certificates are used properly, which brings the next point.

Who may join and receive copies

A node that joins the cluster can be assigned shard copies, and recovery would then send it entire shards. An attacker who could start a node that joins your cluster would receive real data. Prevent this by requiring certificates for cluster membership, by restricting network access to the ports nodes use to talk to each other, and by giving each cluster a unique name and credentials.

Privileges to change replication

Changing the number of replicas, moving shards, or forcing an unsafe promotion are administrative actions. They can weaken protection, reduce availability, or, in the case of forcing a stale copy, destroy data. Give the right to perform them only to trusted administrators and automation. Ordinary applications should have rights to read and write documents only.

More copies, more places to protect

Each replica is a full copy of the data on another machine’s disk. Security controls that apply to the primary, such as disk encryption, access restrictions on the machine, and secure disposal of retired hardware, must apply equally to every replica. A cluster is as secure as its least protected node.

Deletion and privacy

When you delete a document, the change is replicated to every copy. But because history is kept for recovery, the old version of the document may remain on disk for a while, in the retained history, until leases release it and a merge removes it. For most uses, this is invisible. For strict privacy requirements, such as erasing personal data on request, be aware that removal from disk is not instant, and design your deletion and retention processes with this window in mind. Snapshots also hold copies until they expire.

Snapshots as a separate security domain

Snapshots are stored outside the cluster, often in cloud storage. Protect that storage with its own access rules, encryption, and retention policy. A well-protected cluster with an open snapshot bucket is not a secure system.

Synchronization problems announce themselves early, if you are watching the right numbers.

Reading the patterns

Logs that help

The master’s log records when copies are marked as failed, when nodes leave, and when primaries are promoted. Data nodes log the start and end of recoveries, slow operations, and the reasons that a replica was failed. When something odd happens, read the master log around that time first, then the logs of the nodes involved. Ship logs to a central place, so that they survive the loss of a machine, and keep them long enough to investigate incidents that appeared days ago.

A troubleshooting playbook

Good habits keep replicas healthy and recoveries quick. Bad habits are behind most replication incidents.

Patterns that work

Anti-patterns to avoid

A short list you could hand to a new teammate on their first day.

Best practices

Common mistakes

• Assuming a replica is always identical to the primary. Copies agree on history for acknowledged writes, but refreshes and in-flight operations mean small differences at any instant.
• Assuming acknowledged means searchable. Visibility depends on refresh, which is separate.
• Adding replicas to speed up writes. More replicas increase the write cost.
• Forgetting that replicas cost storage. Every replica doubles the data on disk, and history retention adds more.
• Planning maintenance longer than the lease. A node away longer than the retention lease must recover by copying files.
• Forcing a stale primary without understanding the loss. It permanently discards writes that the stale copy missed.
• Leaving load-mode settings in place. Zero replicas or relaxed durability after a bulk load leave the index exposed.
• Never testing a failover. The first real failure should not be the first time you see how the cluster reacts.

Myths versus facts

Where does synchronization quietly decide success or trouble?

Many well-known organizations have said publicly that Elasticsearch is part of their systems, including Wikipedia’s search, Netflix, Uber, and Stack Overflow. They rarely publish how they set up replication, so the scenarios below are typical patterns for such workloads, not descriptions of anyone’s private configuration.

Short answers to the questions that come up most often.

If you skim only one part of this guide, make it this one.

Primary-replica synchronization is how Elasticsearch turns several machines into one reliable home for each shard. The primary numbers every write and passes it to its replicas, the cluster tracks which copies are fully up to date, and a strict set of rules governs failures, promotions, and recovery. It costs extra work and storage on every copy, and in return it lets the cluster survive failures without losing an acknowledged write.