commit

Author: ankikuma

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

@@ -120,11 +120,9 @@ public Settings indexSettings() {
             .build();
     }
 
-    public void testSimpleRelocationIndexingPaused() {
+    public void testSimpleRelocationNoIndexing() {
         logger.info("--> starting [node1] ...");
-        final String node_1 = internalCluster().startNode(
-            Settings.builder()
-                .put(IndexingMemoryController.PAUSE_INDEXING_ON_THROTTLE.getKey(), true));
+        final String node_1 = internalCluster().startNode();
 
         logger.info("--> creating test index ...");
         prepareCreate("test", indexSettings(1, 0)).get();
@@ -152,16 +150,6 @@ public void testSimpleRelocationIndexingPaused() {
             .get();
         assertThat(clusterHealthResponse.isTimedOut(), equalTo(false));
 
-        IndicesService indicesService = internalCluster().getInstance(IndicesService.class, node_1);
-        IndexService indexService = indicesService.indexService(resolveIndex("test"));
-        IndexShard shard = indexService.getShard(0);
-        shard.activateThrottling();
-        LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(100));
-        logger.info("--> index 1 more doc");
-        IndexRequestBuilder indexRequestBuilder = prepareIndex("test").setId(Integer.toString(20)).setSource("field", "value" + 20);
-        var future = indexRequestBuilder.execute();
-        //future.actionGet();
-
         logger.info("--> relocate the shard from node1 to node2");
         ClusterRerouteUtils.reroute(client(), new MoveAllocationCommand("test", 0, node_1, node_2));
 
@@ -170,17 +158,23 @@ public void testSimpleRelocationIndexingPaused() {
             .setWaitForNoRelocatingShards(true)
             .setTimeout(ACCEPTABLE_RELOCATION_TIME)
             .get();
-        //assertThat(clusterHealthResponse.isTimedOut(), equalTo(false));
-        assertThat(clusterHealthResponse.isTimedOut(), equalTo(true));
+        assertThat(clusterHealthResponse.isTimedOut(), equalTo(false));
 
         logger.info("--> verifying count again...");
         indicesAdmin().prepareRefresh().get();
         assertHitCount(prepareSearch("test").setSize(0), 20);
     }
 
-    public void testSimpleRelocationNoIndexing() {
+    // This tests that relocation can successfully suspend index throttling to grab
+    // indexing permits required for relocation to succeed.
+    public void testSimpleRelocationWithIndexingPaused() throws Exception {
         logger.info("--> starting [node1] ...");
-        final String node_1 = internalCluster().startNode();
+        // Start node with PAUSE_INDEXING_ON_THROTTLE setting set to true. This means that if we activate
+        // index throttling for a shard on this node, it will pause indexing for that shard until throttling
+        // is deactivated.
+        final String node_1 = internalCluster().startNode(
+            Settings.builder()
+                .put(IndexingMemoryController.PAUSE_INDEXING_ON_THROTTLE.getKey(), true));
 
         logger.info("--> creating test index ...");
         prepareCreate("test", indexSettings(1, 0)).get();
@@ -208,19 +202,38 @@ public void testSimpleRelocationNoIndexing() {
             .get();
         assertThat(clusterHealthResponse.isTimedOut(), equalTo(false));
 
+        // Activate index throttling on "test" index primary shard
+        IndicesService indicesService = internalCluster().getInstance(IndicesService.class, node_1);
+        IndexService indexService = indicesService.indexService(resolveIndex("test"));
+        IndexShard shard = indexService.getShard(0);
+        shard.activateThrottling();
+        // Verify that indexing is paused for the throttled shard
+        assertBusy(() -> { assertThat(shard.isIndexingPaused(), equalTo(true)); });
+        // Try to index a document into the "test" index which is currently throttled
+        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();
+        // Verify that the new document has not been indexed indicating that the indexing thread is paused.
+        logger.info("--> verifying count is unchanged...");
+        indicesAdmin().prepareRefresh().get();
+        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));
 
+        // Relocation will suspend throttling for the paused shard, allow the indexing thread to proceed, thereby releasing
+        // the indexing permit it holds, in turn allowing relocation to acquire the permits and proceed.
         clusterHealthResponse = clusterAdmin().prepareHealth(TEST_REQUEST_TIMEOUT)
             .setWaitForEvents(Priority.LANGUID)
             .setWaitForNoRelocatingShards(true)
             .setTimeout(ACCEPTABLE_RELOCATION_TIME)
             .get();
         assertThat(clusterHealthResponse.isTimedOut(), equalTo(false));
 
-        logger.info("--> verifying count again...");
+        logger.info("--> verifying count after relocation ...");
         indicesAdmin().prepareRefresh().get();
-        assertHitCount(prepareSearch("test").setSize(0), 20);
+        assertHitCount(prepareSearch("test").setSize(0), 21);
+        logger.info("--> Test finished ...");
     }
 
     public void testRelocationWhileIndexingRandom() throws Exception {

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

@@ -461,7 +461,10 @@ protected static final class IndexThrottle {
         private final Condition pauseCondition = pauseIndexingLock.newCondition();
         private final ReleasableLock pauseLockReference = new ReleasableLock(pauseIndexingLock);
         private volatile AtomicBoolean suspendThrottling = new AtomicBoolean();
-        private final boolean pauseWhenThrottled; // Should throttling pause indexing ?
+
+        // Should throttling pause indexing ? This is decided by the
+        // IndexingMemoryController#PAUSE_INDEXING_ON_THROTTLE setting for this node.
+        private final boolean pauseWhenThrottled;
         private volatile ReleasableLock lock = NOOP_LOCK;
 
         public IndexThrottle(boolean pause) {
@@ -547,7 +550,7 @@ boolean isIndexingPaused() {
         /** Suspend throttling to allow another task such as relocation to acquire all indexing permits */
         public void suspendThrottle() {
             if (pauseWhenThrottled) {
-                try (Releasable releasableLock = pauseLockReference.acquire()) {
+                try (Releasable ignored = pauseLockReference.acquire()) {
                     suspendThrottling.setRelease(true);
                     pauseCondition.signalAll();
                 }
@@ -557,7 +560,7 @@ public void suspendThrottle() {
         /** Reverse what was done in {@link #suspendThrottle()} */
         public void resumeThrottle() {
             if (pauseWhenThrottled) {
-                try (Releasable releasableLock = pauseLockReference.acquire()) {
+                try (Releasable ignored = pauseLockReference.acquire()) {
                     suspendThrottling.setRelease(false);
                     pauseCondition.signalAll();
                 }

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

@@ -2758,6 +2758,16 @@ public void deactivateThrottling() {
         }
     }
 
+    public boolean isIndexingPaused() {
+        Engine engine = getEngineOrNull();
+        final boolean indexingPaused;
+        if (engine == null) {
+            indexingPaused = false;
+        } else {
+            indexingPaused = engine.isIndexingPaused();
+        }
+        return (indexingPaused);
+    }
     public boolean suspendThrottling() {
         Engine engine = getEngineOrNull();
         final boolean indexingPaused;

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

@@ -88,15 +88,11 @@ public void blockOperations(
         @Nullable IndexShard indexShard
     ) {
         delayOperations();
-        // If indexing is paused on the shard, suspend it so that any currently paused task can
+        // 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.
-        //boolean throttlingPaused = indexShard.suspendThrottling();
+        indexShard.suspendThrottling();
         waitUntilBlocked(ActionListener.assertOnce(onAcquired), timeout, timeUnit, executor);
-        /*
-        if (throttlingPaused) {
-            indexShard.resumeThrottling();
-        }
-         */
+        indexShard.resumeThrottling();
     }
 
     private void waitUntilBlocked(ActionListener<Releasable> onAcquired, long timeout, TimeUnit timeUnit, Executor executor) {