Cluster State Publishing

A ClusterState is only useful if every node sees the same one. This chapter explains how a new state is produced on the elected cluster manager and distributed to the whole cluster. There are two distinct services and two distinct jobs: MasterService (the cluster-manager service) computes new states from a queue of update tasks, and ClusterApplierService applies committed states to local components. Between produce and apply sits a two-phase publish→commit protocol that guarantees a state is durably accepted by a quorum before it is allowed to take effect anywhere. Get this chapter wrong in code and you cause cluster-wide hangs; get it right and you can safely react to any cluster change.

After this chapter you should be able to: write a correct ClusterStateUpdateTask; explain why update tasks are batched and on which thread they run; distinguish ClusterStateApplier from ClusterStateListener and know which runs first; trace a state from "submitted" to "committed and applied everywhere"; and explain why blocking inside an applier is a cardinal sin.

Note: MasterService is the cluster-manager service. The class keeps the old name; treat "master" as "cluster manager." OpenSearch also exposes ClusterManagerService as an alias in places.

Two services, two jobs

find server/src/main/java/org/opensearch/cluster/service -name "*.java"
grep -n "class MasterService\|class ClusterApplierService\|class ClusterService" \
  server/src/main/java/org/opensearch/cluster/service/*.java

flowchart LR
    subgraph "Cluster manager node"
      Q[update task queue] --> MS[MasterService]
      MS -->|compute new state| PUB[Publication]
    end
    PUB -->|publish phase| F1[follower 1]
    PUB -->|publish phase| F2[follower 2]
    F1 -->|accept| PUB
    F2 -->|accept| PUB
    PUB -->|"commit (quorum accepted)"| F1
    PUB -->|commit| F2
    F1 --> CAS1[ClusterApplierService applies]
    F2 --> CAS2[ClusterApplierService applies]
    MS --> CASM[cluster manager applies locally too]

The update-task model

You never mutate cluster state directly. You submit a task describing the change, and the cluster-manager service runs it. Two APIs:

find server -name "ClusterStateUpdateTask.java" -o -name "ClusterStateTaskExecutor.java"
grep -n "abstract ClusterState execute\|onFailure\|clusterStateProcessed" \
  server/src/main/java/org/opensearch/cluster/ClusterStateUpdateTask.java

A minimal update task:

clusterService.submitStateUpdateTask("add-my-custom", new ClusterStateUpdateTask() {
    @Override
    public ClusterState execute(ClusterState current) {
        // pure function: derive a NEW immutable state from current
        return ClusterState.builder(current)
            .metadata(Metadata.builder(current.metadata()).putCustom(MyCustom.TYPE, value))
            .build();
    }
    @Override
    public void onFailure(String source, Exception e) { /* log; never throw */ }
    @Override
    public void clusterStateProcessed(String source, ClusterState before, ClusterState after) {
        // runs after the state is committed AND applied locally; safe to ack here
    }
});

Warning: execute(ClusterState) must be a pure function: derive a new immutable state, do no I/O, do not block, do not sleep. It runs on the single cluster-manager update thread; blocking it stalls all cluster state progress for the entire cluster. Do expensive work before submitting, or in clusterStateProcessed after.

Batching

The cluster-manager service batches tasks that share a ClusterStateTaskExecutor instance and can be coalesced (e.g. 50 shard-started notifications arriving at once produce one new cluster state instead of 50). Batching is the reason a busy cluster does not publish a new state per shard event.

grep -n "executeTasks\|batch\|TaskBatcher\|class MasterService" \
  server/src/main/java/org/opensearch/cluster/service/MasterService.java

flowchart TD
    T1[shard-started s0] --> B[TaskBatcher]
    T2[shard-started s1] --> B
    T3[shard-started s2] --> B
    B -->|one executor run| Exec["ShardStartedClusterStateTaskExecutor.execute(batch)"]
    Exec --> NS["one new ClusterState (version+1)"]

Two-phase publish → commit

When MasterService has computed a new state, it does not simply broadcast it. It runs a Publication, modeled on the coordination consensus from discovery-coordination.md:

find server -name "Publication.java" -o -name "PublicationTransportHandler.java"
grep -n "publish\|sendPublishRequest\|handlePublishRequest\|sendApplyCommit\|handleApplyCommit" \
  server/src/main/java/org/opensearch/cluster/coordination/PublicationTransportHandler.java

sequenceDiagram
    participant MS as MasterService
    participant P as Publication
    participant Q as quorum of followers
    MS->>P: publish(new state v101)
    P->>Q: PHASE 1 PublishRequest (full state or Diff)
    Q-->>P: accept (persisted to lastAcceptedState)
    Note over P: wait for a QUORUM of accepts
    P->>Q: PHASE 2 ApplyCommit
    Q->>Q: ClusterApplierService applies v101
    Q-->>P: ack (after appliers run)
    P-->>MS: publication succeeded (or timed out)

This two-phase shape is what makes cluster state changes safe across partitions: a state is only committed once a quorum has durably accepted it, so a failed cluster manager cannot leave the cluster half-updated. The Diff optimization from cluster-state.md rides on phase 1.

Note: Publication has a timeout (cluster.publish.timeout, default 30s). If a quorum does not accept in time, publication fails and the state is not committed; the cluster manager may step down. Slow appliers on followers are a common cause of publish timeouts.

Applying a committed state: appliers vs listeners

Once a node receives the commit, ClusterApplierService applies the state by running two ordered groups of callbacks:

find server -name "ClusterStateApplier.java" -o -name "ClusterStateListener.java"
grep -n "applyClusterState\|clusterChanged\|addStateApplier\|addListener\|callClusterStateAppliers\|callClusterStateListeners" \
  server/src/main/java/org/opensearch/cluster/service/ClusterApplierService.java

The ordering contract is precise and load-bearing:

1. All appliers run, in order, on the single applier thread.
2. Only after every applier returns do listeners run.
3. Both run synchronously on the applier thread — so both block cluster state progress while they execute.

flowchart TD
    Commit[ApplyCommit received] --> A1["applier 1 (IndicesClusterStateService)"]
    A1 --> A2["applier 2 (...)"]
    A2 --> A3[applier N]
    A3 --> L1[listener 1]
    L1 --> L2[listener N]
    L2 --> Done["state applied; ack sent"]

Warning: An applier or listener that blocks (does I/O, waits on a lock, calls back into the cluster service, or sleeps) stalls the applier thread, which delays the commit ack, which can blow the publish timeout, which can cause the cluster manager to step down — a cluster-wide cascade from one bad callback. If you need to do slow work, hand it off to a thread pool from inside the callback. This is the most common newcomer mistake; see Level 4 lab 4.2 and Level 4 lab 4.3.

The biggest applier is IndicesClusterStateService — it reconciles the local node's shards with the new routing table, creating, starting, and closing shards to match. That is the bridge from cluster state into index-shard-lifecycle.md:

grep -n "applyClusterState\|createOrUpdateShards\|removeShards\|failAndRemoveShard" \
  server/src/main/java/org/opensearch/indices/cluster/IndicesClusterStateService.java

The ackListener: knowing the change took effect

Some operations (e.g. PUT mapping with ?wait_for_active_shards, or any API that returns "acknowledged": true) must wait until enough nodes have applied the new state. The publication carries an ackListener that fires per node as each one acks the commit. The REST response is held until the ack condition is satisfied (or times out).

grep -n "AckListener\|onNodeAck\|onCommit\|ackTimeout\|AckedClusterStateUpdateTask" \
  server/src/main/java/org/opensearch/cluster/AckedClusterStateUpdateTask.java \
  server/src/main/java/org/opensearch/cluster/coordination/Publication.java 2>/dev/null

This is why a PUT /index/_mapping can return "acknowledged": false — it means the change was committed but not acknowledged by all nodes within the master/cluster-manager timeout, not that it failed.

End-to-end sequence

sequenceDiagram
    participant C as Client / internal caller
    participant CS as ClusterService
    participant MS as MasterService
    participant EX as ClusterStateTaskExecutor
    participant PUB as Publication
    participant F as Followers
    participant CAS as ClusterApplierService (each node)
    C->>CS: submitStateUpdateTask(source, task)
    CS->>MS: enqueue task
    Note over MS: batch compatible tasks
    MS->>EX: execute(currentState, tasks) -> newState
    MS->>PUB: publish(newState)
    PUB->>F: PublishRequest (state or Diff)
    F-->>PUB: accept (lastAcceptedState)
    Note over PUB: quorum reached
    PUB->>F: ApplyCommit
    F->>CAS: applyClusterState (appliers, then listeners)
    CAS-->>PUB: ack
    PUB-->>MS: publication complete
    MS->>EX: clusterStateProcessed callbacks
    EX-->>C: response (acknowledged if ack condition met)

Reading exercise

# 1. The update-task loop and batching.
grep -n "runTasks\|executeTasks\|publish\|TaskBatcher" \
  server/src/main/java/org/opensearch/cluster/service/MasterService.java

# 2. The applier/listener split and ordering.
grep -n "callClusterStateAppliers\|callClusterStateListeners\|addStateApplier\|addListener" \
  server/src/main/java/org/opensearch/cluster/service/ClusterApplierService.java

# 3. The publish/commit transport handlers.
grep -n "PUBLISH_STATE_ACTION_NAME\|COMMIT_STATE_ACTION_NAME\|handlePublishRequest\|handleApplyCommit" \
  server/src/main/java/org/opensearch/cluster/coordination/PublicationTransportHandler.java

# 4. Tests that exercise publication.
./gradlew :server:test --tests "org.opensearch.cluster.service.MasterServiceTests"
./gradlew :server:test --tests "org.opensearch.cluster.service.ClusterApplierServiceTests"

# 5. Watch real publishes on a running node.
curl -s "localhost:9200/_cluster/health?pretty&wait_for_events=languid"

Answer:

1. On which node does MasterService run, and on which node does ClusterApplierService run? Why the difference?
2. Why must ClusterStateUpdateTask.execute() be pure and non-blocking? What exactly stalls if it blocks?
3. What is the difference between phase 1 (publish) and phase 2 (commit)? At which phase has a follower applied the state?
4. In ClusterApplierService, prove from the code that all appliers run before any listener. What is the consequence if an applier throws?
5. A PUT _mapping returns "acknowledged": false. Trace, via the ackListener, what that means and what timeout governs it.
6. Find IndicesClusterStateService.applyClusterState. How does this one applier turn a routing-table change into shard creation/removal on the local node?

Common bugs and symptoms

Validation: prove you understand this

1. Draw the produce→publish→commit→apply pipeline and label which steps happen on the cluster manager vs every node.
2. Write a correct ClusterStateUpdateTask skeleton and annotate which method must be pure, which may do post-commit work, and which handles failure.
3. Explain batching with a concrete example (N shard-started events) and state how many cluster states result.
4. Distinguish phase 1 from phase 2 of publication and explain why committing only after a quorum accepts prevents a half-updated cluster.
5. Explain the applier-before-listener ordering and give one task that belongs in an applier and one that belongs in a listener.
6. Explain precisely how blocking inside an applier can make the elected cluster manager step down. Name every link in the cascade.