commit

Author: ankikuma

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

@@ -461,7 +461,7 @@ protected static final class IndexThrottle {
         private final Lock pauseIndexingLock = new ReentrantLock();
         private final Condition pauseCondition = pauseIndexingLock.newCondition();
         private final ReleasableLock pauseLockReference = new ReleasableLock(pauseIndexingLock);
-        private volatile AtomicBoolean pauseThrottling = new AtomicBoolean();
+        private volatile AtomicBoolean suspendThrottling = new AtomicBoolean();
         private final boolean pauseWhenThrottled;
         private volatile ReleasableLock lock = NOOP_LOCK;
 
@@ -472,10 +472,10 @@ public IndexThrottle(boolean pause) {
         public Releasable acquireThrottle() {
             var lockCopy = this.lock;
             if (lockCopy == pauseLockReference) {
-                //try (pauseLockReference.acquire();)
-                try  (var ignored = pauseLockReference.acquire()) {
+                // try (pauseLockReference.acquire();)
+                try (var ignored = pauseLockReference.acquire()) {
                     // If throttling is activated and not temporarily paused
-                    while ((lock == pauseLockReference) && (pauseThrottling.getAcquire() == false)) {
+                    while ((lock == pauseLockReference) && (suspendThrottling.getAcquire() == false)) {
                         logger.trace("Waiting on pause indexing lock");
                         pauseCondition.await();
                     }
@@ -541,7 +541,7 @@ long getThrottleTimeInMillis() {
         public void suspendThrottle() {
             if (pauseWhenThrottled) {
                 try (Releasable releasableLock = pauseLockReference.acquire()) {
-                    pauseThrottling.setRelease(true);
+                    suspendThrottling.setRelease(true);
                     pauseCondition.signalAll();
                 }
             }
@@ -551,7 +551,7 @@ public void suspendThrottle() {
         public void resumeThrottle() {
             if (pauseWhenThrottled) {
                 try (Releasable releasableLock = pauseLockReference.acquire()) {
-                    pauseThrottling.setRelease(false);
+                    suspendThrottling.setRelease(false);
                     pauseCondition.signalAll();
                 }
             }
@@ -2345,12 +2345,12 @@ public interface Warmer {
      * another task trying to get indexing permits might want to pause throttling
      * by letting one thread pass at a time so that it does not get starved.
      */
-    public abstract void pauseThrottling();
+    public abstract void suspendThrottling();
 
     /**
-     * Reverses a previous {@link #pauseThrottling} call.
+     * Reverses a previous {@link #resumeThrottling} call.
      */
-    public abstract void unPauseThrottling();
+    public abstract void resumeThrottling();
 
     public abstract boolean isIndexingPaused();
 

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

@@ -2823,32 +2823,36 @@ public void accept(ElasticsearchDirectoryReader reader, ElasticsearchDirectoryRe
 
     @Override
     public void activateThrottling() {
-        int count = throttleRequestCount.incrementAndGet();
-        assert count >= 1 : "invalid post-increment throttleRequestCount=" + count;
-        if (count == 1) {
-            throttle.activate();
-            logger.info("Activated throttling");
+        synchronized (throttleRequestCount) {
+            int count = throttleRequestCount.incrementAndGet();
+            assert count >= 1 : "invalid post-increment throttleRequestCount=" + count;
+            if (count == 1) {
+                throttle.activate();
+                logger.info("Activated throttling");
+            }
         }
     }
 
     @Override
     public void deactivateThrottling() {
-        int count = throttleRequestCount.decrementAndGet();
-        assert count >= 0 : "invalid post-decrement throttleRequestCount=" + count;
-        if (count == 0) {
-            throttle.deactivate();
-            logger.info("Deactivated throttling");
+        synchronized (throttleRequestCount) {
+            int count = throttleRequestCount.decrementAndGet();
+            assert count >= 0 : "invalid post-decrement throttleRequestCount=" + count;
+            if (count == 0) {
+                throttle.deactivate();
+                logger.info("Deactivated throttling");
+            }
         }
     }
 
     @Override
-    public void pauseThrottling() {
-        throttle.pauseThrottle();
+    public void suspendThrottling() {
+        throttle.suspendThrottle();
     }
 
     @Override
-    public void unPauseThrottling() {
-        throttle.unpauseThrottle();
+    public void resumeThrottling() {
+        throttle.resumeThrottle();
     }
 
     @Override

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

@@ -508,10 +508,10 @@ public void activateThrottling() {}
     public void deactivateThrottling() {}
 
     @Override
-    public void pauseThrottling() {}
+    public void suspendThrottling() {}
 
     @Override
-    public void unPauseThrottling() {}
+    public void resumeThrottling() {}
 
     @Override
     public boolean isIndexingPaused() {

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

@@ -2758,7 +2758,7 @@ public void deactivateThrottling() {
         }
     }
 
-    public boolean pauseThrottling() {
+    public boolean suspendThrottling() {
         Engine engine = getEngineOrNull();
         final boolean indexingPaused;
         if (engine == null) {
@@ -2767,15 +2767,15 @@ public boolean pauseThrottling() {
             indexingPaused = engine.isIndexingPaused();
         }
         if (indexingPaused) {
-            engine.pauseThrottling();
+            engine.suspendThrottling();
             return (true);
         }
         return (false);
     }
 
-    public void unpauseThrottling() {
+    public void resumeThrottling() {
         try {
-            getEngine().unPauseThrottling();
+            getEngine().resumeThrottling();
         } catch (AlreadyClosedException ex) {
             // ignore
         }

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

@@ -88,12 +88,12 @@ public void blockOperations(
         @Nullable IndexShard indexShard
     ) {
         delayOperations();
-        // If indexing is paused on the shard, unpause it so that any currently waiting task can
+        // 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.pauseThrottling();
+        boolean throttlingPaused = indexShard.suspendThrottling();
         waitUntilBlocked(ActionListener.assertOnce(onAcquired), timeout, timeUnit, executor);
         if (throttlingPaused) {
-            indexShard.unpauseThrottling();
+            indexShard.resumeThrottling();
         }
     }