Lab 4.2: ClusterState, Publishing, and the Applier Model

Lab 4.1 elected a cluster manager (formerly master). This lab follows what that manager spends its life doing: computing new ClusterStates and getting every node to agree on them. You will read the state object and its diffs, the two-phase publication protocol, and the applier/listener model — then make a real setting change and watch the cluster state version increment live in TRACE logs.

This is the OpenSearch analog of watching a Tez DAG state machine fire transitions on the async dispatcher. Here the "transition" is a versioned cluster state moving across the whole cluster.

Background

ClusterState is immutable and versioned. Every change produces a new state with version = old.version + 1 and a fresh stateUUID. The elected cluster manager is the only node that may compute a new state; it then publishes it to all nodes in two phases:

1. Publish: send the new state (usually a Diff) to every node; each validates and ACKs. Wait for a quorum of cluster-manager-eligible nodes.
2. Commit: once a quorum ACKs, tell every node to apply the state.

On apply, appliers run before listeners:

• ClusterStateApplier — makes a node's components consistent with the new state (routing, mappings, allocation). Runs first; an applier failure fails the apply.
• ClusterStateListener — observes the change and reacts. Runs after; fire-and-forget.

Deep-dive companions: cluster-state.md, cluster-state-publishing.md. The election that gives you a publisher is Lab 4.1.

Why This Lab Matters for Contributors

Almost every cluster-level operation — create/delete index, update mapping, change a setting, add a node, start/fail a shard — is a cluster state update. When one of those "hangs," "doesn't propagate," or "applies on some nodes but not others," you debug it through exactly the classes in this lab. The applier/listener distinction is also a constant source of plugin bugs (people register a listener where they needed an applier, or block the applier thread). You'll build your own listener in Lab 4.3.

Prerequisites

• A running node (./gradlew run) you can edit logging on, or a built distro.
• Lab 4.1 read (you know what term, quorum, and the manager are).
• Reading log: : > ~/opensearch-notes/reading-log-4.2.md

Part A — Read ClusterState and diffs (25 min)

Step 1 (10 min) — The state object

find server/src/main/java -name "ClusterState.java"
grep -n "private final\|long version\|String stateUUID\|Metadata metadata\|RoutingTable\|DiscoveryNodes\|ClusterBlocks\|public Builder builder\|public static Builder" \
  server/src/main/java/org/opensearch/cluster/ClusterState.java | head -40

Confirm for yourself:

• The fields are final — ClusterState is immutable. You never mutate it; you build a new one.
• The four components are Metadata, RoutingTable, DiscoveryNodes, ClusterBlocks (plus per-feature Customs — relevant in Lab 4.3).
• A version (monotonic) and a stateUUID.
• The Builder pattern: ClusterState.builder(previousState).metadata(...).build() is how every update produces the next state.

Read it live:

curl -s 'localhost:9200/_cluster/state?filter_path=version,state_uuid,master_node&pretty'
curl -s 'localhost:9200/_cluster/state/version?pretty'

Step 2 (15 min) — Diffs: how only the delta crosses the wire

Publishing the whole state to every node on every change would be wasteful. OpenSearch ships a Diff instead.

find server/src -name "Diff.java" -o -name "Diffable.java" | head
grep -rn "interface Diffable\|interface Diff<\|Diff<ClusterState>\|readDiffFrom\|diff(" \
  server/src/main/java/org/opensearch/cluster/ | head
grep -n "public Diff<ClusterState> diff\|static ClusterState readDiffFrom\|ClusterStateDiff\|apply(" \
  server/src/main/java/org/opensearch/cluster/ClusterState.java | head

The model:

• Diffable<T> types know how to compute diff(previous) and how to apply(previous) to reconstruct the new value.
• ClusterState.diff(previousState) produces a Diff<ClusterState> that contains only the changed components (e.g. only the Metadata diff if just a setting changed).
• A receiving node already on version N applies the diff to reach N+1. If it has fallen behind or the stateUUID doesn't line up, the publisher falls back to sending the full state.

This diff/full-state fallback is exactly what PublicationTransportHandler manages. Note: if you ever add a field to a state component, you must handle it in both the Writeable serialization and the diff — a classic source of BWC bugs (Level 9).

Log it:

cat >> ~/opensearch-notes/reading-log-4.2.md <<'EOF'
## ClusterState + diffs
- immutable, final fields; build next via ClusterState.builder(prev)...
- components: Metadata, RoutingTable, DiscoveryNodes, ClusterBlocks (+ Customs)
- versioned (monotonic) + stateUUID
- Diffable/Diff: publish only the delta; fall back to full state if receiver is behind / UUID mismatch
EOF

Part B — Read the publication protocol (25 min)

Step 3 (15 min) — Two-phase publish then commit

find server/src/main/java -name "PublicationTransportHandler.java"
grep -n "PUBLISH_STATE_ACTION_NAME\|COMMIT_STATE_ACTION_NAME\|sendClusterState\|handleIncomingPublishRequest\|handleApplyCommit\|PublishWithJoinResponse\|serializeFullClusterState\|serializeDiffClusterState" \
  server/src/main/java/org/opensearch/cluster/coordination/PublicationTransportHandler.java | head -30
find server/src/main/java -name "Publication.java"
grep -n "onPossibleCommitFailure\|isPublishQuorum\|handlePublishResponse\|onPossibleCompletion\|sendApplyCommit\|PublishResponse\|ApplyCommitRequest" \
  server/src/main/java/org/opensearch/cluster/coordination/Publication.java | head

Read the two transport action names: internal:cluster/coordination/publish_state and internal:cluster/coordination/commit_state (names may carry a version suffix). They correspond to the two phases:

The manager (Publication) waits for a publish quorum of cluster-manager-eligible ACKs before it sends the commit. If it can't reach quorum (e.g. partition), the publication fails and the state is never applied anywhere — the safety property that keeps a minority manager from changing the cluster. Trace the quorum gate:

grep -n "isPublishQuorum\|VoteCollection\|coordinationState.handlePublishResponse\|onPossibleCommitFailure" \
  server/src/main/java/org/opensearch/cluster/coordination/CoordinationState.java | head

Step 4 (10 min) — How publication connects to MasterService

The manager doesn't publish out of nowhere; the cluster-manager service computes the state and calls publish:

find server/src/main/java -name "MasterService.java"
grep -n "publish\|ClusterStatePublisher\|patchVersions\|incrementVersion\|runTasks\|calculateTaskOutputs\|onPublicationSuccess\|onPublicationFailed" \
  server/src/main/java/org/opensearch/cluster/service/MasterService.java | head -30

MasterService.runTasks(...) computes the new state (next subsection), increments its version, then hands it to the publisher (the Coordinator's publication). On success the task listeners are notified; on failure the update is rolled back (the in-memory state never moved). Log it:

cat >> ~/opensearch-notes/reading-log-4.2.md <<'EOF'
## Publication
- MasterService.runTasks computes newState, version++, calls publisher (Coordinator.publish)
- Phase 1 publish_state -> nodes ACK PublishResponse (no apply); manager waits for publish quorum
- Phase 2 commit_state (ApplyCommitRequest) -> nodes apply via ClusterApplierService
- no quorum -> publication fails -> state NEVER applied (safety)
EOF

Part C — Update tasks, batching, appliers, listeners (25 min)

Step 5 (12 min) — ClusterStateUpdateTask and batching on MasterService

Every state change starts as a task submitted to the cluster-manager service:

find server/src/main/java -name "ClusterStateUpdateTask.java" -o -name "ClusterStateTaskExecutor.java"
grep -n "abstract ClusterState execute\|onFailure\|submitStateUpdateTask\|submitStateUpdateTasks" \
  server/src/main/java/org/opensearch/cluster/service/MasterService.java | head
grep -n "interface ClusterStateTaskExecutor\|ClusterTasksResult\|execute(\|batchExecutionContext" \
  server/src/main/java/org/opensearch/cluster/ClusterStateTaskExecutor.java | head

Key facts:

• A ClusterStateUpdateTask (which is a ClusterStateTaskExecutor for itself) implements execute(currentState) -> newState. It must be a pure function of the input state — no blocking, no side effects.
• MasterService batches tasks that share the same ClusterStateTaskExecutor and runs them together in one computation, producing one new state for the whole batch. This is why creating 50 indices doesn't publish 50 times in lockstep — the executor coalesces them.
• The computation runs on the single-threaded CLUSTER_MANAGER_SERVICE (a.k.a. MASTER_SERVICE) thread pool — serialization of state changes is by design.

grep -n "MASTER_SERVICE\|CLUSTER_MANAGER_SERVICE\|masterServiceExecutor\|threadPoolExecutor" \
  server/src/main/java/org/opensearch/cluster/service/MasterService.java | head

Step 6 (13 min) — Appliers vs listeners

The committed state is applied by ClusterApplierService on every node:

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

Read applyChanges(...) (name may vary): it builds a ClusterChangedEvent (old state, new state, what changed) and then:

1. Calls all ClusterStateAppliers first — callClusterStateAppliers(...). These wire the node's internal components to the new state (e.g. IndicesClusterStateService creates/removes shards, RoutingService reacts). If an applier throws, the apply fails loudly.
2. Calls all ClusterStateListeners after — callClusterStateListeners(...). These just observe.

Warning: Both run on the single applier thread. Blocking inside an applier or listener freezes the cluster's ability to apply states. This is one of the most damaging plugin bugs. If you need to do slow work in reaction to a state change, hand it off to another thread pool.

Log it:

cat >> ~/opensearch-notes/reading-log-4.2.md <<'EOF'
## Update tasks + apply
- ClusterStateUpdateTask.execute(currentState)->newState : pure, no blocking
- MasterService batches tasks by ClusterStateTaskExecutor (one state per batch)
- runs on single-threaded CLUSTER_MANAGER_SERVICE pool
- ClusterApplierService.applyChanges: appliers FIRST (consistency, may fail apply), listeners AFTER (observe)
- never block in an applier/listener -> freezes apply
EOF

Part D — Watch a state update live in TRACE logs (20 min)

Now make a real change and watch the version increment.

Step 7 — Enable TRACE on the cluster service

Dynamically, on a running node:

curl -s -XPUT 'localhost:9200/_cluster/settings' -H 'Content-Type: application/json' -d '{
  "transient": {
    "logger.org.opensearch.cluster.service": "TRACE"
  }
}'

(Or set logger.org.opensearch.cluster.service: TRACE in config/log4j2.properties before start.)

Step 8 — Trigger an update and read the log

# Record the version before:
curl -s 'localhost:9200/_cluster/state/version?pretty'

# A trivially harmless cluster-state change:
curl -s -XPUT 'localhost:9200/_cluster/settings' -H 'Content-Type: application/json' -d '{
  "transient": { "cluster.routing.allocation.enable": "all" }
}'

# Version after:
curl -s 'localhost:9200/_cluster/state/version?pretty'

In the node log (logs/<cluster>.log for a distro, or stdout for ./gradlew run) you'll see lines like:

[TRACE][o.o.c.s.MasterService] cluster state updated, version [N+1], source [cluster_update_settings...]
[TRACE][o.o.c.s.MasterService] will process [cluster_update_settings ...]
[DEBUG][o.o.c.s.MasterService] publishing cluster state version [N+1]
[TRACE][o.o.c.s.ClusterApplierService] applying settings from cluster state with version [N+1]
[DEBUG][o.o.c.s.ClusterApplierService] processing [...]: execute
[DEBUG][o.o.c.s.ClusterApplierService] cluster state updated, version [N+1], source [...]

You are watching: MasterService compute v(N+1) → publish → ClusterApplierService apply v(N+1). The version you printed went up by exactly one. This is the whole chapter, made visible.

Turn TRACE back off when done:

curl -s -XPUT 'localhost:9200/_cluster/settings' -H 'Content-Type: application/json' -d '{
  "transient": { "logger.org.opensearch.cluster.service": null,
                 "cluster.routing.allocation.enable": null }
}'

Reading Exercises

# 1. Where does the new state's version get incremented?
grep -n "incrementVersion\|version() + 1\|builder(.*).version\|patchVersions" \
  server/src/main/java/org/opensearch/cluster/service/MasterService.java | head

# 2. How is "what changed" computed for appliers/listeners?
grep -n "ClusterChangedEvent\|indicesDeleted\|nodesChanged\|metadataChanged\|routingTableChanged" \
  server/src/main/java/org/opensearch/cluster/ClusterChangedEvent.java | head

# 3. Find a real applier in the engine.
grep -rln "implements ClusterStateApplier\|addStateApplier" server/src/main/java | head

Answer:

1. ClusterState is immutable. Show, with the builder API, how a setting change becomes a new state object. Where does the version++ happen?
2. Why are states published as diffs, and what makes the publisher fall back to a full state?
3. Explain the two phases of publication. What exactly is guaranteed if the commit phase never runs because quorum was lost?
4. MasterService batches tasks by ClusterStateTaskExecutor. Give a concrete example where batching matters for performance, and explain why the executor must be a pure function of the input state.
5. Appliers run before listeners. Give one operation that must be an applier (not a listener) and explain what would break if it were a listener.
6. What happens to the cluster if a plugin's ClusterStateListener blocks for 30 seconds on every state change? Which thread is it blocking?
7. From the TRACE logs you captured, quote the two log lines that show (a) the manager computing the new version and (b) a node applying it.

When you can produce the TRACE log of a version increment and answer all seven, you've completed Lab 4.2. Continue to Lab 4.3: Build It — A Custom ClusterStateListener / Custom Metadata.