Lab DC4: Allocation and Rebalancing Under Churn

Background

The control plane decides what shards exist; the data plane decides where they live. That placement is computed by AllocationService — wrapping the BalancedShardsAllocator (the balancer), the AllocationDeciders (the YES/NO/THROTTLE gate), and GatewayAllocator (placing existing shards on recovery). Every time a node joins or leaves, an index is created, or a shard fails, the cluster reroutes: it recomputes the routing table, moves shards, and recovers them — all of it published through the same consensus you traced in Lab DC3.

This lab makes you drive that machinery under churn and read the result. You will:

1. create indices with replicas and watch them allocate across nodes;
2. add a node and watch rebalancing relocate shards to flatten load;
3. remove a node and watch replicas reallocate + recover;
4. deliberately force an UNASSIGNED shard and diagnose it with _cluster/allocation/explain;
5. nudge allocation with _cluster/reroute;
6. observe recovery throttling and map every behavior back to the real classes.

This is the data-plane counterpart to the consensus labs. The mechanics live in the deep-dives shard-allocation and recovery; the routing model is in the masterclass sharding-routing. Here you operate it and diagnose it.

Note: "cluster manager" is the role formerly called master; allocation runs on the elected cluster manager as part of computing each new cluster state.

Why This Matters for Contributors

"Shards stuck unassigned," "cluster yellow forever," "rebalancing won't stop," "recovery is too slow" — these are among the most common operational issues, and fixing or even reproducing them requires reading _cluster/allocation/explain and knowing which decider said NO/THROTTLE. Contributors who touch allocation must be able to induce these states and diagnose them from the explain API down to the decider. This lab is that loop.

Prerequisites

• OpenSearch builds; you can run a multi-node cluster (./gradlew run -PnumNodes=3).
• You read the intensive data-plane section.
• Notes:

mkdir -p ~/opensearch-notes/dc4

Find the allocation classes you'll map behavior to:

cd ~/src/OpenSearch
find server/src/main/java/org/opensearch/cluster/routing/allocation -name "*.java" | sort | head -40
grep -rln "extends AllocationDecider" \
  server/src/main/java/org/opensearch/cluster/routing/allocation/decider/ | sort
grep -n "class BalancedShardsAllocator\|weight\|balance\|relocate" \
  server/src/main/java/org/opensearch/cluster/routing/allocation/allocator/BalancedShardsAllocator.java | head

Bring up 3 nodes:

./gradlew run -PnumNodes=3 | tee ~/opensearch-notes/dc4/run.out

Step-by-Step Tasks

Step 1 — Create indices with replicas; watch allocation

# An index with 6 primaries + 1 replica each = 12 shards to place across 3 nodes.
curl -s -XPUT 'localhost:9200/dc4-a?pretty' -H 'content-type: application/json' -d '{
  "settings": { "number_of_shards": 6, "number_of_replicas": 1 }
}'

# Where did they land? (one row per shard copy)
curl -s 'localhost:9200/_cat/shards/dc4-a?v&h=index,shard,prirep,state,node'
# index  shard prirep state   node
# dc4-a  0     p      STARTED  runTask-0
# dc4-a  0     r      STARTED  runTask-1
# dc4-a  1     p      STARTED  runTask-2
# ...

# Health: green means every primary AND replica is STARTED.
curl -s 'localhost:9200/_cluster/health/dc4-a?pretty' \
  | grep -E 'status|active_shards|unassigned_shards|relocating'

Observe: primaries and replicas are spread so that no primary shares a node with its own replica (SameShardAllocationDecider), and shard counts are roughly even across the 3 nodes (BalancedShardsAllocator).

# shard count per node — should be roughly balanced (12 / 3 = 4 each):
curl -s 'localhost:9200/_cat/allocation?v&h=node,shards,disk.percent'

cat >> ~/opensearch-notes/dc4/notes.md <<'EOF'
## Initial allocation (3 nodes, 12 shards)
- ~4 shards/node (BalancedShardsAllocator)
- no shard's primary and replica on the same node (SameShardAllocationDecider)
- status: green
EOF

Step 2 — Add a node; watch rebalancing relocate shards

Start a 4th node so the balancer wants to move ~3 shards onto it (12 / 4 = 3 each):

# In a separate shell, add a node to the running gradlew cluster is fiddly;
# the simplest reproducible path is to restart with numNodes=4:
./gradlew --stop
./gradlew run -PnumNodes=4 | tee -a ~/opensearch-notes/dc4/run.out
# Re-create dc4-a if it didn't persist:
curl -s -XPUT 'localhost:9200/dc4-a' -H 'content-type: application/json' -d '{"settings":{"number_of_shards":6,"number_of_replicas":1}}' >/dev/null

# Watch RELOCATING shards as the balancer flattens onto the 4th node:
watch -n1 "curl -s 'localhost:9200/_cat/shards/dc4-a?h=shard,prirep,state,node' | sort | grep -c RELOCATING"
# and health shows relocating_shards > 0 transiently:
curl -s 'localhost:9200/_cluster/health/dc4-a?pretty' | grep -E 'relocating|status'

Note: Adding a node to an already-running gradlew run cluster is awkward; for a live add/remove without restarts, use the InternalTestCluster path in Step 7. The REST observations are identical either way.

After it settles, the per-node shard count should be ~3 each:

curl -s 'localhost:9200/_cat/allocation?v&h=node,shards'

The relocations are BalancedShardsAllocator lowering its weight function (per-node weight ~ shard count + index skew + load) — confirm the weight knobs:

grep -n "INDEX_BALANCE_FACTOR\|SHARD_BALANCE_FACTOR\|THRESHOLD\|weight(" \
  server/src/main/java/org/opensearch/cluster/routing/allocation/allocator/BalancedShardsAllocator.java | head

Step 3 — Remove a node; watch replicas reallocate + recover

Stop one node (simulate a leave). Replicas that lived there go UNASSIGNED, then recover from their primaries onto surviving nodes:

# Identify a non-manager node and stop it (or kill its pid as in Lab DC1).
curl -s 'localhost:9200/_cat/nodes?v&h=name,cluster_manager'
# kill -9 <pid-of-a-non-manager node>   # via the *.pid files (Lab DC1 Step 6)

# Immediately: health goes yellow (replicas missing), then back to green as they recover.
curl -s 'localhost:9200/_cluster/health/dc4-a?pretty' \
  | grep -E 'status|unassigned_shards|initializing_shards'

# Watch the recovery (INITIALIZING → STARTED) of the displaced replicas:
curl -s 'localhost:9200/_cat/recovery/dc4-a?v&active_only=true&h=index,shard,type,stage,source_node,target_node,bytes_percent'

Losing a node with replicas elsewhere → yellow then green. Losing the only copy of a primary → red.

Step 4 — Force an UNASSIGNED shard and diagnose it

Make a shard impossible to allocate, then read why with the explain API. The cleanest way is to demand more replicas than there are nodes to hold them:

# 4 nodes → ask for 5 copies (1 primary + 4 replicas). One replica per shard can't be
# placed without violating SameShardAllocationDecider → UNASSIGNED.
curl -s -XPUT 'localhost:9200/dc4-stuck?pretty' -H 'content-type: application/json' -d '{
  "settings": { "number_of_shards": 1, "number_of_replicas": 4 }
}'

curl -s 'localhost:9200/_cluster/health/dc4-stuck?pretty' | grep -E 'status|unassigned'
# status: yellow, unassigned_shards: 1

# THE diagnostic tool — explain the unassigned shard:
curl -s 'localhost:9200/_cluster/allocation/explain?pretty' -H 'content-type: application/json' -d '{
  "index": "dc4-stuck", "shard": 0, "primary": false
}'

Read the response fields that matter:

For the over-replicated case you'll see, on every candidate node, a NO from same_shard ("a copy of this shard is already on node X"). That decider name maps straight to SameShardAllocationDecider:

grep -n "same_shard\|class SameShardAllocationDecider\|a copy of this shard is already" \
  server/src/main/java/org/opensearch/cluster/routing/allocation/decider/SameShardAllocationDecider.java | head

cat >> ~/opensearch-notes/dc4/notes.md <<'EOF'
## Forced UNASSIGNED + explain
- dc4-stuck: 1p/4r on 4 nodes → 1 replica cannot be placed
- explain: can_allocate=no; every node's same_shard decider = NO
- maps to SameShardAllocationDecider
EOF

Now induce a different cause — a disk watermark — and re-explain to see a different decider win:

# Drop the high watermark very low so DiskThresholdDecider refuses placement.
curl -s -XPUT 'localhost:9200/_cluster/settings' -H 'content-type: application/json' -d '{
  "transient": {
    "cluster.routing.allocation.disk.threshold_enabled": true,
    "cluster.routing.allocation.disk.watermark.low": "1b",
    "cluster.routing.allocation.disk.watermark.high": "1b",
    "cluster.routing.allocation.disk.watermark.flood_stage": "1b"
  }
}'
curl -s -XPUT 'localhost:9200/dc4-disk' -H 'content-type: application/json' -d '{"settings":{"number_of_shards":1,"number_of_replicas":0}}'
curl -s 'localhost:9200/_cluster/allocation/explain?pretty' -H 'content-type: application/json' -d '{"index":"dc4-disk","shard":0,"primary":true}' \
  | grep -A3 -iE 'disk_threshold|watermark|can_allocate'
# Now reset the watermarks!
curl -s -XPUT 'localhost:9200/_cluster/settings' -H 'content-type: application/json' -d '{
  "transient": { "cluster.routing.allocation.disk.watermark.low": null,
                 "cluster.routing.allocation.disk.watermark.high": null,
                 "cluster.routing.allocation.disk.watermark.flood_stage": null }
}'

Same API, different decider (disk_threshold → DiskThresholdDecider). The explain API is how you turn "stuck shard" into "this exact decider, on these nodes, for this reason."

Step 5 — Nudge allocation with _cluster/reroute

_cluster/reroute lets you force or retry allocation. Two common moves:

# (a) Retry shards that hit MaxRetryAllocationDecider (gave up after N failures):
curl -s -XPOST 'localhost:9200/_cluster/reroute?retry_failed=true&pretty' | grep -E 'acknowledged'

# (b) Dry-run a manual move and see the deciders WITHOUT applying it:
curl -s -XPOST 'localhost:9200/_cluster/reroute?explain=true&dry_run=true&pretty' \
  -H 'content-type: application/json' -d '{
    "commands": [
      { "move": { "index": "dc4-a", "shard": 0,
                  "from_node": "runTask-0", "to_node": "runTask-1" } }
    ]
  }' | grep -E 'decisions|explanation|"decision"' | head

retry_failed=true resets the MaxRetryAllocationDecider counter — the fix for "the shard failed 5 times and the cluster stopped trying." Confirm the decider:

grep -n "MAX_RETRY\|class MaxRetryAllocationDecider\|retry_failed" \
  server/src/main/java/org/opensearch/cluster/routing/allocation/decider/MaxRetryAllocationDecider.java | head

Step 6 — Observe recovery throttling

Recovery is throttled so a node join doesn't saturate disk/network. Two knobs and the ThrottlingAllocationDecider:

# How many shards can recover at once per node, and at what byte rate:
curl -s 'localhost:9200/_cluster/settings?include_defaults=true&filter_path=**.recovery,**.node_concurrent_recoveries&flat_settings=true&pretty' \
  | grep -iE 'recover|concurrent'

# Force visible throttling: lower the limits, then add load.
curl -s -XPUT 'localhost:9200/_cluster/settings' -H 'content-type: application/json' -d '{
  "transient": {
    "indices.recovery.max_bytes_per_sec": "1mb",
    "cluster.routing.allocation.node_concurrent_recoveries": 1
  }
}'
# Bump replicas to trigger many recoveries; watch some sit in INITIALIZING/THROTTLED:
curl -s -XPUT 'localhost:9200/dc4-a/_settings' -H 'content-type: application/json' -d '{"index":{"number_of_replicas":3}}'
curl -s 'localhost:9200/_cat/recovery/dc4-a?v&active_only=true&h=shard,stage,bytes_percent,source_node,target_node'
# The explain API will show THROTTLE (not NO) for replicas waiting their turn:
curl -s 'localhost:9200/_cluster/allocation/explain?pretty' -H 'content-type: application/json' -d '{"index":"dc4-a","shard":0,"primary":false}' \
  | grep -iE 'throttl|can_allocate'
# Reset:
curl -s -XPUT 'localhost:9200/_cluster/settings' -H 'content-type: application/json' -d '{
  "transient": { "indices.recovery.max_bytes_per_sec": null,
                 "cluster.routing.allocation.node_concurrent_recoveries": null }
}'

THROTTLE ≠ NO: it means "yes, eventually, but not right now" — the cluster is pacing recovery. Confirm:

grep -n "THROTTLE\|class ThrottlingAllocationDecider\|node_concurrent_recoveries" \
  server/src/main/java/org/opensearch/cluster/routing/allocation/decider/ThrottlingAllocationDecider.java | head

Step 7 (optional) — Live add/remove with InternalTestCluster

For deterministic add/remove without restarting gradlew run:

// server/src/test/java/.../DC4ChurnIT.java
package org.opensearch.cluster.routing.allocation;

import org.opensearch.cluster.health.ClusterHealthStatus;
import org.opensearch.test.OpenSearchIntegTestCase;
import org.opensearch.test.OpenSearchIntegTestCase.ClusterScope;
import org.opensearch.test.OpenSearchIntegTestCase.Scope;

import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked;
import static org.hamcrest.Matchers.equalTo;

@ClusterScope(scope = Scope.TEST, numDataNodes = 3)
public class DC4ChurnIT extends OpenSearchIntegTestCase {

    public void testAddNodeRebalancesThenRemoveRecovers() throws Exception {
        assertAcked(prepareCreate("dc4")
            .setSettings(org.opensearch.common.settings.Settings.builder()
                .put("index.number_of_shards", 6)
                .put("index.number_of_replicas", 1)));
        ensureGreen("dc4");

        // ADD a node → rebalancing should move shards onto it.
        String added = internalCluster().startNode();
        ensureGreen("dc4");
        long onNew = client().admin().cluster().prepareState().get().getState()
            .getRoutingNodes().node(
                internalCluster().clusterService(added).localNode().getId()).size();
        logger.info("--> shards on the new node after rebalance: {}", onNew);
        assertTrue("new node should receive shards", onNew > 0);

        // REMOVE a node → replicas reallocate + recover; cluster returns to green.
        internalCluster().stopRandomDataNode();
        ensureGreen("dc4");   // recovers within the green timeout
        ClusterHealthStatus status = client().admin().cluster().prepareHealth("dc4")
            .get().getStatus();
        assertThat(status, equalTo(ClusterHealthStatus.GREEN));
    }
}

./gradlew :server:internalClusterTest \
  --tests "org.opensearch.cluster.routing.allocation.DC4ChurnIT" \
  -Dtests.logger.org.opensearch.cluster.routing.allocation=DEBUG

This asserts the two churn behaviors deterministically: the added node receives shards (rebalance), and after a removal the cluster recovers back to green.

Deliverables

• ~/opensearch-notes/dc4/notes.md with: initial balanced allocation; the rebalance onto an added node; the yellow→green recovery after a removal; the _cluster/allocation/explain output for two different stuck causes (same_shard and disk_threshold) each mapped to its decider class; and the THROTTLE-during-recovery observation.
• One _cluster/reroute invocation (retry or dry-run move) with its decision output.
• (Optional) A green DC4ChurnIT.

Troubleshooting

Expected Output

Indices allocate ~evenly with no primary/replica colocated; adding a node triggers visible RELOCATING shards and re-flattens the per-node count; removing a node goes yellow→green as replicas recover from primaries (throttled); a forced over-replication stays UNASSIGNED with _cluster/allocation/explain naming the same_shard decider, and a forced low disk watermark names disk_threshold; lowering recovery limits makes the explain show THROTTLE.

Stretch Goals

• Use index.routing.allocation.require._name to pin an index to one node, then explain a replica that can't be placed (the filter decider → FilterAllocationDecider).
• Turn on awareness (cluster.routing.allocation.awareness.attributes: zone) with faked node.attr.zone values and watch AwarenessAllocationDecider spread copies across zones.
• Set cluster.routing.allocation.enable: none, create an index, and confirm everything stays UNASSIGNED with the enable decider (EnableAllocationDecider); then flip it back to all and watch allocation proceed.
• Read RoutingNodes and trace how AllocationService.reroute mutates it during one pass, then publishes the new routing table as a cluster-state change (Lab DC3). Link shard-allocation.

Coding Exercises

You drove allocation by hand with curl and explain; now turn each behaviour into a test or a small variant. You already have the DC4ChurnIT skeleton (Step 7) — build on it. Locate every class with rg first (rg -l "class SameShardAllocationDecider" server/src/main); never trust a stale line number.

1. (warm-up) Assert the SameShardAllocationDecider invariant in an integ test. Extend DC4ChurnIT: after ensureGreen("dc4"), walk the RoutingNodes and assert that no node holds both the primary and a replica of the same shard. Read the routing model with rg -n "class RoutingNodes|node(\|assignedShards" server/src/main/java/org/opensearch/cluster/routing/RoutingNodes.java. Verify: ./gradlew :server:internalClusterTest --tests "...DC4ChurnIT".

2. (core) Assert rebalance onto an added node, quantitatively. Strengthen Step 7's assertTrue(onNew > 0) into a tighter bound: with 12 shards over 4 nodes the new node should end with at least 2 (ideally ~3). Read the weight knobs first (rg -n "INDEX_BALANCE_FACTOR|SHARD_BALANCE_FACTOR|THRESHOLD" server/src/main/java/org/opensearch/cluster/routing/allocation/allocator/BalancedShardsAllocator.java) and write the assertion so it documents what the balancer is optimising.

3. (core) Reproduce the same_shard NO as a unit test. Find the decider's tests (rg -l "class SameShardAllocationDeciderTests|class SameShardRoutingTests" server/src/test). Add a unit test that builds a RoutingAllocation with a shard copy already on a node and asserts decide(...) returns Decision.NO for that node — the in-code version of the explain output you captured in Step 4. This avoids a live cluster entirely.

4. (core) Allocation-under-churn integ test: removal recovers to green. Generalise Step 7 into a loop: add a node, ensureGreen, remove a random data node, ensureGreen, repeat 3 times, asserting green after each step and that unassignedShards() returns to 0. This is the churn-stability contract a balancer change must not break. Read stopRandomDataNode/startNode semantics in rg -n "stopRandomDataNode|startNode" test/framework/src/main/java/org/opensearch/test/InternalTestCluster.java.

5. (core) Assert THROTTLE ≠ NO for the throttling decider. Find ThrottlingAllocationDeciderTests (rg -l "ThrottlingAllocationDeciderTests"). Add a test that sets node_concurrent_recoveries: 1, puts one recovery in flight, and asserts a second waiting replica's decision is Decision.THROTTLE (not NO) — "yes, eventually, not now." This is the in-code form of the explain THROTTLE you saw in Step 6.

6. (advanced) An allocation-under-churn IT with a custom decider. Advanced challenge: read how deciders are registered (rg -n "AllocationDecider|class AllocationDeciders|createAllocationDeciders" server/src/main/java/org/opensearch/cluster/routing/allocation/), then write a tiny AllocationDecider that refuses to place more than K shards of one index on any node (an index-spread cap), register it in a test plugin's getAllocationDeciders, and write an OpenSearchIntegTestCase that creates an index, churns nodes, and asserts your cap is never violated across the run — and that _cluster/allocation/explain (drive it via the Java client) names your decider when it bites. The deliverable is one IT proving a custom placement constraint holds under add/remove churn, propagated through the same publish/commit you traced in Lab DC3.

Issues to Practice On

"Shards stuck unassigned" and "rebalancing won't stop" are perennial on opensearch-project/OpenSearch. Hunt them (labels move; confirm on the tracker):

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 "allocation OR unassigned OR rebalance OR recovery"
gh issue list --repo opensearch-project/OpenSearch --label "Storage" --state open
gh label list --repo opensearch-project/OpenSearch | grep -iE "alloc|cluster|storage|recovery|flaky"

Representative patterns. (1) "Shard stuck UNASSIGNED, explain shows decider X" — reproduce the exact decider with a small index + settings (as in Step 4), confirm the verdict in a ...DeciderTests unit test, fix the logic or the message, ship the test. (2) "Rebalancing thrashes / never settles" — reproduce with a churn IT, read the BalancedShardsAllocator weight function via rg, and either fix the weight/threshold or add a regression IT that asserts the cluster reaches a stable shard count. Arc: reproduce → locate via rg → fix → test → PR with CHANGELOG + DCO.

Planted-bug drill. In SameShardAllocationDecider (locate with rg -n "canForceAllocatePrimary|decide" server/src/main/java/org/opensearch/cluster/routing/allocation/decider/SameShardAllocationDecider.java), flip the colocation check so it returns Decision.YES where it should return NO. Run ./gradlew :server:test --tests "*SameShard*" and watch which test goes red — that test is the guard against primary+replica colocation (a durability hazard). Revert, then add the RoutingNodes invariant assertion from Exercise 1 so this regression is caught at the integ level too.

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 an API field or a class for each:

1. After initial allocation, which decider guarantees a shard's primary and replica are never on the same node? How would you confirm it from _cat/shards?
2. You added a node and saw RELOCATING shards. What is the balancer optimizing, and which class/setting controls the weight function?
3. A node left and the cluster went yellow then green — but a different failure made it red. Explain the difference in terms of primaries vs replicas.
4. You forced a stuck shard two different ways. For each, name the field in _cluster/allocation/explain that revealed the cause and the decider class it maps to.
5. What is the difference between a decider returning NO vs THROTTLE? Give a concrete example of each you produced in this lab.
6. When is _cluster/reroute?retry_failed=true the right fix, and which decider's counter does it reset?
7. Tie it back to consensus: every one of these reallocations changed the cluster state. Through which two-phase protocol did each new routing table reach the other nodes? (Link Lab DC3.)

When you can induce an unassigned shard, diagnose it down to the exact decider with the explain API, fix it with reroute, and explain how the new routing table propagates via consensus — you've completed Lab DC4 and the Distributed Consensus masterclass. Return to the intensive and answer its Validation questions, then continue to a sibling masterclass: query-engine or sharding-routing.