Suspend Index throttling when relocating #128797

ankikuma · 2025-06-03T03:56:11Z

Addresses ES-11770.

If index throttling is enabled such that it pauses all indexing threads that try to index into a shard, this can starve other tasks such as relocation that try to acquire all indexing permits. This PR addresses this by suspending throttling to allow the indexing threads that are holding the permits to pass.

…exingForPermits Refresh

elasticsearchmachine · 2025-06-03T03:57:02Z

Pinging @elastic/es-distributed-indexing (Team:Distributed Indexing)

ankikuma · 2025-06-03T04:02:14Z

I noticed during testing that throttling gets disabled once a shard is moved. I guess this is because of the way the engine is created for the relocated shard. But I haven't had a chance to dig into the relocation code to verify that this is expected behaviour.

henningandersen

Thanks for working on this. Left a number of comments.

henningandersen · 2025-06-03T04:31:30Z

server/src/main/java/org/elasticsearch/index/shard/IndexShardOperationPermits.java

+        indexShard.suspendThrottling();
        waitUntilBlocked(ActionListener.assertOnce(onAcquired), timeout, timeUnit, executor);
+        // TODO: Does this do anything ? Looks like the relocated shard does not have throttling enabled
+        indexShard.resumeThrottling();


I think I would prefer to handle this outside this class, we can make a method in IndexShard that wraps blockOperations and does this, avoiding sending an object to this method and the effect on testing etc.

Also, notice that this is sort of incorrect as is in that we sometimes call this with the executor set to the generic thread pool. We should instead resume throttling when the listener is called, that will handle all cases.

henningandersen · 2025-06-03T04:35:34Z

server/src/main/java/org/elasticsearch/index/shard/IndexShard.java

@@ -2752,6 +2758,40 @@ public void deactivateThrottling() {
        }
    }

+    public boolean isIndexingPaused() {


This does not seem necessary to expose outside IndexShard?

I was calling it from RelocationIT. I can remove it. Just for my understanding, why is it risky to expose this ?

Removed in the latest upload

It exposes internal state from the engine. As such it is not "risky", but exposing more than necessary breaks encapsulation. In particular this one is only there for testing and can be fetched just as easily without this. The IndexShard interface is huge and I'd like to keep the surface it has down.

Thanks for explaining Henning

henningandersen · 2025-06-03T04:36:07Z

server/src/main/java/org/elasticsearch/index/shard/IndexShard.java

+        } else {
+            indexingPaused = engine.isIndexingPaused();
+        }
+        if (indexingPaused) {


I would prefer to just suspend throttling regardless of pausing or not.

Yes I agree that is safer because what if throttling activation sneaks in concurrently with the "acquire permits" operation and an indexing thread gets paused, then we might not be able to acquire permits.

henningandersen · 2025-06-03T04:36:57Z

server/src/main/java/org/elasticsearch/index/shard/IndexShard.java

+        return (indexingPaused);
+    }
+
+    public boolean suspendThrottling() {


We do not seem to need the return value. I think we should remove it.

Sure, removed.

henningandersen · 2025-06-03T09:26:02Z

server/src/internalClusterTest/java/org/elasticsearch/recovery/RelocationIT.java

+        logger.info("--> index more docs so we have something in the translog");
+        for (int i = 10; i < 20; i++) {
+            prepareIndex("test").setId(Integer.toString(i)).setSource("field", "value" + i).get();
+        }


I do not folllow why this is important to the test?

It is not. I wrote this test by modifying testRelocationWhileIndexingRandom() so it's just a carry over from there. Removed it.

henningandersen · 2025-06-03T09:33:01Z

server/src/internalClusterTest/java/org/elasticsearch/recovery/RelocationIT.java

+        assertHitCount(prepareSearch("test").setSize(0), 20);
+
+        logger.info("--> relocate the shard from node1 to node2");
+        ClusterRerouteUtils.reroute(client(), new MoveAllocationCommand("test", 0, node_1, node_2));


I prefer to set an allocation rule through index settings. Someting like index.routing.allocation.include._id = node_2.

I am not sure I follow this comment.

Or maybe I do. Like this ?
updateIndexSettings(Settings.builder().put("index.routing.allocation.include._id", node_2), "test");

Do you want me to change this everywhere in this file ?

I am not sure how to make that work. I tried this:
updateIndexSettings(Settings.builder().put("index.routing.allocation.include._id", nodes[toNode]), "test");
ensureGreen(ACCEPTABLE_RELOCATION_TIME, "test");

But it looks like this is not enough to ensure that the shard has moved to the target node.

You need to use ._name if you use node_2, like done here (though that one excludes, you can do that too - or use include, both should work).

henningandersen · 2025-06-03T09:36:26Z

server/src/internalClusterTest/java/org/elasticsearch/recovery/RelocationIT.java

+        assertThat(clusterHealthResponse.isTimedOut(), equalTo(false));
+
+        // Relocated shard is not throttled
+        assertThat(shard.isIndexingPaused(), equalTo(false));


This seems surprising, why is it not throttled?

I initially thought it might be because the node that we relocate the shard to does not have PAUSE_THROTTLING enabled. But that doesn't help either. So I am guessing it has to do with how we do the relocation, wouldn't we have to recreate the engine on the new node and it probably will not transfer throttling ?

But wait, this is the original source shard we are talking about, not the relocated target shard, so it should have throttling enabled after we resume throttling. I will need to look into this a bit more.

Oh I figured it out, it's because the engine is null for the source shard. I will just get rid of this check, I don't think it is useful.

henningandersen · 2025-06-03T09:37:22Z

server/src/internalClusterTest/java/org/elasticsearch/recovery/RelocationIT.java

+
+        // Relocated shard is not throttled
+        assertThat(shard.isIndexingPaused(), equalTo(false));
+        logger.info("--> verifying count after relocation ...");


I think we need to wait for future.actionGet() before we can assert the below?

Well I figured that if the relocation had gone ahead, the indexing request that future is responsible for must have completed already. But I can add it anyways.

Sort of. The problem is that we do not (safely) wait for the thread to be blocked on the lock. So in principle, the indexing request may not have blocked relocation at all and only be done on the target node. Hence it may not have completed yet.

henningandersen · 2025-06-03T09:38:20Z

server/src/internalClusterTest/java/org/elasticsearch/recovery/RelocationIT.java

        String[] nodes = new String[numberOfNodes];
        logger.info("--> starting [node1] ...");
-        nodes[0] = internalCluster().startNode();
+        nodes[0] = internalCluster().startNode(Settings.builder().put(IndexingMemoryController.PAUSE_INDEXING_ON_THROTTLE.getKey(), true));


Should we do this randomly? No need to not verify this for the regular throttling too.

Suggested change

nodes[0] = internalCluster().startNode(Settings.builder().put(IndexingMemoryController.PAUSE_INDEXING_ON_THROTTLE.getKey(), true));

nodes[0] = internalCluster().startNode(Settings.builder().put(IndexingMemoryController.PAUSE_INDEXING_ON_THROTTLE.getKey(), randomBoolean()));

Sure good idea. Took me a minute to parse "No need to not verify" 😄

Sorry for my double negated danglish 🙂

henningandersen · 2025-06-03T09:38:46Z

server/src/internalClusterTest/java/org/elasticsearch/recovery/RelocationIT.java

+                IndexShard shard = indicesService.indexServiceSafe(resolveIndex("test")).getShard(0);
+                shard.activateThrottling();
+                // Verify that indexing is paused for the throttled shard
+                assertBusy(() -> { assertThat(shard.isIndexingPaused(), equalTo(true)); });


This may not work with the random enable above, but I am also not sure it is important.

Yup I changed it to check for throttling which will cover both cases.

…exingForPermits Refresh

…exingForPermits Refresh branch

…exingForPermits Refresh

henningandersen

Left more comments.

henningandersen · 2025-06-05T06:01:03Z

server/src/internalClusterTest/java/org/elasticsearch/recovery/RelocationIT.java

+        IndexShard shard = indicesService.indexServiceSafe(resolveIndex("test")).getShard(0);
+        shard.activateThrottling();
+        // Verify that indexing is paused for the throttled shard
+        assertThat(shard.isIndexingPaused(), equalTo(true));


I am unsure what value this check brings? Could:

Suggested change

assertThat(shard.isIndexingPaused(), equalTo(true));

assertThat(shard.getEngine().isThrottled(), equalTo(true));

suffice?

henningandersen · 2025-06-05T06:02:33Z

server/src/internalClusterTest/java/org/elasticsearch/recovery/RelocationIT.java

+        logger.info("--> Try to index a doc while indexing is paused");
+        IndexRequestBuilder indexRequestBuilder = prepareIndex("test").setId(Integer.toString(20)).setSource("field", "value" + 20);
+        var future = indexRequestBuilder.execute();
+        expectThrows(ElasticsearchException.class, () -> future.actionGet(10, TimeUnit.SECONDS));


I think 10s is too long here, perhaps 500ms? Otherwise IIUC, this test always takes minimum 10s?

Yes it takes 10s right now. Will change it to 500 ms

henningandersen · 2025-06-05T06:04:30Z

server/src/internalClusterTest/java/org/elasticsearch/recovery/RelocationIT.java

+
+        // Relocated shard is not throttled
+        assertThat(shard.isIndexingPaused(), equalTo(false));
+        logger.info("--> verifying count after relocation ...");


Sort of. The problem is that we do not (safely) wait for the thread to be blocked on the lock. So in principle, the indexing request may not have blocked relocation at all and only be done on the target node. Hence it may not have completed yet.

henningandersen · 2025-06-05T06:23:06Z

server/src/internalClusterTest/java/org/elasticsearch/recovery/RelocationIT.java

+        assertHitCount(prepareSearch("test").setSize(0), 20);
+
+        logger.info("--> relocate the shard from node1 to node2");
+        ClusterRerouteUtils.reroute(client(), new MoveAllocationCommand("test", 0, node_1, node_2));


You need to use ._name if you use node_2, like done here (though that one excludes, you can do that too - or use include, both should work).

henningandersen · 2025-06-05T06:31:27Z

server/src/internalClusterTest/java/org/elasticsearch/recovery/RelocationIT.java

+        prepareCreate("test", indexSettings(1, 0)).get();
+
+        logger.info("--> index 10 docs");
+        for (int i = 0; i < 10; i++) {


Let us randomize the doc count to signal that 10 is not a special number:

Suggested change

for (int i = 0; i < 10; i++) {

int numDocs = between(1,10);

for (int i = 0; i < numDocs; i++) {

henningandersen · 2025-06-05T06:38:52Z

server/src/main/java/org/elasticsearch/index/engine/Engine.java

+     */
+    public abstract void resumeThrottling();
+
+    public abstract boolean isIndexingPaused();


I believe my comment above could remove the last usage of this - in that case we should remove the method too.

henningandersen · 2025-06-05T06:40:15Z

server/src/main/java/org/elasticsearch/index/shard/IndexShard.java

+        return (indexingPaused);
+    }
+
+    public void suspendThrottling() {


Suggested change

public void suspendThrottling() {

private void suspendThrottling() {

henningandersen · 2025-06-05T06:40:27Z

server/src/main/java/org/elasticsearch/index/shard/IndexShard.java

+        }
+    }
+
+    public void resumeThrottling() {


Suggested change

public void resumeThrottling() {

private void resumeThrottling() {

henningandersen · 2025-06-05T06:41:52Z

server/src/main/java/org/elasticsearch/index/shard/IndexShardOperationPermits.java

@@ -90,7 +90,7 @@ public void blockOperations(
        waitUntilBlocked(ActionListener.assertOnce(onAcquired), timeout, timeUnit, executor);
    }

-    private void waitUntilBlocked(ActionListener<Releasable> onAcquired, long timeout, TimeUnit timeUnit, Executor executor) {
+    protected void waitUntilBlocked(ActionListener<Releasable> onAcquired, long timeout, TimeUnit timeUnit, Executor executor) {


I'd like to keep these methods private.

henningandersen · 2025-06-05T06:48:04Z

server/src/main/java/org/elasticsearch/index/shard/IndexShard.java

+        indexShardOperationPermits.delayOperations();
+        // In case indexing is paused on the shard, suspend throttling so that any currently paused task can
+        // go ahead and release the indexing permit it holds.
+        suspendThrottling();
+        indexShardOperationPermits.waitUntilBlocked(ActionListener.assertOnce(onAcquired), timeout, timeUnit, executor);
+        // TODO: Does this do anything ? Looks like the relocated shard does not have throttling enabled
+        resumeThrottling();


I think we can just use blockOperations instead, rather than expose the underlying private methods.

Also, we should resume the throttle in the action listener instead.

Suggested change

indexShardOperationPermits.delayOperations();

// In case indexing is paused on the shard, suspend throttling so that any currently paused task can

// go ahead and release the indexing permit it holds.

suspendThrottling();

indexShardOperationPermits.waitUntilBlocked(ActionListener.assertOnce(onAcquired), timeout, timeUnit, executor);

// TODO: Does this do anything ? Looks like the relocated shard does not have throttling enabled

resumeThrottling();

// In case indexing is paused on the shard, suspend throttling so that any currently paused task can

// go ahead and release the indexing permit it holds.

suspendThrottling();

try {

indexShardOperationPermits.blockOperations(ActionListener.runAfter(onAcquired, this::resumeThrottling), timeout, timeUnit, executor);

} catch (IndexShardClosedException e) {

resumeThrottling();

throw e;

}

We can change the exception throw in delayOperations to use the listener instead in a follow-up (I worry it might have side-effects, so best to tackle separately). Happy to leave with the catch too though not too pretty.

Ah ok, you want to suspendThrottling() before we start delaying operations even. That should be fine too.

Yes, or rather I do not want to expose the difference from the permits class.

…exingForPermits Refresh branch

…uma/elasticsearch into 05192025/UnpauseIndexingForPermits pull

…exingForPermits refresh branch

…uma/elasticsearch into 05192025/UnpauseIndexingForPermits Refresh branch

…exingForPermits refresh branch

ankikuma added 9 commits May 23, 2025 13:59

pause indexing and race condition diags

dfe639f

commit

2601960

commit

ec91a19

refresh branch

3ddb78b

commit

f12949e

commit

90670f3

commit

45e3799

Merge remote-tracking branch 'upstream/main' into 05192025/UnpauseInd…

cd43ab3

…exingForPermits Refresh

commit

bf91cab

elasticsearchmachine added needs:triage Requires assignment of a team area label v9.1.0 labels Jun 3, 2025

ankikuma added :Distributed Indexing/Distributed A catch all label for anything in the Distributed Indexing Area. Please avoid if you can. and removed needs:triage Requires assignment of a team area label v9.1.0 labels Jun 3, 2025

elasticsearchmachine added the Team:Distributed Indexing Meta label for Distributed Indexing team label Jun 3, 2025

ankikuma added >non-issue and removed Team:Distributed Indexing Meta label for Distributed Indexing team labels Jun 3, 2025

elasticsearchmachine added the Team:Distributed Indexing Meta label for Distributed Indexing team label Jun 3, 2025

ankikuma added v9.1.0 and removed Team:Distributed Indexing Meta label for Distributed Indexing team labels Jun 3, 2025

elasticsearchmachine added the Team:Distributed Indexing Meta label for Distributed Indexing team label Jun 3, 2025

ankikuma requested a review from henningandersen June 3, 2025 04:03

henningandersen reviewed Jun 3, 2025

View reviewed changes

ankikuma added 5 commits June 3, 2025 20:14

address review comments

e642fea

Merge remote-tracking branch 'upstream/main' into 05192025/UnpauseInd…

a249357

…exingForPermits Refresh

Merge remote-tracking branch 'upstream/main' into 05192025/UnpauseInd…

a11d2fd

…exingForPermits Refresh branch

test failure

82a37f5

remove commented code

560a035

ankikuma and others added 5 commits June 4, 2025 09:21

Merge remote-tracking branch 'upstream/main' into 05192025/UnpauseInd…

1bb089e

…exingForPermits Refresh branch

minor changes

048f944

modified testRelocationWhileIndexingRandom

3e044bf

[CI] Auto commit changes from spotless

7abedf2

Merge remote-tracking branch 'upstream/main' into 05192025/UnpauseInd…

76f59b6

…exingForPermits Refresh branch

elasticsearchmachine added the serverless-linked Added by automation, don't add manually label Jun 4, 2025

ankikuma requested a review from tlrx June 4, 2025 20:29

ankikuma added 3 commits June 4, 2025 23:00

Merge remote-tracking branch 'upstream/main' into 05192025/UnpauseInd…

4afc8c5

…exingForPermits Refresh

update index settings

c5cce6e

test

8485a8e

henningandersen reviewed Jun 5, 2025

View reviewed changes

ankikuma and others added 13 commits June 5, 2025 10:51

Merge remote-tracking branch 'upstream/main' into 05192025/UnpauseInd…

6315189

…exingForPermits Refresh branch

address comments

5e81c31

test

661fa12

[CI] Auto commit changes from spotless

77058ad

test

1d872c1

Merge branch '05192025/UnpauseIndexingForPermits' of github.com:ankik…

129622a

…uma/elasticsearch into 05192025/UnpauseIndexingForPermits pull

old changes

610bc0f

pull changes

e50c200

Merge remote-tracking branch 'upstream/main' into 05192025/UnpauseInd…

343d9f2

…exingForPermits refresh branch

Merge branch '05192025/UnpauseIndexingForPermits' of github.com:ankik…

94f212a

…uma/elasticsearch into 05192025/UnpauseIndexingForPermits Refresh branch

fix test

6a2bf2b

[CI] Auto commit changes from spotless

ee22887

Merge remote-tracking branch 'upstream/main' into 05192025/UnpauseInd…

8e9d456

…exingForPermits refresh branch

	nodes[0] = internalCluster().startNode(Settings.builder().put(IndexingMemoryController.PAUSE_INDEXING_ON_THROTTLE.getKey(), true));
	nodes[0] = internalCluster().startNode(Settings.builder().put(IndexingMemoryController.PAUSE_INDEXING_ON_THROTTLE.getKey(), randomBoolean()));

	assertThat(shard.isIndexingPaused(), equalTo(true));
	assertThat(shard.getEngine().isThrottled(), equalTo(true));

	for (int i = 0; i < 10; i++) {
	int numDocs = between(1,10);
	for (int i = 0; i < numDocs; i++) {

	public void suspendThrottling() {
	private void suspendThrottling() {

	public void resumeThrottling() {
	private void resumeThrottling() {

Suspend Index throttling when relocating #128797

Are you sure you want to change the base?

Suspend Index throttling when relocating #128797

Conversation

ankikuma commented Jun 3, 2025

Uh oh!

elasticsearchmachine commented Jun 3, 2025

Uh oh!

ankikuma commented Jun 3, 2025

Uh oh!

henningandersen left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ankikuma Jun 3, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ankikuma Jun 3, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ankikuma Jun 3, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

ankikuma Jun 3, 2025 •

edited

Loading

ankikuma Jun 3, 2025 •

edited

Loading

ankikuma Jun 3, 2025 •

edited

Loading