address comments

Author: ankikuma

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

@@ -38,6 +38,7 @@
 import org.elasticsearch.env.NodeEnvironment;
 import org.elasticsearch.index.IndexService;
 import org.elasticsearch.index.IndexSettings;
+import org.elasticsearch.index.engine.Engine;
 import org.elasticsearch.index.seqno.ReplicationTracker;
 import org.elasticsearch.index.seqno.RetentionLease;
 import org.elasticsearch.index.shard.IndexEventListener;
@@ -179,16 +180,17 @@ public void testSimpleRelocationWithIndexingPaused() throws Exception {
         logger.info("--> creating test index ...");
         prepareCreate("test", indexSettings(1, 0)).get();
 
-        logger.info("--> index 10 docs");
-        for (int i = 0; i < 10; i++) {
+        logger.info("--> index docs");
+        int numDocs = between(1,10);
+        for (int i = 0; i < numDocs; i++) {
             prepareIndex("test").setId(Integer.toString(i)).setSource("field", "value" + i).get();
         }
         logger.info("--> flush so we have an actual index");
         indicesAdmin().prepareFlush().get();
 
         logger.info("--> verifying count");
         indicesAdmin().prepareRefresh().get();
-        assertHitCount(prepareSearch("test").setSize(0), 10L);
+        assertHitCount(prepareSearch("test").setSize(0), numDocs);
 
         logger.info("--> start another node");
         final String node_2 = internalCluster().startNode();
@@ -203,20 +205,21 @@ public void testSimpleRelocationWithIndexingPaused() throws Exception {
         IndexShard shard = indicesService.indexServiceSafe(resolveIndex("test")).getShard(0);
         shard.activateThrottling();
         // Verify that indexing is paused for the throttled shard
-        assertThat(shard.isIndexingPaused(), equalTo(true));
+        Engine engine = shard.getEngineOrNull();
+        assertThat(engine != null && engine.isThrottled(), 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();
-        expectThrows(ElasticsearchException.class, () -> future.actionGet(10, TimeUnit.SECONDS));
+        expectThrows(ElasticsearchException.class, () -> future.actionGet(500, TimeUnit.MILLISECONDS));
         // 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), 10);
+        assertHitCount(prepareSearch("test").setSize(0), numDocs);
 
         logger.info("--> relocate the shard from node1 to node2");
-        ClusterRerouteUtils.reroute(client(), new MoveAllocationCommand("test", 0, node_1, node_2));
-        // updateIndexSettings(Settings.builder().put("index.routing.allocation.include._id", node_2), "test");
+        updateIndexSettings(Settings.builder().put("index.routing.allocation.include._name", node_2), "test");
+        ensureGreen(ACCEPTABLE_RELOCATION_TIME, "test");
 
         // 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.
@@ -230,7 +233,7 @@ public void testSimpleRelocationWithIndexingPaused() throws Exception {
         logger.info("--> verifying count after relocation ...");
         future.actionGet();
         indicesAdmin().prepareRefresh().get();
-        assertHitCount(prepareSearch("test").setSize(0), 11);
+        assertHitCount(prepareSearch("test").setSize(0), numDocs + 1);
     }
 
     public void testRelocationWhileIndexingRandom() throws Exception {
@@ -283,7 +286,8 @@ public void testRelocationWhileIndexingRandom() throws Exception {
                 // Activate index throttling on "test" index primary shard
                 shard.activateThrottling();
                 // Verify that indexing is throttled for this shard
-                assertThat(shard.getEngineOrNull().isThrottled(), equalTo(true));
+                Engine engine = shard.getEngineOrNull();
+                assertThat(engine != null && engine.isThrottled(), equalTo(true));
             }
             logger.info("--> starting relocations...");
             int nodeShiftBased = numberOfReplicas; // if we have replicas shift those
@@ -296,10 +300,9 @@ public void testRelocationWhileIndexingRandom() throws Exception {
                 logger.debug("--> Allow indexer to index [{}] documents", numDocs);
                 indexer.continueIndexing(numDocs);
                 logger.info("--> START relocate the shard from {} to {}", nodes[fromNode], nodes[toNode]);
-                ClusterRerouteUtils.reroute(client(), new MoveAllocationCommand("test", 0, nodes[fromNode], nodes[toNode]));
 
-                // updateIndexSettings(Settings.builder().put("index.routing.allocation.include._id", nodes[toNode]), "test");
-                // ensureGreen(ACCEPTABLE_RELOCATION_TIME, "test");
+                updateIndexSettings(Settings.builder().put("index.routing.allocation.include._name", nodes[toNode]), "test");
+                ensureGreen(ACCEPTABLE_RELOCATION_TIME, "test");
 
                 if (rarely()) {
                     logger.debug("--> flushing");
@@ -321,7 +324,8 @@ public void testRelocationWhileIndexingRandom() throws Exception {
                     shard = indicesService.indexServiceSafe(resolveIndex("test")).getShard(0);
                     shard.activateThrottling();
                     // Verify that indexing is throttled for this shard
-                    assertThat(shard.getEngineOrNull().isThrottled(), equalTo(true));
+                    Engine engine = shard.getEngineOrNull();
+                    assertThat(engine != null && engine.isThrottled(), equalTo(true));
                 }
             }
             logger.info("--> done relocations");

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

@@ -2276,8 +2276,6 @@ public interface Warmer {
      */
     public abstract void resumeThrottling();
 
-    public abstract boolean isIndexingPaused();
-
     /**
      * This method replays translog to restore the Lucene index which might be reverted previously.
      * This ensures that all acknowledged writes are restored correctly when this engine is promoted.

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

@@ -2867,11 +2867,6 @@ public boolean isThrottled() {
         return throttle.isThrottled();
     }
 
-    @Override
-    public boolean isIndexingPaused() {
-        return throttle.isIndexingPaused();
-    }
-
     boolean throttleLockIsHeldByCurrentThread() {  // to be used in assertions and tests only
         return throttle.throttleLockIsHeldByCurrentThread();
     }

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

@@ -513,11 +513,6 @@ public void suspendThrottling() {}
     @Override
     public void resumeThrottling() {}
 
-    @Override
-    public boolean isIndexingPaused() {
-        return (false);
-    }
-
     @Override
     public void trimUnreferencedTranslogFiles() {}
 

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

@@ -2757,26 +2757,15 @@ public void deactivateThrottling() {
         }
     }
 
-    public boolean isIndexingPaused() {
-        Engine engine = getEngineOrNull();
-        final boolean indexingPaused;
-        if (engine == null) {
-            indexingPaused = false;
-        } else {
-            indexingPaused = engine.isIndexingPaused();
-        }
-        return (indexingPaused);
-    }
-
-    public void suspendThrottling() {
+    private void suspendThrottling() {
         try {
             getEngine().suspendThrottling();
         } catch (AlreadyClosedException ex) {
             // ignore
         }
     }
 
-    public void resumeThrottling() {
+    private void resumeThrottling() {
         try {
             getEngine().resumeThrottling();
         } catch (AlreadyClosedException ex) {
@@ -3850,13 +3839,16 @@ public void blockOperations(
         final TimeUnit timeUnit,
         final Executor executor
     ) {
-        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();
+        try {
+            indexShardOperationPermits.blockOperations(ActionListener.runAfter(onAcquired, this::resumeThrottling),
+                timeout, timeUnit, executor);
+        } catch (IndexShardClosedException e) {
+            resumeThrottling();
+            throw e;
+        }
     }
 
     private void asyncBlockOperations(ActionListener<Releasable> onPermitAcquired, long timeout, TimeUnit timeUnit) {

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

@@ -90,7 +90,7 @@ public void blockOperations(
         waitUntilBlocked(ActionListener.assertOnce(onAcquired), timeout, timeUnit, executor);
     }
 
-    protected void waitUntilBlocked(ActionListener<Releasable> onAcquired, long timeout, TimeUnit timeUnit, Executor executor) {
+    private void waitUntilBlocked(ActionListener<Releasable> onAcquired, long timeout, TimeUnit timeUnit, Executor executor) {
         executor.execute(new AbstractRunnable() {
 
             final Releasable released = Releasables.releaseOnce(() -> releaseDelayedOperations());
@@ -126,7 +126,7 @@ protected void doRun() {
         });
     }
 
-    protected void delayOperations() {
+    private void delayOperations() {
         if (closed) {
             throw new IndexShardClosedException(shardId);
         }