Lab DC3: Cluster State Publish/Commit

Background

A cluster state change — PUT _cluster/settings, create index, a shard starting — does not happen by magic and does not happen instantly. It is computed on the elected cluster manager (formerly master) by ClusterManagerService from a queue of ClusterStateUpdateTasks, published to every node in a two-phase publish → commit, and applied locally by each node's ClusterApplierService. Get this path wrong in code and you cause cluster-wide hangs; understand it and you can trace any "the cluster is frozen" / "settings didn't take" / "publish timed out" problem.

This lab is the runnable companion to the deep-dive cluster-state-publishing (read it first). You will submit one change, then watch — in TRACE logs and in the REST API — the full arc:

1. the update task batches on the cluster-manager service and an executor computes a new immutable ClusterState (version+1);
2. the two-phase publish→commit (PublicationTransportHandler), including Diff-based publishing (the manager ships a diff, not the whole state);
3. the ClusterApplierService applying it (appliers, then listeners) on every node, and the ack model that decides when your REST call returns "acknowledged": true.

Why This Matters for Contributors

Most newcomer cluster-state bugs are on this path: an applier that blocks the applier thread (freezing the whole cluster), a task that does I/O in execute() (stalling all state progress), or a "acknowledged": false that a user mistakes for a failure. You will see, with your own eyes, the version increment, the diff on the wire, and the appliers running in order — which is what lets you reason about every one of those bugs. It is also the foundation for writing a custom ClusterStateListener/applier (see Level 4 lab 4.3).

Prerequisites

• OpenSearch builds; a cluster from Lab DC1 Step 2.
• You read cluster-state-publishing.
• Notes:

mkdir -p ~/opensearch-notes/dc3
: > ~/opensearch-notes/dc3/pubsub-log.md

Bring up the 3-node cluster with the service and coordination loggers on TRACE (these are the two packages that print the publish/apply arc):

cd ~/src/OpenSearch
./gradlew run -PnumNodes=3 \
  -Dtests.opensearch.logger.org.opensearch.cluster.service=TRACE \
  -Dtests.opensearch.logger.org.opensearch.cluster.coordination=TRACE \
  | tee ~/opensearch-notes/dc3/run.out

(If that flag form isn't honored, set the loggers at runtime as in Lab DC1 Step 2.)

Step-by-Step Tasks

Step 1 — Learn the log sites before you trigger a change

cd ~/src/OpenSearch

# Where the cluster-manager service runs the batch and publishes:
grep -n "runTasks\|executeTasks\|TaskBatcher\|publish\|cluster state updated, version" \
  server/src/main/java/org/opensearch/cluster/service/ClusterManagerService.java | head

# The two-phase transport handlers (publish then commit):
grep -n "PUBLISH_STATE_ACTION_NAME\|COMMIT_STATE_ACTION_NAME\|handlePublishRequest\|handleApplyCommit\|sendApplyCommit\|serializeFullClusterState\|serializeDiffClusterState" \
  server/src/main/java/org/opensearch/cluster/coordination/PublicationTransportHandler.java | head

# The applier side: appliers run before listeners:
grep -n "callClusterStateAppliers\|callClusterStateListeners\|applyChanges\|cluster state applied" \
  server/src/main/java/org/opensearch/cluster/service/ClusterApplierService.java | head

Note the three packages that print, in order: ClusterManagerService (compute+publish) → PublicationTransportHandler (wire, with Diff) → ClusterApplierService (apply, ack).

Step 2 — Snapshot the current version, then submit a change

The cluster-state version increments on every committed publish. Grab it first:

curl -s 'localhost:9200/_cluster/state?filter_path=version,cluster_uuid&pretty'
# { "cluster_uuid":"...", "version": 12 }

Now submit the simplest possible change — a persistent setting (one task, one publish):

curl -s -XPUT 'localhost:9200/_cluster/settings?pretty' \
  -H 'content-type: application/json' -d '{
    "persistent": { "indices.recovery.max_bytes_per_sec": "60mb" }
  }'
# { "acknowledged": true, "persistent": { "indices": { "recovery": { "max_bytes_per_sec": "60mb" }}}, ... }

Check the version bumped by one:

curl -s 'localhost:9200/_cluster/state?filter_path=version&pretty'
# { "version": 13 }

cat >> ~/opensearch-notes/dc3/pubsub-log.md <<'EOF'
## One change → one version bump
- before: version 12
- PUT _cluster/settings indices.recovery.max_bytes_per_sec=60mb -> acknowledged:true
- after: version 13   (exactly +1: one task, one published state)
EOF

Step 3 — Read the compute step in the manager log

On the cluster-manager node's log, find the task→executor→new-state lines:

LCM=$(find ~/src/OpenSearch -path "*testclusters*" -name "*.log" \
  | xargs grep -l "cluster-manager node changed\|elected-as" 2>/dev/null | head -1)
echo "manager log: $LCM"

grep -nE "cluster state update task \[cluster_update_settings\]|executing cluster state update|took .* to compute cluster state update|publishing cluster state version \[13\]|cluster state updated, version \[13\]" \
  "$LCM" | head

Typical arc (your strings vary):

[TRACE][o.o.c.s.ClusterManagerService] [cm] executing cluster state update for [cluster_update_settings]
[DEBUG][o.o.c.s.ClusterManagerService] [cm] took [3ms] to compute cluster state update for [cluster_update_settings]
[TRACE][o.o.c.s.ClusterManagerService] [cm] cluster state updated, version [13], source [cluster_update_settings]
[TRACE][o.o.c.s.ClusterManagerService] [cm] publishing cluster state version [13]

The executor for settings is SettingsUpdater/the settings update task; for an index create it's the metadata-create executor. The point: a pure function mapped the old state (v12) to a new immutable state (v13).

Step 4 — Watch the two-phase publish and the Diff on the wire

grep -nE "publish.*version \[13\]|PublishRequest|sending full cluster state|sending cluster state diff|received diff cluster state|applying cluster state diff|ApplyCommit|committing version \[13\]" \
  "$LCM" | head
# Then a follower's log:
LF=$(find ~/src/OpenSearch -path "*testclusters*runTask-1*" -name "*.log" | head -1)
grep -nE "received|diff|full cluster state|applying|version \[13\]" "$LF" | head

What you are looking for, mapped to phases:

Why a diff? Cluster state can be large (thousands of indices/mappings). Shipping the whole thing on every change would be enormous. The manager sends a Diff against the version the follower already holds; only if the follower is behind (missed a version) does it fall back to a full state. Confirm:

grep -n "incompatible-clusters\|fallback to full|serializeFullClusterState\|Diff<ClusterState>" \
  server/src/main/java/org/opensearch/cluster/coordination/PublicationTransportHandler.java | head

Step 5 — Watch the applier→listener ordering and the ack

On any node, the apply step runs appliers first, then listeners, synchronously on one thread:

grep -nE "applying.*version \[13\]|callClusterStateAppliers|callClusterStateListeners|cluster state applied|set local cluster state to version 13" \
  "$LF" | head

The ack is what made your curl return "acknowledged": true: the publication carries an AckListener, and the REST response is held until enough nodes ack the commit (or the cluster_manager_timeout elapses).

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

Demonstrate the timeout dimension directly — ask for a tiny manager timeout and see the response field reflect ack timing (still committed, just maybe not acked in time):

curl -s -XPUT 'localhost:9200/_cluster/settings?cluster_manager_timeout=1ms&pretty' \
  -H 'content-type: application/json' -d '{ "persistent": { "indices.recovery.max_bytes_per_sec": "55mb" } }'
# On a fast local cluster it will still say acknowledged:true; the field MEANS
# "committed AND acked by all nodes within cluster_manager_timeout", not "succeeded".

cat >> ~/opensearch-notes/dc3/pubsub-log.md <<'EOF'
## Apply + ack
- follower log: callClusterStateAppliers BEFORE callClusterStateListeners (one thread)
- acknowledged:true == committed AND acked by all nodes within cluster_manager_timeout
- acknowledged:false would mean committed but not acked in time — NOT a failure
EOF

Step 6 — Prove batching with many tasks at once

A busy cluster does not publish a state per event; the manager batches compatible tasks into one published state. Trigger many shard events at once by creating an index with several shards and replicas:

curl -s -XPUT 'localhost:9200/dc3-batch?pretty' -H 'content-type: application/json' -d '{
  "settings": { "number_of_shards": 5, "number_of_replicas": 1 }
}'

# Watch the manager log: many shard-started events collapse into far fewer published versions.
grep -nE "shard-started|started shard|cluster state update.*shard-started|version \[" "$LCM" | tail -30

# Count distinct published versions vs shard-started events in a short window:
grep -cE "shard-started|started shard" "$LCM"
grep -cE "publishing cluster state version" "$LCM"

You should see the shard-started count exceed the published-version count — that gap is TaskBatcher coalescing many ShardStartedClusterStateTaskExecutor tasks into single publishes.

grep -n "TaskBatcher\|ShardStartedClusterStateTaskExecutor\|batchedTask\|executeTasks" \
  server/src/main/java/org/opensearch/cluster/service/ClusterManagerService.java \
  server/src/main/java/org/opensearch/cluster/action/shard/ShardStateAction.java 2>/dev/null | head

Step 7 — Observe a slow-applier publish timeout (optional, instructive)

The deep-dive warns: a slow applier on a follower blows cluster.publish.timeout and can make the manager step down. You can approach this safely by lowering the timeout and watching publishes get tighter (don't make it pathological on a shared box):

curl -s -XPUT 'localhost:9200/_cluster/settings' -H 'content-type: application/json' -d '{
  "persistent": { "cluster.publish.timeout": "200ms" }
}'
# Then create/delete a few indices and watch for "timed out waiting for ..." in logs:
grep -nE "timed out|publication ... did not complete|failed to publish cluster state" "$LCM" | tail
# Reset:
curl -s -XPUT 'localhost:9200/_cluster/settings' -H 'content-type: application/json' -d '{
  "persistent": { "cluster.publish.timeout": null }
}'

Deliverables

• ~/opensearch-notes/dc3/pubsub-log.md with: the +1 version bump for one task; the compute→publish→apply arc mapped to ClusterManagerService / PublicationTransportHandler / ClusterApplierService; evidence of a Diff (not full) on the wire; the applier-before-listener ordering; and the batching count (shard-started events > published versions).
• The exact meaning of "acknowledged" written in your own words.
• (Optional) The publish-timeout observation.

Troubleshooting

Expected Output

One PUT _cluster/settings → "acknowledged": true and the cluster-state version incremented by exactly one. The manager log shows compute → publishing version [N], the wire shows a diff to up-to-date followers, each follower's log shows appliers running before listeners, and the REST call returned only after the ack condition was met. Creating a multi-shard index shows many shard-started events collapsing into fewer published versions (batching).

Stretch Goals

• curl -s 'localhost:9200/_cluster/health?wait_for_events=languid&pretty' and explain what "wait for the cluster-manager queue to drain at LANGUID priority" proves about the update-task queue.
• Read ClusterStateUpdateTask.execute for the settings task and confirm it is a pure function (no I/O, returns a new immutable state). grep -rn "class SettingsUpdater\|updateSettings\|ClusterState execute" server/src/main/java/org/opensearch/cluster/ | head
• Find IndicesClusterStateService.applyClusterState — the biggest applier — and show how the v13→v… routing change turns into shard create/start/close on the local node. Link cluster-state-publishing.
• Write (or read) a ClusterManagerServiceTests case that submits N tasks with one executor and asserts one published state: ./gradlew :server:test --tests "org.opensearch.cluster.service.ClusterManagerServiceTests".

Coding Exercises

You watched compute → publish → apply → ack with curl and TRACE; now make each phase an assertion. Locate every class with rg before you touch it (rg --files -g 'ClusterManagerServiceTests.java' server/src/test); never trust a stale line number.

1. (warm-up) Assert "one task → one version bump" in a unit test. Find ClusterManagerServiceTests (rg -l "class ClusterManagerServiceTests"). Add a test that submits a single ClusterStateUpdateTask, captures the published state, and asserts its version() is exactly the previous version + 1. Use the in-process ClusterManagerService harness already in that file (no real cluster). Verify: ./gradlew :server:test --tests "org.opensearch.cluster.service.ClusterManagerServiceTests".

2. (core) Prove batching: N tasks, fewer published states. Build on Step 6. In ClusterManagerServiceTests, submit N tasks that share one executor and assert the number of distinct published ClusterState versions is strictly less than N (ideally 1 if they batch into a single run). Read how the existing batching tests count publications (rg -n "submitStateUpdateTask|TaskBatcher" server/src/test/java/org/opensearch/cluster/service/ClusterManagerServiceTests.java) and reuse that counting scaffold. This encodes the TaskBatcher coalescing you observed.

3. (core) Assert applier-before-listener ordering. In Step 5 you read that appliers run before listeners on one thread. Find ClusterApplierServiceTests (rg -l "class ClusterApplierServiceTests") and add a test that registers a ClusterStateApplier and a ClusterStateListener, both appending to a shared List, then asserts the applier's entry precedes the listener's for the same state version — and that both ran on the same named applier thread (assert via Thread.currentThread().getName()).

4. (core) Assert the Diff path on the wire. The manager ships a Diff, not the full state, to up-to-date followers (Step 4). Locate the serialization seam (rg -n "serializeDiffClusterState|serializeFullClusterState" server/src/main/java/org/opensearch/cluster/coordination/PublicationTransportHandler.java) and find the test that exercises it (rg -l "PublicationTransportHandlerTests"). Add a case proving that when a follower is at version V, the handler sends a diff for V+1, but a follower that is behind triggers the full-state fallback. Assert on the chosen BytesReference/branch, not on log strings.

5. (core) A ClusterStatePublisher ack assertion (integ test). Write an OpenSearchIntegTestCase that submits an AckedClusterStateUpdateTask and asserts the REST/transport call returns acknowledged only after the publication's AckListener saw acks from all nodes. Locate the publisher seam (rg -n "interface ClusterStatePublisher|ackListener|AckListener" server/src/main/java/org/opensearch/cluster/coordination/) and assert that with a deliberately tiny cluster_manager_timeout the state still commits (version bumps on a follower) even when acknowledged is false — proving false ≠ failure.

6. (advanced) Build a custom applier and prove it cannot stall the cluster. Advanced challenge: implement a ClusterStateApplier that does real work, then add a cluster.publish.timeout-driven integ test that proves a slow applier on a follower blows the publish timeout and is visible on the manager (search for the timeout site with rg -n "PUBLISH_TIMEOUT_SETTING|did not complete" server/src/main). Then fix your applier to offload its work off the applier thread and show the timeout no longer fires. This is the runnable companion to Level 4 lab 4.3: the deliverable is one integ test that goes red with the blocking applier and green with the fixed one.

Issues to Practice On

Cluster-state publish/apply bugs are where newcomers cause (and fix) cluster-wide hangs. Hunt them on opensearch-project/OpenSearch (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 "good first issue" --state open
gh issue list --repo opensearch-project/OpenSearch --label "bug" --state open --search "applier OR publish OR cluster state OR acknowledged"
gh label list --repo opensearch-project/OpenSearch | grep -iE "cluster|performance|flaky"

Representative patterns. (1) A "settings update returns acknowledged:false, users think it failed" report: reproduce by tightening cluster_manager_timeout, show the state did commit, and improve docs/messaging — or fix the genuine slow node. (2) An applier that blocks the applier thread and freezes the cluster: reproduce with a slow ClusterStateApplier, locate the offending applier via rg, move its work off the thread, and add a publish-timeout regression test. Arc: reproduce → locate via rg → fix → test → PR with CHANGELOG + DCO.

Planted-bug drill. In ClusterManagerService, find where the new state's version is set (rg -n "incrementVersion|version() + 1|builder().version" server/src/main/java/org/opensearch/cluster/service/ClusterManagerService.java) and change it to not increment (publish the same version twice). Run ./gradlew :server:test --tests "org.opensearch.cluster.service.ClusterManagerServiceTests" and watch which test catches the stuck version. Revert, then add the explicit version == previous + 1 assertion from Exercise 1 so any future regression here goes red immediately.

Etiquette: claim the issue first, reproduce before theorising, and every PR ships a test + CHANGELOG.md entry + DCO Signed-off-by (git commit -s). See community-interaction.

Validation / Self-check

Cite a class or log line for each:

1. Why did one PUT _cluster/settings bump the version by exactly one? On which node was the new state computed, and on which node(s) was it applied?
2. Distinguish phase 1 (publish) from phase 2 (commit) from the logs. At which phase has a follower applied the state? Why is a quorum of accepts required before commit?
3. What is a Diff here, and why does the manager send it instead of the full state? When does it fall back to a full state?
4. Prove from the log that all appliers ran before any listener. What stalls for the whole cluster if an applier blocks?
5. What does "acknowledged": true actually mean, and what governs "acknowledged": false? Why is false not a failure?
6. From the batching experiment, how many shard-started events did you see vs how many published versions? What class coalesced them, and why does batching matter for a busy cluster?

When you can narrate a single setting change from curl → ClusterManagerService.execute → diff-publish → ClusterApplierService apply → ack, and explain the version bump and the ack semantics, you've completed Lab DC3. Continue to Lab DC4: Allocation and Rebalancing Under Churn.