Distributed Systems and Consensus — Intensive

Everything else in this book — the storage engine, the query engine, k-NN — assumes a working cluster: a set of nodes that agree on who is in charge, what indices exist, where every shard lives, and which copy of a document is authoritative. That agreement is not free. It is a distributed systems problem, and OpenSearch solves it with a real consensus protocol plus a separate data-plane distribution layer. This masterclass treats the cluster as the distributed-systems artifact it is: a replicated state machine for the cluster state (strongly consistent, CP), sitting above an eventually-consistent data plane (your documents, replicated across shards).

This chapter goes deeper than the two deep-dives it extends — discovery-coordination (the Coordinator state machine) and cluster-state-publishing (the two-phase publish). Read those first; this chapter assumes them and adds the why it is safe (linearizability, quorum overlap, the FLP/CAP framing), the data-plane distribution (allocation, rebalancing, recovery under churn), and a single mental model that ties the control plane and data plane together.

Note: "cluster manager" is the role formerly called master. In org.opensearch.cluster.coordination and org.opensearch.cluster.service the Java symbols were renamed — MasterService→ClusterManagerService, getMasterNode()→getClusterManagerNode(), INITIAL_MASTER_NODES→ INITIAL_CLUSTER_MANAGER_NODES — so the old Master* names no longer resolve in current source; only user-facing settings/REST params keep a deprecated master alias (cluster.initial_master_nodes, master_timeout). Read every master in an older snippet as its ClusterManager* equivalent, and grep for both since branches differ.

After this chapter you can:

• explain why OpenSearch needs consensus at all, and exactly what is under consensus (cluster state) versus what is not (document data);
• state the safety property — linearizable cluster state ordered by (term, version) — and prove from quorum overlap why two cluster managers cannot both commit;
• name the class that runs each step of an election and each phase of a publish, and grep to find it on your branch;
• describe the data-plane distribution — how AllocationService / BalancedShardsAllocator / AllocationDeciders place shards, how rebalancing works, and how recovery brings a shard up under node join/leave;
• place OpenSearch precisely on the CAP map (CP control plane, AP-ish data plane) and reason about behavior during a partition.

The labs that go with this chapter:

• Lab DC1: Trace a Cluster-Manager Election — bring up 3 nodes, TRACE org.opensearch.cluster.coordination, watch a clean election and then a re-election after you kill the leader.
• Lab DC2: Failure Detection and Network Partitions — induce a real partition with the test framework's NetworkDisruption, watch the minority step down, prove quorum stops a second cluster manager.
• Lab DC3: Cluster State Publish/Commit — trace one cluster-state update end to end, Diff-based publish, applier ack model.
• Lab DC4: Allocation and Rebalancing Under Churn — create replicas, add/remove nodes, force an unassigned shard, diagnose with _cluster/allocation/explain.

Two planes: what is under consensus and what is not

The single most important distinction in this entire chapter:

These are not the same problem and are not solved the same way. The cluster state is small, changes infrequently (relative to document writes), and must be totally ordered — every node must agree, in the same order, on what shards exist and where. Document data is large, changes constantly, and only needs ordering within a replication group (one primary + its replicas), enforced by the primary, not by cluster-wide consensus.

flowchart TB
    subgraph CP["Control plane — CP, consensus"]
      CM["Cluster manager (elected)"] -->|publish→commit| N1["node A state"]
      CM -->|publish→commit| N2["node B state"]
      CM -->|publish→commit| N3["node C state"]
    end
    subgraph DP["Data plane — per-group, eventually consistent"]
      P["primary shard"] -->|replicate| R1["replica 1"]
      P -->|replicate| R2["replica 2"]
    end
    CP -.->|"routing table says where P, R1, R2 live"| DP

The control plane describes the data plane (the routing table inside the cluster state names every shard's node), but the two run on different rails. A write to a document goes primary→replica without touching consensus; a change to where that shard lives goes through consensus. Keep this split in your head for the whole chapter — most "is OpenSearch CP or AP?" confusion is from collapsing these two.

Why consensus at all? The agreement problem

Strip the system down. You have N machines, an asynchronous network (messages can be delayed, reordered, or dropped), and machines that can crash and restart. You want all live machines to agree on a sequence of values (here: cluster states), such that:

• Agreement / safety: no two machines ever commit different values at the same position in the sequence. (Never two conflicting cluster states at the same version.)
• Validity: a committed value was actually proposed by some node.
• Termination / liveness: if enough machines are up and can talk, they eventually decide.

The FLP impossibility result (Fischer–Lynch–Paterson) says: in a fully asynchronous system with even one crash-faulty node, no deterministic protocol can guarantee both safety and liveness. Real systems escape FLP by keeping safety always and sacrificing liveness only when the network is too broken — using timeouts (a partial-synchrony assumption) to make progress when things are healthy. OpenSearch's coordination layer is exactly this: it never violates safety (no split-brain commit), but it will stop making progress (no elected cluster manager, writes to the control plane blocked) when a quorum is unreachable.

That trade is the whole game. The rest of this chapter is how it keeps safety and when it gives up liveness.

The Coordinator: a Raft-like / "Zen2" protocol

OpenSearch inherits the Zen2 design (from Elasticsearch 7, post-7.10.2 fork) — a custom, Raft-inspired consensus algorithm. It is not literal Raft (no replicated log of arbitrary commands; the "log" is the sequence of cluster states), but it shares Raft's spine: a single leader per term, terms as a logical clock, and quorum-based voting.

find server/src/main/java/org/opensearch/cluster/coordination -name "*.java" | sort

The classes you must be able to name and find:

grep -n "enum Mode\|becomeCandidate\|becomeLeader\|becomeFollower" \
  server/src/main/java/org/opensearch/cluster/coordination/Coordinator.java

Terms: the logical clock

A term is a monotonically increasing election epoch. Every election attempt bumps the term. The invariant that makes terms useful:

A node accepts a publication only from a leader whose term is ≥ its own current term, and there is at most one leader per term (because becoming leader of term T required a quorum of votes for term T, and a node votes at most once per term).

This is the same trick as Raft's term. It gives you a total order on leadership epochs for free: term 7's leader strictly supersedes term 6's. A stale ex-leader that was partitioned away comes back at the old term, tries to publish, and every node rejects it (CoordinationStateRejectedException) because they have moved to a higher term. That is how the protocol fences a zombie cluster manager.

grep -n "currentTerm\|handleStartJoin\|handleJoin\|handlePublishRequest\|handleCommit\|CoordinationStateRejectedException" \
  server/src/main/java/org/opensearch/cluster/coordination/CoordinationState.java

The election, message by message

sequenceDiagram
    participant C as Candidate (term T)
    participant PV as PreVoteCollector
    participant P1 as Peer 1 (eligible)
    participant P2 as Peer 2 (eligible)
    Note over C: LeaderChecker fired / no leader → CANDIDATE
    C->>PV: startPreVoting()
    PV->>P1: PreVoteRequest
    PV->>P2: PreVoteRequest
    P1-->>PV: PreVoteResponse (last term/version)
    P2-->>PV: PreVoteResponse
    Note over PV: pre-vote quorum? yes → ElectionScheduler fires
    C->>C: incrementTerm → T+1
    C->>P1: StartJoinRequest(term T+1)
    C->>P2: StartJoinRequest(term T+1)
    P1-->>C: Join(term T+1)
    P2-->>C: Join(term T+1)
    Note over C: CoordinationState.handleJoin → VoteCollection
    Note over C: isElectionQuorum(lastCommittedConfiguration)? yes
    C->>C: becomeLeader → LEADER
    Note over C: now publish first state of term T+1 (two-phase)

The four classes that cooperate to win an election, in firing order:

1. ElectionSchedulerFactory — schedules the attempt (randomized backoff so two candidates rarely fire simultaneously).
2. PreVoteCollector — runs the non-binding pre-vote; bails if no quorum is plausible, so a flapping/partitioned node cannot keep disrupting a healthy cluster by bumping the term.
3. JoinHelper — broadcasts the StartJoinRequest at term T+1 and collects Join votes.
4. CoordinationState.handleJoin / isElectionQuorum — counts votes into a VoteCollection and checks whether they form a quorum of the last-committed voting configuration. When they do, Coordinator.becomeLeader runs.

grep -n "startElection\|StartJoinRequest\|incrementTerm\|joinLeaderInTerm" \
  server/src/main/java/org/opensearch/cluster/coordination/Coordinator.java
grep -n "handleJoin\|isElectionQuorum\|VoteCollection\|electionWon" \
  server/src/main/java/org/opensearch/cluster/coordination/CoordinationState.java

Trace this live in Lab DC1: TRACE the coordination package and map every log line to one of these four classes.

The voting configuration, quorum, and split-brain prevention

The voting configuration is the set of cluster-manager-eligible node IDs whose votes count toward quorum. Quorum = strict majority of the voting configuration. This one rule is the entire split-brain guarantee, and it is worth understanding why mathematically, not just memorizing it.

find server -name "VotingConfiguration.java"
grep -n "hasQuorum\|getNodeIds" \
  server/src/main/java/org/opensearch/cluster/coordination/CoordinationMetadata.java

Quorum overlap is the safety proof

Two facts about a single voting configuration of size n:

1. A leader of any term needed a strict majority (≥ ⌊n/2⌋ + 1) of votes.
2. Any two strict majorities of the same set share at least one member (their sizes sum to more than n, so by pigeonhole they intersect).

Therefore: if leader X (term Tx) and leader Y (term Ty) both got elected from the same voting config, some node z voted for both. But a node votes at most once per term and never votes for a term it has already seen exceeded — so X and Y cannot have the same term, and the higher term wins. Two disjoint halves can never both elect a leader, because two disjoint majorities of one set do not exist. That is the formal split-brain prevention.

flowchart LR
    subgraph VC["Voting config {A,B,C,D,E} — quorum = 3"]
      direction LR
      A; B; C; D; E
    end
    subgraph S3["Partition side 1: {A,B,C}"]
      A2["A"]; B2["B"]; C2["C"]
    end
    subgraph S2["Partition side 2: {D,E}"]
      D2["D"]; E2["E"]
    end
    S3 -->|"3 ≥ quorum → elects + commits"| OK["✓ has cluster manager, serves control-plane writes"]
    S2 -->|"2 < quorum → cannot elect"| NO["✗ stays CANDIDATE, no cluster manager"]

Worked example — failures tolerated

Warning: Deploy an odd number of cluster-manager-eligible nodes (3 or 5 for HA). Two is the worst topology: quorum is 2, so the loss of either node halts the cluster manager — strictly worse than a single node. Reconfigurator auto-shrinks/grows the voting config to stay odd when nodes are added/removed, but it cannot fix a fundamentally even, two-node design.

grep -n "reconfigure\|auto_shrink_voting_configuration\|ODD\|isEven" \
  server/src/main/java/org/opensearch/cluster/coordination/Reconfigurator.java

Bootstrapping the very first config

A brand-new cluster has no committed voting configuration, so there is a chicken/egg problem: no leader without a config, no committed config without a leader. ClusterBootstrapService breaks it — the operator lists the initial eligible nodes once via cluster.initial_cluster_manager_nodes, and that list seeds the first voting configuration.

grep -rn "INITIAL_CLUSTER_MANAGER_NODES\|initial_cluster_manager_nodes\|INITIAL_MASTER_NODES" \
  server/src/main/java/org/opensearch/cluster/coordination/ClusterBootstrapService.java

Warning: This is bootstrap-only. Set it on first formation, then remove it. Leaving it set, or setting it differently per node, can bootstrap two independent clusters from one fleet — a true split-brain that bypasses the quorum guarantee, because each side thinks it is the seed config.

Two-phase publish → commit (the control-plane write path)

Winning an election is not enough; a leader must publish a cluster state to make its leadership durable and to push every later change. Publication is a two-phase protocol modeled on the consensus above. This is covered in depth in cluster-state-publishing; here is the safety-relevant shape.

sequenceDiagram
    participant MS as ClusterManagerService (cluster manager)
    participant P as Publication
    participant Q as quorum of followers
    MS->>P: publish(new state v=V+1, term=T)
    P->>Q: PHASE 1 PublishRequest (full state or Diff)
    Q->>Q: validate (term ≥ mine, version > mine) → persist as lastAcceptedState
    Q-->>P: accept
    Note over P: wait for a QUORUM of accepts
    P->>Q: PHASE 2 ApplyCommit
    Q->>Q: ClusterApplierService applies (appliers, then listeners)
    Q-->>P: ack
    P-->>MS: published (or timed out → step down)

The key safety property: a state is committed only after a quorum durably accepted it. So if the cluster manager dies mid-publish, the next leader's election will see (from the quorum it talks to) the highest accepted (term, version) and continue from there — the cluster is never left half-updated with a committed-on-some, not-on-others state.

grep -n "PUBLISH_STATE_ACTION_NAME\|COMMIT_STATE_ACTION_NAME\|handlePublishRequest\|handleApplyCommit\|sendApplyCommit" \
  server/src/main/java/org/opensearch/cluster/coordination/PublicationTransportHandler.java

cluster.publish.timeout (default 30s): if a quorum does not accept in time, publication fails and the cluster manager may step down. The usual culprit is a slow applier on a follower blocking the applier thread — trace it in Lab DC3.

Safety: linearizability of the cluster state via (term, version)

Put the pieces together. Every committed cluster state carries a (term, version):

• term increases on every election (leadership epoch).
• version increases on every published state within a term.

Order states lexicographically by (term, version). The protocol guarantees:

1. Uniqueness per position: at most one committed state exists for any (term, version) — quorum overlap + at-most-one-leader-per-term.
2. Monotonicity: a node accepts a state only if its (term, version) is strictly greater than what the node last accepted.
3. Durability before effect: a state takes effect (phase 2) only after a quorum accepted it (phase 1).

Together these make the sequence of committed cluster states linearizable: there is one global total order, every node sees a prefix of it, and once a state at (T, V) is committed and acknowledged to a client, no node ever sees a different state at (T, V) or "goes backward." This is the formal sense in which the control plane is CP.

Why only a quorum of the voting config can elect or commit: both operations require a strict majority of the same set. Because two strict majorities of one set always intersect, any new leader's quorum overlaps the previous commit's quorum, so the new leader is guaranteed to learn the latest committed state before it commits anything new. That overlap is the linchpin — it is why a minority partition can neither elect a new manager nor commit a new state, and therefore cannot fork the timeline.

Liveness: the three fault detectors

Safety is permanent; liveness is conditional. Three mechanisms detect failures and drive recovery:

grep -n "leaderFailed\|handleLeaderCheck" \
  server/src/main/java/org/opensearch/cluster/coordination/LeaderChecker.java
grep -n "onNodeFailure\|handleDisconnectedNode\|FollowerChecker" \
  server/src/main/java/org/opensearch/cluster/coordination/FollowersChecker.java
grep -n "describeQuorum\|getDescription" \
  server/src/main/java/org/opensearch/cluster/coordination/ClusterFormationFailureHelper.java

The two heartbeat loops run in opposite directions on purpose:

• LeaderChecker (followers → leader) catches a dead or partitioned-away leader: followers stop hearing it, step down to CANDIDATE, and elect a new one.
• FollowersChecker (leader → followers) catches a dead or partitioned-away follower: the leader removes it from the cluster state (which moves its shards).

A single bidirectional check would not distinguish "the leader is gone" from "I am the one who is isolated." Two unidirectional checks let each side independently reach the correct conclusion — which, combined with quorum, is what makes the minority side step down rather than declare itself the cluster.

Timing is governed by cluster.fault_detection.*. Too aggressive → spurious elections on transient blips or GC pauses; too loose → slow recovery from real failures. This tension is exactly what you tune and observe in Lab DC2.

A partition, step by step

sequenceDiagram
    participant L as Leader (term 5) — minority side {L}
    participant F1 as Follower 1 — majority {F1,F2}
    participant F2 as Follower 2 — majority {F1,F2}
    Note over L,F2: 3 eligible nodes, quorum = 2. Partition isolates L from {F1,F2}.
    F1->>L: LeaderCheck (times out)
    F2->>L: LeaderCheck (times out)
    Note over F1,F2: leader lost → both become CANDIDATE
    F1->>F2: pre-vote, then StartJoin(term 6)
    F2-->>F1: Join(term 6)
    Note over F1: quorum (2 of 3) → becomeLeader term 6, publishes
    L->>F1: FollowersCheck (times out — cannot reach quorum)
    Note over L: cannot publish (no quorum) → step down to CANDIDATE
    Note over L: minority of 1 can never reach quorum → no cluster manager
    Note over L,F2: HEAL: L rejoins, sees term 6 > 5, becomes FOLLOWER of new leader

The minority side (here, the lone old leader) cannot reach quorum, so it cannot publish, cannot stay leader, and cannot elect itself again — it simply stops serving control-plane writes until the partition heals. On heal, it sees the higher term and rejoins as a follower. There is never a moment where two nodes both commit as cluster manager. That is the guarantee made concrete.

The data plane: allocation, rebalancing, recovery

The control plane decides what shards exist; the data plane decides where they live and how copies stay in sync. The bridge is the routing table inside the cluster state — but computing that routing table (which node hosts which shard) is itself a substantial subsystem, and recomputing it under churn is where most "why is my shard unassigned?" pain comes from.

find server/src/main/java/org/opensearch/cluster/routing/allocation -name "*.java" | sort | head -40

grep -rln "extends AllocationDecider" \
  server/src/main/java/org/opensearch/cluster/routing/allocation/decider/ | head

Deciders: the YES/NO/THROTTLE gate

Every shard placement decision is the conjunction of many deciders. Some you will meet constantly:

THROTTLE is not NO: it means "yes eventually, but not right now" — it is how the cluster paces recovery so adding a node does not stampede the disk and network. You will see THROTTLE constantly in _cluster/allocation/explain.

flowchart TD
    Trigger["node join / leave / reroute / setting change"] --> AS["AllocationService.reroute"]
    AS --> GA["GatewayAllocator: place existing shards (find best copy)"]
    AS --> BSA["BalancedShardsAllocator: place new + rebalance"]
    GA --> DEC{"AllocationDeciders: YES / NO / THROTTLE"}
    BSA --> DEC
    DEC -->|YES| RN["RoutingNodes: assign / relocate"]
    DEC -->|THROTTLE| LATER["leave for next reroute"]
    DEC -->|NO| UNASSIGNED["stay UNASSIGNED with reason"]
    RN --> NS["new routing table → new ClusterState → publish"]

Allocation vs rebalancing

Two different jobs, often confused:

Allocation is about correctness/availability (every shard must live somewhere); rebalancing is about efficiency (spread load). Rebalancing is gated by cluster.routing.rebalance.enable and cluster.routing.allocation.allow_rebalance (by default, only rebalance once the cluster is green/indices_all_active).

Recovery: making a placed shard real

Placing a shard in the routing table is a promise; recovery fulfills it by actually getting the shard's data onto the node and bringing it to STARTED.

find server -name "PeerRecoverySourceService.java" -o -name "PeerRecoveryTargetService.java"
grep -n "recoverToTarget\|phase1\|phase2\|prepareTargetForTranslog" \
  server/src/main/java/org/opensearch/indices/recovery/RecoverySourceHandler.java 2>/dev/null | head

Recovery is throttled (indices.recovery.max_bytes_per_sec, ThrottlingAllocationDecider) so a node join does not saturate the network. The full mechanics are in recovery and replication; for allocation, the point is: a shard goes UNASSIGNED → INITIALIZING (recovering) → STARTED, and only when STARTED is it counted in _cluster/health.

Deep-link: the data-plane distribution mechanics live in shard-allocation, recovery, replication, and the masterclass sharding-routing. This chapter connects them to consensus: every routing-table change is a cluster-state change, published through the very two-phase protocol above.

CAP, consistency, and where OpenSearch actually sits

Now the honest CAP placement. CAP says: under a network Partition, you choose Consistency or Availability. OpenSearch's two planes choose differently:

So "is OpenSearch CP or AP?" is the wrong question. The cluster state is CP (linearizable, quorum-guarded). The data is eventually consistent across replicas: a primary acknowledges a write once the required copies (wait_for_active_shards) have it, replicas catch up asynchronously (especially under segment replication), and a partitioned replica serves slightly-stale reads until it reconciles. There is a per-document total order (the primary's sequence numbers, seqNo/primaryTerm), but not the global linearizability the cluster state has.

Note: This is why a partitioned minority node can still serve search from a replica it hosts (data plane, available) but cannot create an index (control plane, needs the cluster manager / quorum). Two different answers from one node — because two different consistency models.

The data-plane analog of the control-plane term is the shard's primary term (primaryTerm), bumped each time a primary is (re)assigned, plus per-operation sequence numbers (seqNo). These fence a stale primary the same way the cluster term fences a stale cluster manager — see replication.

Worked example: the life of a node-leave event

Tie control plane and data plane together with one event — a data node holding primaries and replicas crashes.

1. Detect (liveness): the leader's FollowersChecker stops hearing the node and, after the configured retries, declares it failed.
2. Control-plane write: the leader submits a ClusterStateUpdateTask that removes the node from DiscoveryNodes. AllocationService.reroute runs as part of computing the new state: every shard that lived on the dead node becomes UNASSIGNED; for each, a surviving in-sync replica (if any) is promoted to primary (primaryTerm bumps), and new replicas are scheduled.
3. Publish→commit: the new cluster state (new routing table, higher version) goes through the two-phase publish. Every node now agrees: shard X's primary is now on node B.
4. Data-plane recovery: on the nodes that gained shards, IndicesClusterStateService (the big applier) sees the routing change and starts peer recovery to rebuild the missing replicas from the new primaries — throttled.
5. Green again: as each recovered shard reaches STARTED, a further cluster-state update records it; _cluster/health climbs from red/yellow back to green.

Notice every structural change (steps 2, 3, 5) is a consensus-ordered cluster-state change, while every data movement (step 4) is throttled peer replication off the critical consensus path. That separation is the design.

Common bugs and symptoms

Validation: prove you understand this

1. State precisely what is under consensus and what is not. Give one operation that the control plane must serialize and one that the data plane handles without cluster-wide consensus.
2. Explain linearizability of the cluster state in terms of (term, version). Why can no two committed states share a (term, version)? Use the quorum-overlap argument.
3. Draw the CANDIDATE/LEADER/FOLLOWER machine and label each transition with the class that triggers it (ElectionSchedulerFactory, PreVoteCollector, JoinHelper+CoordinationState, LeaderChecker, FollowersChecker, a valid publication).
4. With the quorum arithmetic, explain why a 2-eligible-node cluster is strictly worse than both 1-node and 3-node. What does Reconfigurator do and what can it not fix?
5. Describe the two-phase publish. At which phase has a follower applied the state? Why is committing only after a quorum accepts what prevents a half-updated cluster?
6. Walk through a 3-node partition (1 vs 2). Which side keeps a cluster manager, which steps down, and exactly which detector fires on each side? Why can the minority never elect itself?
7. Trace a node-leave: name every cluster-state change it causes, which are consensus-ordered, and where data movement (recovery) happens off that path.
8. For an UNASSIGNED replica, name three distinct deciders that could be the cause and the field in _cluster/allocation/explain that would tell you which.
9. Place OpenSearch on the CAP map for both planes and justify each placement with a concrete partition scenario.
10. Explain how primaryTerm/seqNo fence a stale primary, and draw the analogy to how the cluster term fences a stale cluster manager.