Stage 4 — Cluster State and Coordination Bugs

What this stage teaches

Stage 4 is the first stage that touches the distributed core. The skill is reasoning about ClusterState — the single immutable snapshot of cluster metadata — and the machinery that mutates and applies it:

• Cluster state update tasks (ClusterStateUpdateTask, ClusterStateTaskExecutor) that run on the cluster manager (formerly master) and compute a new state from the old one. Getting a no-op wrong here is a classic bug: returning a new but equal state forces an unnecessary publish; mutating in place is a correctness disaster.
• Appliers and listeners (ClusterStateApplier, ClusterStateListener) that react to a committed state on every node. A listener that NPEs on a missing index or assumes a shard exists is the other classic bug.
• The two services on the cluster-manager node — ClusterManagerService (computes states) and ClusterApplierService (applies them) — tied together by ClusterService.

You will write tests against this machinery without a full cluster, using ClusterServiceUtils (a fake ClusterService you drive by hand) and OpenSearchSingleNodeTestCase (one real in-JVM node). The bugs are subtle but the surface area per fix stays small.

Prerequisite: Stage 3, plus the cluster-state deep dives: Cluster state and Cluster state publishing. Read those first — this stage summarises, it does not re-teach them.

The model in one diagram

flowchart TD
  subgraph CM[Cluster-manager node]
    UT[ClusterStateUpdateTask.execute<br/>oldState -> newState] --> MS[ClusterManagerService<br/>batches tasks, computes ClusterState]
    MS --> PUB[Publish two-phase:<br/>send -> commit]
  end
  PUB --> ALL[Every node]
  subgraph N[Each node]
    ALL --> CAS[ClusterApplierService.onNewClusterState<br/>-> applyChanges -> callClusterStateAppliers]
    CAS --> AP[ClusterStateApplier.applyClusterState]
    CAS --> LI[ClusterStateListener.clusterChanged]
  end

Two invariants you must never break:

1. ClusterState is immutable. An update task receives currentState and must return either the same instance (a no-op) or a newly built instance via ClusterState.builder(currentState)....build(). Never mutate Metadata, RoutingTable, or any sub-object of the current state.
2. A no-op returns the identical instance. If your task computes that nothing changed, return currentState; (the same object). ClusterManagerService uses identity/equals to decide whether to publish. Returning a freshly-built equal state triggers a needless cluster-wide publish — a real performance bug that shows up as "cluster-manager is busy."

Finding Stage 4 issues

is:issue is:open label:bug no:assignee label:"Cluster Manager"
is:issue is:open label:bug no:assignee "cluster state" in:title,body
is:issue is:open label:bug no:assignee "NullPointerException" "cluster" in:body
is:issue is:open label:"help wanted" no:assignee "ClusterStateUpdateTask" in:body

The component label has been spelled Cluster Manager, cluster-manager, and historically master — check the current label list. Coordination-layer bugs may carry distributed framework or a Coordination area label.

Fallback grep — listeners and appliers that dereference an index/shard without a null guard:

# Listeners that look up an index by name and may get null:
grep -rn "metadata().index(" server/src/main/java/org/opensearch/ \
  | grep -i "listener\|applier" | head
# clusterChanged implementations:
grep -rln "implements ClusterStateListener" server/src/main/java/org/opensearch/

Walked example — a listener that NPEs on a missing index

Illustrative of the pattern. The grep finds the real listener; do not trust the line numbers below.

Symptom: an issue reports a NullPointerException from a ClusterStateListener when an index is deleted between the state that scheduled some work and the state the listener runs against. The listener does event.state().metadata().index(name).getSettings() and index(name) returns null because the index is gone.

Locate the listener

grep -rn "implements ClusterStateListener" server/src/main/java/org/opensearch/ | head
# Then in the suspect file, find the unguarded lookup:
grep -n "metadata().index(" server/src/main/java/org/opensearch/<path>/SomeService.java
git log --oneline -n 5 -- server/src/main/java/org/opensearch/<path>/SomeService.java
git blame -L <start>,<end> server/src/main/java/org/opensearch/<path>/SomeService.java

The offending code:

@Override
public void clusterChanged(ClusterChangedEvent event) {
    for (String name : trackedIndices) {
        IndexMetadata meta = event.state().metadata().index(name);   // may be null after delete
        Settings s = meta.getSettings();                              // NPE here
        // ...
    }
}

Diff

--- a/server/src/main/java/org/opensearch/<path>/SomeService.java
+++ b/server/src/main/java/org/opensearch/<path>/SomeService.java
@@
     public void clusterChanged(ClusterChangedEvent event) {
         for (String name : trackedIndices) {
-            IndexMetadata meta = event.state().metadata().index(name);
-            Settings s = meta.getSettings();
+            IndexMetadata meta = event.state().metadata().index(name);
+            if (meta == null) {
+                // Index was deleted between scheduling and this applier run; stop tracking it.
+                trackedIndices.remove(name);
+                continue;
+            }
+            Settings s = meta.getSettings();
             // ...
         }
     }

Three things to notice:

1. The fix is defensive, not clever. Concurrency between the publishing of states means a listener must treat every lookup into the new state as possibly absent. The index you saw last round can be gone this round.
2. Clean up your own tracking. Removing the stale name prevents the NPE and a slow leak of dead index names in trackedIndices.
3. ClusterChangedEvent gives you both states. If you need to know what changed, use event.indicesDeleted(), event.indicesCreated(), event.metadataChanged() rather than diffing by hand.

Test with ClusterServiceUtils — no real cluster

You can drive cluster-state changes by hand using a test ClusterService:

public void testListenerSurvivesIndexDeletion() {
    ThreadPool threadPool = new TestThreadPool(getTestName());
    try {
        ClusterService clusterService = ClusterServiceUtils.createClusterService(threadPool);
        SomeService service = new SomeService(clusterService /*, deps */);

        // State A: index "foo" exists and is tracked.
        ClusterState withFoo = ClusterState.builder(clusterService.state())
            .metadata(Metadata.builder().put(
                IndexMetadata.builder("foo")
                    .settings(settings(Version.CURRENT))
                    .numberOfShards(1).numberOfReplicas(0)))
            .build();
        ClusterServiceUtils.setState(clusterService, withFoo);
        service.startTracking("foo");

        // State B: "foo" is deleted. The applier must not throw.
        ClusterState withoutFoo = ClusterState.builder(withFoo)
            .metadata(Metadata.builder(withFoo.metadata()).remove("foo"))
            .build();
        ClusterServiceUtils.setState(clusterService, withoutFoo);   // fires clusterChanged

        // No NPE, and the name is no longer tracked.
        assertThat(service.trackedIndices(), not(hasItem("foo")));
    } finally {
        // terminate the thread pool
        terminate(threadPool);
    }
}

ClusterServiceUtils.setState(...) publishes the new state to the listeners synchronously, so the assertion runs after clusterChanged. This is the standard way to unit-test cluster-state reactions without spinning up InternalTestCluster.

For a test that needs a real index lifecycle (create, then delete, then assert the service recovered), step up to OpenSearchSingleNodeTestCase:

public class SomeServiceIT extends OpenSearchSingleNodeTestCase {
    public void testTrackingSurvivesDeleteOnRealNode() throws Exception {
        createIndex("foo");
        // ... trigger tracking ...
        assertAcked(client().admin().indices().prepareDelete("foo").get());
        assertBusy(() -> assertThat(serviceUnderTest().trackedIndices(), not(hasItem("foo"))));
    }
}

assertBusy (not Thread.sleep) waits for the asynchronous applier to run — a habit Stage 9 drills hard.

Build and PR

./gradlew :server:test --tests "*SomeServiceTests" --tests "*SomeServiceIT" -q
./gradlew :server:precommit

CHANGELOG under ### Fixed, git commit -s, push, open the PR. Mention in the description which two states trigger the race (delete-between-publish) so the reviewer can reason about it without re-deriving it.

The no-op variant of the bug

The other half of this stage is the update task that does too much. A task that should be a no-op but rebuilds the state anyway floods publishing. The fix:

--- a/server/src/main/java/org/opensearch/<path>/SomeUpdateTask.java
+++ b/server/src/main/java/org/opensearch/<path>/SomeUpdateTask.java
@@
     public ClusterState execute(ClusterState currentState) {
-        Metadata.Builder md = Metadata.builder(currentState.metadata());
-        md.put(updatedIndexMetadata, true);
-        return ClusterState.builder(currentState).metadata(md).build();
+        IndexMetadata existing = currentState.metadata().index(indexName);
+        if (existing != null && existing.equals(updatedIndexMetadata)) {
+            return currentState;   // no change — same instance, no publish
+        }
+        Metadata.Builder md = Metadata.builder(currentState.metadata());
+        md.put(updatedIndexMetadata, true);
+        return ClusterState.builder(currentState).metadata(md).build();
+    }

Test it by asserting instance identity:

public void testNoOpReturnsSameState() {
    ClusterState before = /* state already containing updatedIndexMetadata */;
    ClusterState after = new SomeUpdateTask(updatedIndexMetadata).execute(before);
    assertSame(before, after);   // not assertEquals — must be the *same* object
}

assertSame is the whole point: ClusterManagerService keys publishing off identity, so a no-op must return the same reference.

Walked example — batching: many tasks, one published state

The no-op invariant only matters because ClusterManagerService is a batching service: it rarely runs one task and publishes one state. It runs a pile of tasks that share an executor, folds them into a single new ClusterState, and publishes once. Understanding this coalescing is the difference between a fix that scales and one that melts the cluster manager under load.

Read the real path; the class names below are exact, the walkthrough is the mechanism — no line numbers, run the greps.

The batching key is the executor instance

When you submit an update, you hand ClusterManagerService a (task, config, executor, listener) tuple. The batching rule is one line of Javadoc worth memorising:

grep -n "submitted while there" \
  server/src/main/java/org/opensearch/cluster/service/ClusterManagerService.java

Tasks submitted while others are pending for the same executor are executed together in a single batch. Under the hood TaskBatcher groups pending work by batching key, and for the cluster manager the batching key is the executor object:

grep -n "tasksPerBatchingKey\|batchingKey" \
  server/src/main/java/org/opensearch/cluster/service/TaskBatcher.java
grep -n "ClusterStateTaskExecutor<Object> taskExecutor\|batchingKey" \
  server/src/main/java/org/opensearch/cluster/service/ClusterManagerService.java

That is why the interface takes a list, not a single task:

grep -n "ClusterTasksResult<T> execute(ClusterState currentState, List<T> tasks)" \
  server/src/main/java/org/opensearch/cluster/ClusterStateTaskExecutor.java

// ClusterStateTaskExecutor<T>
ClusterTasksResult<T> execute(ClusterState currentState, List<T> tasks) throws Exception;

One currentState in, one resultingState out — for the whole batch.

Trace one batch through the service

grep -n "executeTasks\|calculateTaskOutputs\|patchVersions\|clusterStateUnchanged\|publish(" \
  server/src/main/java/org/opensearch/cluster/service/ClusterManagerService.java

The path, in order:

1. executeTasks maps every pending task to its payload, builds one List, and calls executor.execute(previousClusterState, inputs) once for the whole batch.
2. patchVersions bumps the cluster-state version only if the executor returned a different instance (previousClusterState != newClusterState) — the same identity check as the no-op invariant above, now applied to the whole batch.
3. clusterStateUnchanged() is literally previousClusterState == newClusterState. If the batch changed nothing, the service skips publishing and only notifies listeners.
4. Otherwise there is exactly one publish(...) for the batch, no matter how many tasks it held.

That is the coalescing: N tasks → one recompute → at most one published state.

A concrete executor: N shard-started events → one reroute

The clearest real example is shard-started processing. Each time a shard finishes recovery its node sends a shard-started event; on a large recovery the cluster manager can receive hundreds in a burst. They all share ShardStartedClusterStateTaskExecutor, so they batch:

grep -n "public ClusterTasksResult<StartedShardEntry> execute\|applyStartedShards\|seenShardRoutings" \
  server/src/main/java/org/opensearch/cluster/action/shard/ShardStateAction.java

Inside execute, the executor walks the whole List<StartedShardEntry>, drops duplicates and already-started shards (via seenShardRoutings), collects the survivors into one shardRoutingsToBeApplied, and calls the allocator once:

maybeUpdatedState = allocationService.applyStartedShards(currentState, shardRoutingsToBeApplied);

Two hundred shard-started events become one applyStartedShards call, one rerouted ClusterState, one publish — not two hundred publishes. The follow-up reroute is scheduled from clusterStatePublished after that single publish, not per task.

Why this matters for your fix

• A per-task publish is the bug. If you write an executor whose execute triggers a publish per element (e.g. by submitting a fresh task from inside the loop), you have defeated batching and reintroduced the publish storm.
• Return the same instance when the batch is a no-op. If a list of tasks all cancel out, return currentState; — the identity check in patchVersions/clusterStateUnchanged then suppresses the publish for the entire batch.
• describeTasks and the throttling key are batch-aware. describeTasks(List<T>) builds the log summary for the whole batch, and getClusterManagerThrottlingKey() lets the cluster manager throttle a flood of same-executor tasks. Override them when you add an executor; a reviewer will ask.

Test a batch the way you tested the no-op — assert on the single resulting state:

public void testBatchCoalescesToOneState() throws Exception {
    ClusterState before = /* two initializing shards for index "idx" */;
    List<StartedShardEntry> batch = List.of(startedEntry(shard0), startedEntry(shard1));
    ClusterTasksResult<StartedShardEntry> result = executor.execute(before, batch);
    assertThat(result.executionResults.size(), equalTo(2));   // both tasks accounted for
    assertNotSame(before, result.resultingState);             // the batch did change the state
}

Pitfalls

• Mutating the current state. Any currentState.metadata().getIndices().put(...) is a bug even if it "works" in a test — other components hold the same immutable reference. Always go through a Builder.
• Returning a fresh-but-equal state for a no-op. Use assertSame in the test to catch this. It is the most common publish-storm bug.
• Assuming an index/shard from the previous state still exists. Every lookup into the applied state can be null. Guard it; consume ClusterChangedEvent.indicesDeleted().
• Doing slow work on the applier thread. applyClusterState runs on the cluster applier thread; blocking it stalls the whole node's view of the cluster. Hand heavy work to a thread pool — see threadpools & concurrency.
• Testing a race with Thread.sleep. It is flaky by construction. Use assertBusy with a tight assertion, or drive state synchronously with ClusterServiceUtils.setState.
• Forgetting the cluster-manager-only context. Update tasks run only on the elected cluster manager. Logic that must run on every node belongs in an applier/listener, not a task. Confusing the two is a design bug a reviewer will flag immediately.

Exit criteria — when you're ready for Stage 5

• One cluster-state fix is merged with a ClusterServiceUtils-based unit test (and, if the behaviour needs a real index, an OpenSearchSingleNodeTestCase).
• You can state the immutability and no-op-identity invariants without looking them up, and you used assertSame to defend the no-op.
• You can read a ClusterChangedEvent and use its indicesCreated/Deleted/metadataChanged helpers instead of diffing two states by hand.
• You know which logic belongs in an update task (cluster manager) versus an applier (every node), and why.

You now understand how state is mutated and applied. Stage 5 reads one specific part of that state — the RoutingTable — and the deciders that compute it.