commit es changes

Author: ankikuma

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

@@ -438,22 +438,33 @@ protected static final class IndexThrottle {
         private volatile long startOfThrottleNS;
         private static final ReleasableLock NOOP_LOCK = new ReleasableLock(new NoOpLock());
         private final ReleasableLock lockReference = new ReleasableLock(new ReentrantLock());
+        private final ReleasableLock pauseLockReference = new ReleasableLock(new PauseLock());
         private volatile ReleasableLock lock = NOOP_LOCK;
 
         public Releasable acquireThrottle() {
             return lock.acquire();
         }
 
         /** Activate throttling, which switches the lock to be a real lock */
-        public void activate() {
+        public void activate(AtomicBoolean pauseIndexing) {
             assert lock == NOOP_LOCK : "throttling activated while already active";
             startOfThrottleNS = System.nanoTime();
-            lock = lockReference;
+            if (pauseIndexing.getAcquire()) {
+                lock = pauseLockReference;
+            }
+            else {
+                lock = lockReference;
+            }
         }
 
         /** Deactivate throttling, which switches the lock to be an always-acquirable NoOpLock */
         public void deactivate() {
             assert lock != NOOP_LOCK : "throttling deactivated but not active";
+
+            if(lock == pauseLockReference)
+            {
+                lock.close();
+            }
             lock = NOOP_LOCK;
 
             assert startOfThrottleNS > 0 : "Bad state of startOfThrottleNS";
@@ -543,6 +554,67 @@ public Condition newCondition() {
         }
     }
 
+    /** A Lock implementation that forces the lock to wait on a condition */
+    protected static final class PauseLock implements Lock {
+
+        private final ReentrantLock internalLock = new ReentrantLock();
+        private final Condition condition = internalLock.newCondition();
+        private volatile boolean paused = true;
+
+        @Override
+        public void lock() {
+            internalLock.lock();
+            try {
+                while (paused) {
+                    condition.await();
+                }
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                throw new RuntimeException(e);
+            } finally {
+                internalLock.unlock();
+            }
+        }
+
+        @Override
+        public void lockInterruptibly() throws InterruptedException {
+            internalLock.lockInterruptibly();
+            try {
+                while (paused) {
+                    condition.await();
+                }
+            } finally {
+                internalLock.unlock();
+            }
+        }
+
+        @Override
+        public boolean tryLock() {
+            return false;
+        }
+
+        @Override
+        public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
+            return false;
+        }
+
+        @Override
+        public void unlock() {
+            internalLock.lock();
+            try {
+                paused = false;
+                condition.signalAll();
+            } finally {
+                internalLock.unlock();
+            }
+        }
+
+        @Override
+        public Condition newCondition() {
+            return condition;
+        }
+    }
+
     /**
      * Perform document index operation on the engine
      * @param index operation to perform
@@ -2173,7 +2245,7 @@ public interface Warmer {
      * Request that this engine throttle incoming indexing requests to one thread.
      * Must be matched by a later call to {@link #deactivateThrottling()}.
      */
-    public abstract void activateThrottling();
+    public abstract void activateThrottling(boolean pauseIndexing);
 
     /**
      * Reverses a previous {@link #activateThrottling} call.

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

@@ -171,9 +171,12 @@ public class InternalEngine extends Engine {
     private final CombinedDeletionPolicy combinedDeletionPolicy;
 
     // How many callers are currently requesting index throttling. Currently there are only two situations where we do this: when merges
-    // are falling behind and when writing indexing buffer to disk is too slow. When this is 0, there is no throttling, else we throttling
+    // are falling behind and when writing indexing buffer to disk is too slow. When this is 0, there is no throttling, else we throttle
     // incoming indexing ops to a single thread:
     private final AtomicInteger throttleRequestCount = new AtomicInteger();
+    // Should we throttle indexing ops by pausing them completely. Default value is false which will throttle
+    // incoming indexing ops to a single thread.
+    private final AtomicBoolean throttleShouldPauseIndexing = new AtomicBoolean(false);
     private final AtomicBoolean pendingTranslogRecovery = new AtomicBoolean(false);
     private final AtomicLong maxUnsafeAutoIdTimestamp = new AtomicLong(-1);
     private final AtomicLong maxSeenAutoIdTimestamp = new AtomicLong(-1);
@@ -2822,11 +2825,15 @@ public void accept(ElasticsearchDirectoryReader reader, ElasticsearchDirectoryRe
     }
 
     @Override
-    public void activateThrottling() {
+    public void activateThrottling(boolean pauseIndexing) {
         int count = throttleRequestCount.incrementAndGet();
+        if (pauseIndexing)
+        {
+            throttleShouldPauseIndexing.setRelease(true);
+        }
         assert count >= 1 : "invalid post-increment throttleRequestCount=" + count;
         if (count == 1) {
-            throttle.activate();
+            throttle.activate(throttleShouldPauseIndexing);
         }
     }
 
@@ -2920,7 +2927,7 @@ protected synchronized void enableIndexingThrottling(int numRunningMerges, int n
                 numQueuedMerges,
                 configuredMaxMergeCount
             );
-            InternalEngine.this.activateThrottling();
+            InternalEngine.this.activateThrottling(false);
         }
 
         @Override
@@ -2959,7 +2966,7 @@ public synchronized void beforeMerge(OnGoingMerge merge) {
             if (numMergesInFlight.incrementAndGet() > maxNumMerges) {
                 if (isThrottling.getAndSet(true) == false) {
                     logger.info("now throttling indexing: numMergesInFlight={}, maxNumMerges={}", numMergesInFlight, maxNumMerges);
-                    activateThrottling();
+                    activateThrottling(false);
                 }
             }
         }

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

@@ -502,7 +502,7 @@ public SafeCommitInfo getSafeCommitInfo() {
     }
 
     @Override
-    public void activateThrottling() {}
+    public void activateThrottling(boolean pauseIndexing) {}
 
     @Override
     public void deactivateThrottling() {}

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

@@ -2730,14 +2730,23 @@ public IndexEventListener getIndexEventListener() {
         return indexEventListener;
     }
 
-    public void activateThrottling() {
+    /** Asks this shard to throttle indexing to one thread. If pauseIndexing is set to true, throttling will
+     * pause indexing completely.
+     *
+     * @param pauseIndexing   pauses indexing completely when set to true, otherwise throttles to one thread.
+     */
+    public void activateThrottling(boolean pauseIndexing) {
         try {
-            getEngine().activateThrottling();
+            getEngine().activateThrottling(pauseIndexing);
         } catch (AlreadyClosedException ex) {
             // ignore
         }
     }
 
+    public void pauseIndexing() {
+
+    }
+
     public void deactivateThrottling() {
         try {
             getEngine().deactivateThrottling();

server/src/main/java/org/elasticsearch/indices/IndexingMemoryController.java

@@ -232,9 +232,10 @@ void forceCheck() {
         statusChecker.run();
     }
 
-    /** Asks this shard to throttle indexing to one thread */
+    /** Asks this shard to throttle indexing to one thread. If pauseIndexing is set to true, throttling will
+     * pause indexing completely. */
     protected void activateThrottling(IndexShard shard) {
-        shard.activateThrottling();
+        shard.activateThrottling(false);
     }
 
     /** Asks this shard to stop throttling indexing to one thread */

server/src/test/java/org/elasticsearch/index/engine/InternalEngineTests.java

@@ -7007,7 +7007,7 @@ public void testIndexThrottling() throws Exception {
                 throw new RuntimeException(ex);
             }
         }).when(indexWithoutThrottlingCheck).startTime();
-        engine.activateThrottling();
+        engine.activateThrottling(false);
         engine.index(indexWithThrottlingCheck);
         engine.deactivateThrottling();
         engine.index(indexWithoutThrottlingCheck);