Pause indexing completely in serverless when throttling

server/src/main/java/org/elasticsearch/common/settings/ClusterSettings.java

@@ -526,6 +526,7 @@ public void apply(Settings value, Settings current, Settings previous) {
         IndexingMemoryController.MAX_INDEX_BUFFER_SIZE_SETTING,
         IndexingMemoryController.SHARD_INACTIVE_TIME_SETTING,
         IndexingMemoryController.SHARD_MEMORY_INTERVAL_TIME_SETTING,
+        IndexingMemoryController.PAUSE_INDEXING_ON_THROTTLE,
         ResourceWatcherService.ENABLED,
         ResourceWatcherService.RELOAD_INTERVAL_HIGH,
         ResourceWatcherService.RELOAD_INTERVAL_MEDIUM,

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

@@ -9,6 +9,7 @@
 
 package org.elasticsearch.index.engine;
 
+import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.apache.lucene.codecs.perfield.PerFieldKnnVectorsFormat;
 import org.apache.lucene.index.ByteVectorValues;
@@ -87,6 +88,7 @@
 import org.elasticsearch.index.store.Store;
 import org.elasticsearch.index.translog.Translog;
 import org.elasticsearch.index.translog.TranslogStats;
+import org.elasticsearch.indices.IndexingMemoryController;
 import org.elasticsearch.indices.recovery.RecoverySettings;
 import org.elasticsearch.search.suggest.completion.CompletionStats;
 import org.elasticsearch.threadpool.ThreadPool;
@@ -145,6 +147,7 @@ public abstract class Engine implements Closeable {
     protected final ReentrantLock failEngineLock = new ReentrantLock();
     protected final SetOnce<Exception> failedEngine = new SetOnce<>();
     protected final boolean enableRecoverySource;
+    protected final boolean pauseIndexingOnThrottle;
 
     private final AtomicBoolean isClosing = new AtomicBoolean();
     private final SubscribableListener<Void> drainOnCloseListener = new SubscribableListener<>();
@@ -176,6 +179,9 @@ protected Engine(EngineConfig engineConfig) {
         this.enableRecoverySource = RecoverySettings.INDICES_RECOVERY_SOURCE_ENABLED_SETTING.get(
             engineConfig.getIndexSettings().getSettings()
         );
+        this.pauseIndexingOnThrottle = IndexingMemoryController.PAUSE_INDEXING_ON_THROTTLE.get(
+            engineConfig.getIndexSettings().getSettings()
+        );
     }
 
     /**
@@ -444,14 +450,37 @@ public interface IndexCommitListener {
      * is enabled
      */
     protected static final class IndexThrottle {
+        private static final Logger logger = LogManager.getLogger(IndexThrottle.class);
         private final CounterMetric throttleTimeMillisMetric = new CounterMetric();
         private volatile long startOfThrottleNS;
         private static final ReleasableLock NOOP_LOCK = new ReleasableLock(new NoOpLock());
         private final ReleasableLock lockReference = new ReleasableLock(new ReentrantLock());
+        private final Lock pauseIndexingLock = new ReentrantLock();
+        private final Condition pauseCondition = pauseIndexingLock.newCondition();
+        private final ReleasableLock pauseLockReference = new ReleasableLock(pauseIndexingLock);
+        private volatile AtomicBoolean pauseIndexing = new AtomicBoolean();
         private volatile ReleasableLock lock = NOOP_LOCK;
 
         public Releasable acquireThrottle() {
-            return lock.acquire();
+            if (lock == pauseLockReference) {
+                pauseLockReference.acquire();
+                try {
+                    while (pauseIndexing.getAcquire()) {
+                        // System.out.println("Waiting on pause indexing lock");
+                        logger.trace("Waiting on pause indexing lock");
+                        pauseCondition.await();
+                    }
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                    throw new RuntimeException(e);
+                } finally {
+                    // System.out.println("Acquired pause indexing lock");
+                    logger.trace("Acquired pause indexing lock");
+                }
+                return pauseLockReference;
+            } else {
+                return lock.acquire();
+            }
         }
 
         /** Activate throttling, which switches the lock to be a real lock */
@@ -461,9 +490,31 @@ public void activate() {
             lock = lockReference;
         }
 
+        public void activatePause() {
+            assert lock == NOOP_LOCK : "throttling activated while already active";
+            startOfThrottleNS = System.nanoTime();
+            // System.out.println("Activate pause");
+            pauseIndexing.setRelease(true);
+            lock = pauseLockReference;
+        }
+
         /** 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) {
+                logger.trace("Deactivate index throttling pause");
+
+                // Signal the threads that are waiting on pauseCondition
+                pauseLockReference.acquire();
+                try {
+                    // System.out.println("Deactivate pause");
+                    pauseIndexing.setRelease(false);
+                    pauseCondition.signalAll();
+                } finally {
+                    pauseLockReference.close();
+                }
+            }
             lock = NOOP_LOCK;
 
             assert startOfThrottleNS > 0 : "Bad state of startOfThrottleNS";

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

@@ -171,7 +171,7 @@ 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();
     private final AtomicBoolean pendingTranslogRecovery = new AtomicBoolean(false);
@@ -2826,7 +2826,11 @@ public void activateThrottling() {
         int count = throttleRequestCount.incrementAndGet();
         assert count >= 1 : "invalid post-increment throttleRequestCount=" + count;
         if (count == 1) {
-            throttle.activate();
+            if (pauseIndexingOnThrottle) {
+                throttle.activatePause();
+            } else {
+                throttle.activate();
+            }
         }
     }
 

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

@@ -2733,6 +2733,9 @@ public IndexEventListener getIndexEventListener() {
         return indexEventListener;
     }
 
+    /** Activate throttling for this shard. If {@link IndexingMemoryController#PAUSE_INDEXING_ON_THROTTLE}
+     * setting is set to true, throttling will pause indexing completely. Otherwise, indexing will be throttled to one thread.
+     */
     public void activateThrottling() {
         try {
             getEngine().activateThrottling();

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

@@ -87,6 +87,17 @@ public class IndexingMemoryController implements IndexingOperationListener, Clos
         Property.NodeScope
     );
 
+    /* Currently, indexing is throttled due to memory pressure in stateful/stateless or disk pressure in stateless.
+     * This limits the number of indexing threads to 1 per shard. However, this might not be enough when the number of
+     * shards that need indexing is larger than the number of threads. So we might opt to pause indexing completely.
+     * The default value for this setting is false, but it will be set to true in stateless.
+     */
+    public static final Setting<Boolean> PAUSE_INDEXING_ON_THROTTLE = Setting.boolSetting(
+        "indices.pause.on.throttle",
+        false,
+        Property.NodeScope
+    );
+
     private final ThreadPool threadPool;
 
     private final Iterable<IndexShard> indexShards;
@@ -236,7 +247,8 @@ 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();
     }