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 ClusterManagerService

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

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

ClusterManagerService.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
- ClusterManagerService.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 ClusterManagerService

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/ClusterManagerService.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.
• ClusterManagerService 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/ClusterManagerService.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
- ClusterManagerService 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.ClusterManagerService] cluster state updated, version [N+1], source [cluster_update_settings...]
[TRACE][o.o.c.s.ClusterManagerService] will process [cluster_update_settings ...]
[DEBUG][o.o.c.s.ClusterManagerService] 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: ClusterManagerService 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/ClusterManagerService.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. ClusterManagerService 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.

Coding Exercises

You watched a version increment in TRACE logs. Now make that observable behavior testable: write listeners and update tasks and assert they fire, increment, and apply in the right order. Locate every class with rg first.

1. (warm-up) Assert immutability and the builder bump. Write an OpenSearchTestCase that takes a ClusterState (build one via ClusterState.builder(ClusterName.DEFAULT).build() or ClusterState.EMPTY_STATE), derives a new state with a changed Metadata via ClusterState.builder(prev).version(prev.version() + 1).build(), and asserts the new state's version() is exactly one higher while the original's is unchanged. This pins the "immutable, versioned" contract from Part A in code.

2. (warm-up) Compute a ClusterChangedEvent delta. Build an "old" state with no indices and a "new" state with index foo (use the test helpers — rg -l "extends OpenSearchTestCase" server/src/test/java/org/opensearch/cluster | head), construct a ClusterChangedEvent("test", newState, oldState) and assert indicesCreated() contains "foo" and indicesDeleted() is empty. Find the constructor signature: rg -n "ClusterChangedEvent\(|indicesCreated|indicesDeleted" server/src/main/java/org/opensearch/cluster/ClusterChangedEvent.java. This is the same diff your Lab 4.3 listener will rely on.

3. (core) A ClusterStateListener that fires, asserted with a latch. Write an OpenSearchSingleNodeTestCase that registers a listener on the node's ClusterService (getInstanceFromNode(ClusterService.class).addListener(...)), counts down a CountDownLatch when event.indicesCreated().contains("watch-me"), creates the index via client().admin().indices().prepareCreate("watch-me").get(), and asserts the latch reaches zero within 10s. This is the deterministic version of the TRACE-log observation — the same technique Lab 4.3 formalizes.

4. (core) A ClusterStateUpdateTask that increments a counter. Get ClusterService from the test node and submit a ClusterStateUpdateTask (find the submit method: rg -n "submitStateUpdateTask|submitClusterStateUpdateTask" server/src/main/java/org/opensearch/cluster/service/ClusterService.java server/src/main/java/org/opensearch/cluster/service/ClusterManagerService.java) whose execute(currentState) returns ClusterState.builder(currentState).build() (a no-op state that still bumps the version when published). In clusterStateProcessed(...), count down a latch. Assert the latch fires and that clusterService.state().version() is strictly greater after. You have now driven the ClusterManagerService path you read in Part C.

5. (core) Prove appliers run before listeners. Register both a ClusterStateApplier (addStateApplier) and a ClusterStateListener (addListener) on the same node, each appending a token to a shared List<String> when they see a target index created. Trigger a create and assert the applier's token appears before the listener's. This codifies the ordering guarantee from the Part C table — the single most misunderstood fact in the chapter.

6. (advanced challenge) Instrument the version increment, then pin it. First, as a temporary patch, add a logger.info("computed cluster state version {}", newClusterState.version()) line at the version-bump site in ClusterManagerService (rg -n "incrementVersion|version\(.*\+ 1|patchVersions" server/src/main/java/org/opensearch/cluster/service/ClusterManagerService.java), ./gradlew run, change a setting (Part D), and capture the line. Then revert and write an OpenSearchIntegTestCase (@ClusterScope(numDataNodes = 2)) that updates a transient setting via client().admin().cluster().prepareUpdateSettings(), reads the cluster-state version on both nodes, and asserts they converge to the same version — proving publication reached every node, not just the manager. This turns the whole live demo into a regression test.

Issues to Practice On

Cluster-state and applier bugs ("setting didn't propagate", "applier threw", "listener leaked") are frequent and high-impact. Target opensearch-project/OpenSearch.

# Cluster-state / coordination / cluster-manager areas (labels move; confirm on the tracker):
gh issue list --repo opensearch-project/OpenSearch --label "Cluster Manager" --state open
gh issue list --repo opensearch-project/OpenSearch --label "Cluster:Coordination" --state open
gh issue list --repo opensearch-project/OpenSearch --label "bug" --search "cluster state apply" --state open
gh issue list --repo opensearch-project/OpenSearch --label "good first issue" --state open
gh label list --repo opensearch-project/OpenSearch | rg -i "cluster|coordination|state|manager"

Representative patterns:

• A setting/metadata change that doesn't propagate or applies inconsistently. Reproduce with a multi-node integ test (exercise 6), trace which applier/listener handles it via rg -ln "implements ClusterStateApplier|addListener" server/src/main/java, fix, and pin with an assertion that versions converge across nodes.
• A blocking/slow applier or listener stalling the applier thread. Reproduce, identify the blocking call, offload it to another thread pool, and add a test that the apply completes promptly.

Planted-bug drill. In ClusterManagerService, find the version-increment site (exercise-6 grep) and remove the increment (publish the new state with the same version as the previous one). Run ./gradlew :server:test --tests "*ClusterManagerServiceTests*" and observe which assertions break, and how the applier reacts to a non-monotonic version (it may skip the apply as "already seen"). Revert, then add an assertion (in exercise 1 or 4) that the version strictly increases — the check that would have caught it.

Etiquette: claim before working, reproduce first, every PR needs a test + CHANGELOG.md entry + DCO Signed-off-by (git commit -s). See community interaction and the Level 2 PR lab.

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.