commit

Author: ankikuma

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

@@ -43,6 +43,8 @@
 import org.elasticsearch.index.shard.IndexShard;
 import org.elasticsearch.index.shard.IndexShardState;
 import org.elasticsearch.index.shard.ShardId;
+import org.elasticsearch.indices.IndexingMemoryController;
+import org.elasticsearch.indices.IndicesService;
 import org.elasticsearch.indices.recovery.PeerRecoveryTargetService;
 import org.elasticsearch.indices.recovery.RecoveryFileChunkRequest;
 import org.elasticsearch.plugins.Plugin;
@@ -74,6 +76,7 @@
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.LockSupport;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -117,6 +120,64 @@ public Settings indexSettings() {
             .build();
     }
 
+    public void testSimpleRelocationIndexingPaused() {
+        logger.info("--> starting [node1] ...");
+        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();
+
+        logger.info("--> index 10 docs");
+        for (int i = 0; i < 10; 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("--> 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();
+        }
+
+        logger.info("--> verifying count");
+        indicesAdmin().prepareRefresh().get();
+        assertHitCount(prepareSearch("test").setSize(0), 20L);
+
+        logger.info("--> start another node");
+        final String node_2 = internalCluster().startNode();
+        ClusterHealthResponse clusterHealthResponse = clusterAdmin().prepareHealth(TEST_REQUEST_TIMEOUT)
+            .setWaitForEvents(Priority.LANGUID)
+            .setWaitForNodes("2")
+            .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));
+
+        clusterHealthResponse = clusterAdmin().prepareHealth(TEST_REQUEST_TIMEOUT)
+            .setWaitForEvents(Priority.LANGUID)
+            .setWaitForNoRelocatingShards(true)
+            .setTimeout(ACCEPTABLE_RELOCATION_TIME)
+            .get();
+        //assertThat(clusterHealthResponse.isTimedOut(), equalTo(false));
+        assertThat(clusterHealthResponse.isTimedOut(), equalTo(true));
+
+        logger.info("--> verifying count again...");
+        indicesAdmin().prepareRefresh().get();
+        assertHitCount(prepareSearch("test").setSize(0), 20);
+    }
+
     public void testSimpleRelocationNoIndexing() {
         logger.info("--> starting [node1] ...");
         final String node_1 = internalCluster().startNode();

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

@@ -495,6 +495,7 @@ public void activate() {
             startOfThrottleNS = System.nanoTime();
             if (pauseWhenThrottled) {
                 lock = pauseLockReference;
+                System.out.println("Activated index throttling pause");
                 logger.trace("Activated index throttling pause");
             } else {
                 lock = lockReference;

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

@@ -90,11 +90,13 @@ public void blockOperations(
         delayOperations();
         // If indexing is paused on the shard, suspend it so that any currently paused task can
         // go ahead and release the indexing permit it holds.
-        boolean throttlingPaused = indexShard.suspendThrottling();
+        //boolean throttlingPaused = indexShard.suspendThrottling();
         waitUntilBlocked(ActionListener.assertOnce(onAcquired), timeout, timeUnit, executor);
+        /*
         if (throttlingPaused) {
             indexShard.resumeThrottling();
         }
+         */
     }
 
     private void waitUntilBlocked(ActionListener<Releasable> onAcquired, long timeout, TimeUnit timeUnit, Executor executor) {