Threadpool merge executor is aware of available disk space

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

@@ -88,6 +88,7 @@
 import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.IndexingPressure;
 import org.elasticsearch.index.MergePolicyConfig;
+import org.elasticsearch.index.engine.ThreadPoolMergeExecutorService;
 import org.elasticsearch.index.engine.ThreadPoolMergeScheduler;
 import org.elasticsearch.index.shard.IndexingStatsSettings;
 import org.elasticsearch.indices.IndexingMemoryController;
@@ -628,6 +629,9 @@ public void apply(Settings value, Settings current, Settings previous) {
         MergePolicyConfig.DEFAULT_MAX_MERGED_SEGMENT_SETTING,
         MergePolicyConfig.DEFAULT_MAX_TIME_BASED_MERGED_SEGMENT_SETTING,
         ThreadPoolMergeScheduler.USE_THREAD_POOL_MERGE_SCHEDULER_SETTING,
+        ThreadPoolMergeExecutorService.INDICES_MERGE_DISK_HIGH_WATERMARK_SETTING,
+        ThreadPoolMergeExecutorService.INDICES_MERGE_DISK_HIGH_MAX_HEADROOM_SETTING,
+        ThreadPoolMergeExecutorService.INDICES_MERGE_DISK_CHECK_INTERVAL_SETTING,
         TransportService.ENABLE_STACK_OVERFLOW_AVOIDANCE,
         DataStreamGlobalRetentionSettings.DATA_STREAMS_DEFAULT_RETENTION_SETTING,
         DataStreamGlobalRetentionSettings.DATA_STREAMS_MAX_RETENTION_SETTING,

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

@@ -462,6 +462,10 @@ long estimatedMergeSize() {
             return onGoingMerge.getMerge().getStoreMergeInfo().estimatedMergeBytes();
         }
 
+        long estimatedRemainingMergeSize() {
+            return Math.max(0L, estimatedMergeSize() - rateLimiter.getTotalBytesWritten());
+        }
+
         public long getMergeMemoryEstimateBytes() {
             return mergeMemoryEstimateBytes;
         }

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

@@ -293,12 +293,13 @@ protected void doStart() {
     IndicesService(IndicesServiceBuilder builder) {
         this.settings = builder.settings;
         this.threadPool = builder.threadPool;
+        this.pluginsService = builder.pluginsService;
+        this.nodeEnv = builder.nodeEnv;
         this.threadPoolMergeExecutorService = ThreadPoolMergeExecutorService.maybeCreateThreadPoolMergeExecutorService(
             threadPool,
-            settings
+            settings,
+            nodeEnv
         );
-        this.pluginsService = builder.pluginsService;
-        this.nodeEnv = builder.nodeEnv;
         this.parserConfig = XContentParserConfiguration.EMPTY.withDeprecationHandler(LoggingDeprecationHandler.INSTANCE)
             .withRegistry(builder.xContentRegistry);
         this.valuesSourceRegistry = builder.valuesSourceRegistry;
@@ -366,7 +367,8 @@ public void onRemoval(ShardId shardId, String fieldName, boolean wasEvicted, lon
                     indicesFieldDataCache,
                     cacheCleaner,
                     indicesRequestCache,
-                    indicesQueryCache
+                    indicesQueryCache,
+                    threadPoolMergeExecutorService
                 );
             } catch (IOException e) {
                 throw new UncheckedIOException(e);

server/src/test/java/org/elasticsearch/index/IndexModuleTests.java

@@ -194,13 +194,17 @@ public void setUp() throws Exception {
             emptyMap()
         );
         threadPool = new TestThreadPool("test");
-        threadPoolMergeExecutorService = ThreadPoolMergeExecutorService.maybeCreateThreadPoolMergeExecutorService(threadPool, settings);
         circuitBreakerService = new NoneCircuitBreakerService();
         PageCacheRecycler pageCacheRecycler = new PageCacheRecycler(settings);
         bigArrays = new BigArrays(pageCacheRecycler, circuitBreakerService, CircuitBreaker.REQUEST);
         scriptService = new ScriptService(settings, Collections.emptyMap(), Collections.emptyMap(), () -> 1L);
         clusterService = ClusterServiceUtils.createClusterService(threadPool);
         nodeEnvironment = new NodeEnvironment(settings, environment);
+        threadPoolMergeExecutorService = ThreadPoolMergeExecutorService.maybeCreateThreadPoolMergeExecutorService(
+            threadPool,
+            settings,
+            nodeEnvironment
+        );
         mapperRegistry = new IndicesModule(Collections.emptyList()).getMapperRegistry();
         indexNameExpressionResolver = TestIndexNameExpressionResolver.newInstance(threadPool.getThreadContext());
     }

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

@@ -13,6 +13,7 @@
 import org.elasticsearch.common.util.concurrent.ConcurrentCollections;
 import org.elasticsearch.common.util.concurrent.DeterministicTaskQueue;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
+import org.elasticsearch.env.NodeEnvironment;
 import org.elasticsearch.index.engine.ThreadPoolMergeScheduler.MergeTask;
 import org.elasticsearch.index.engine.ThreadPoolMergeScheduler.Schedule;
 import org.elasticsearch.index.merge.OnGoingMerge;
@@ -21,6 +22,7 @@
 import org.elasticsearch.threadpool.ThreadPool;
 import org.mockito.ArgumentCaptor;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Comparator;
@@ -56,9 +58,13 @@
 
 public class ThreadPoolMergeExecutorServiceTests extends ESTestCase {
 
-    public void testNewMergeTaskIsAbortedWhenThreadPoolIsShutdown() {
-        TestThreadPool testThreadPool = new TestThreadPool("test");
-        ThreadPoolMergeExecutorService threadPoolMergeExecutorService = getThreadPoolMergeExecutorService(testThreadPool);
+    public void testNewMergeTaskIsAbortedWhenThreadPoolIsShutdown() throws IOException {
+        TestThreadPool testThreadPool = new TestThreadPool("test", Settings.EMPTY);
+        ThreadPoolMergeExecutorService threadPoolMergeExecutorService = getThreadPoolMergeExecutorService(
+            testThreadPool,
+            Settings.EMPTY,
+            newNodeEnvironment(Settings.EMPTY)
+        );
         // shutdown the thread pool
         testThreadPool.shutdown();
         MergeTask mergeTask = mock(MergeTask.class);
@@ -80,7 +86,11 @@ public void testEnqueuedAndBackloggedMergesAreStillExecutedWhenThreadPoolIsShutd
             .put(EsExecutors.NODE_PROCESSORS_SETTING.getKey(), mergeExecutorThreadCount)
             .build();
         TestThreadPool testThreadPool = new TestThreadPool("test", settings);
-        ThreadPoolMergeExecutorService threadPoolMergeExecutorService = getThreadPoolMergeExecutorService(testThreadPool);
+        ThreadPoolMergeExecutorService threadPoolMergeExecutorService = getThreadPoolMergeExecutorService(
+            testThreadPool,
+            settings,
+            newNodeEnvironment(settings)
+        );
         var countingListener = new CountingMergeEventListener();
         threadPoolMergeExecutorService.registerMergeEventListener(countingListener);
         assertThat(threadPoolMergeExecutorService.getMaxConcurrentMerges(), equalTo(mergeExecutorThreadCount));
@@ -191,7 +201,11 @@ public void testTargetIORateChangesWhenSubmittingMergeTasks() throws Exception {
             .put(EsExecutors.NODE_PROCESSORS_SETTING.getKey(), mergeExecutorThreadCount)
             .build();
         try (TestThreadPool testThreadPool = new TestThreadPool("test", settings)) {
-            ThreadPoolMergeExecutorService threadPoolMergeExecutorService = getThreadPoolMergeExecutorService(testThreadPool);
+            ThreadPoolMergeExecutorService threadPoolMergeExecutorService = getThreadPoolMergeExecutorService(
+                testThreadPool,
+                settings,
+                newNodeEnvironment(settings)
+            );
             assertThat(threadPoolMergeExecutorService.getMaxConcurrentMerges(), equalTo(mergeExecutorThreadCount));
             Semaphore runMergeSemaphore = new Semaphore(0);
             AtomicInteger submittedIOThrottledMergeTasks = new AtomicInteger();
@@ -271,7 +285,11 @@ public void testIORateIsAdjustedForRunningMergeTasks() throws Exception {
             .put(EsExecutors.NODE_PROCESSORS_SETTING.getKey(), mergeExecutorThreadCount)
             .build();
         try (TestThreadPool testThreadPool = new TestThreadPool("test", settings)) {
-            ThreadPoolMergeExecutorService threadPoolMergeExecutorService = getThreadPoolMergeExecutorService(testThreadPool);
+            ThreadPoolMergeExecutorService threadPoolMergeExecutorService = getThreadPoolMergeExecutorService(
+                testThreadPool,
+                settings,
+                newNodeEnvironment(settings)
+            );
             assertThat(threadPoolMergeExecutorService.getMaxConcurrentMerges(), equalTo(mergeExecutorThreadCount));
             ThreadPoolExecutor threadPoolExecutor = (ThreadPoolExecutor) testThreadPool.executor(ThreadPool.Names.MERGE);
             Semaphore runMergeSemaphore = new Semaphore(0);
@@ -333,38 +351,39 @@ public void testIORateIsAdjustedForRunningMergeTasks() throws Exception {
         }
     }
 
-    public void testIORateAdjustedForSubmittedTasksWhenExecutionRateIsSpeedy() {
+    public void testIORateAdjustedForSubmittedTasksWhenExecutionRateIsSpeedy() throws IOException {
         // the executor runs merge tasks at a faster rate than the rate that merge tasks are submitted
         int submittedVsExecutedRateOutOf1000 = randomIntBetween(0, 250);
         testIORateAdjustedForSubmittedTasks(randomIntBetween(50, 1000), submittedVsExecutedRateOutOf1000, randomIntBetween(0, 5));
         // executor starts running merges only after a considerable amount of merge tasks have already been submitted
         testIORateAdjustedForSubmittedTasks(randomIntBetween(50, 1000), submittedVsExecutedRateOutOf1000, randomIntBetween(5, 50));
     }
 
-    public void testIORateAdjustedForSubmittedTasksWhenExecutionRateIsSluggish() {
+    public void testIORateAdjustedForSubmittedTasksWhenExecutionRateIsSluggish() throws IOException {
         // the executor runs merge tasks at a faster rate than the rate that merge tasks are submitted
         int submittedVsExecutedRateOutOf1000 = randomIntBetween(750, 1000);
         testIORateAdjustedForSubmittedTasks(randomIntBetween(50, 1000), submittedVsExecutedRateOutOf1000, randomIntBetween(0, 5));
         // executor starts running merges only after a considerable amount of merge tasks have already been submitted
         testIORateAdjustedForSubmittedTasks(randomIntBetween(50, 1000), submittedVsExecutedRateOutOf1000, randomIntBetween(5, 50));
     }
 
-    public void testIORateAdjustedForSubmittedTasksWhenExecutionRateIsOnPar() {
+    public void testIORateAdjustedForSubmittedTasksWhenExecutionRateIsOnPar() throws IOException {
         // the executor runs merge tasks at a faster rate than the rate that merge tasks are submitted
         int submittedVsExecutedRateOutOf1000 = randomIntBetween(250, 750);
         testIORateAdjustedForSubmittedTasks(randomIntBetween(50, 1000), submittedVsExecutedRateOutOf1000, randomIntBetween(0, 5));
         // executor starts running merges only after a considerable amount of merge tasks have already been submitted
         testIORateAdjustedForSubmittedTasks(randomIntBetween(50, 1000), submittedVsExecutedRateOutOf1000, randomIntBetween(5, 50));
     }
 
-    private void testIORateAdjustedForSubmittedTasks(
-        int totalTasksToSubmit,
-        int submittedVsExecutedRateOutOf1000,
-        int initialTasksToSubmit
-    ) {
+    private void testIORateAdjustedForSubmittedTasks(int totalTasksToSubmit, int submittedVsExecutedRateOutOf1000, int initialTasksToSubmit)
+        throws IOException {
         DeterministicTaskQueue mergeExecutorTaskQueue = new DeterministicTaskQueue();
         ThreadPool mergeExecutorThreadPool = mergeExecutorTaskQueue.getThreadPool();
-        ThreadPoolMergeExecutorService threadPoolMergeExecutorService = getThreadPoolMergeExecutorService(mergeExecutorThreadPool);
+        ThreadPoolMergeExecutorService threadPoolMergeExecutorService = getThreadPoolMergeExecutorService(
+            mergeExecutorThreadPool,
+            Settings.EMPTY,
+            newNodeEnvironment(Settings.EMPTY)
+        );
         final AtomicInteger currentlySubmittedMergeTaskCount = new AtomicInteger();
         final AtomicLong targetIORateLimit = new AtomicLong(ThreadPoolMergeExecutorService.START_IO_RATE.getBytes());
         final AtomicReference<MergeTask> lastRunTask = new AtomicReference<>();
@@ -424,7 +443,11 @@ public void testMergeTasksRunConcurrently() throws Exception {
             .put(EsExecutors.NODE_PROCESSORS_SETTING.getKey(), mergeExecutorThreadCount)
             .build();
         try (TestThreadPool testThreadPool = new TestThreadPool("test", settings)) {
-            ThreadPoolMergeExecutorService threadPoolMergeExecutorService = getThreadPoolMergeExecutorService(testThreadPool);
+            ThreadPoolMergeExecutorService threadPoolMergeExecutorService = getThreadPoolMergeExecutorService(
+                testThreadPool,
+                settings,
+                newNodeEnvironment(settings)
+            );
             assertThat(threadPoolMergeExecutorService.getMaxConcurrentMerges(), equalTo(mergeExecutorThreadCount));
             // more merge tasks than max concurrent merges allowed to run concurrently
             int totalMergeTasksCount = mergeExecutorThreadCount + randomIntBetween(1, 5);
@@ -465,7 +488,7 @@ public void testMergeTasksRunConcurrently() throws Exception {
                     assertThat(threadPoolMergeExecutorService.getRunningMergeTasks().size(), is(mergeExecutorThreadCount));
                     // with the other merge tasks enqueued
                     assertThat(
-                        threadPoolMergeExecutorService.getQueuedMergeTasks().size(),
+                        threadPoolMergeExecutorService.getMergeTasksQueueLength(),
                         is(totalMergeTasksCount - mergeExecutorThreadCount - finalCompletedTasksCount)
                     );
                     // also check thread-pool stats for the same
@@ -485,7 +508,7 @@ public void testMergeTasksRunConcurrently() throws Exception {
                     // there are fewer available merges than available threads
                     assertThat(threadPoolMergeExecutorService.getRunningMergeTasks().size(), is(finalRemainingMergeTasksCount));
                     // no more merges enqueued
-                    assertThat(threadPoolMergeExecutorService.getQueuedMergeTasks().size(), is(0));
+                    assertThat(threadPoolMergeExecutorService.getMergeTasksQueueLength(), is(0));
                     // also check thread-pool stats for the same
                     assertThat(threadPoolExecutor.getActiveCount(), is(finalRemainingMergeTasksCount));
                     assertThat(threadPoolExecutor.getQueue().size(), is(0));
@@ -504,7 +527,11 @@ public void testThreadPoolStatsWithBackloggedMergeTasks() throws Exception {
             .put(EsExecutors.NODE_PROCESSORS_SETTING.getKey(), mergeExecutorThreadCount)
             .build();
         try (TestThreadPool testThreadPool = new TestThreadPool("test", settings)) {
-            ThreadPoolMergeExecutorService threadPoolMergeExecutorService = getThreadPoolMergeExecutorService(testThreadPool);
+            ThreadPoolMergeExecutorService threadPoolMergeExecutorService = getThreadPoolMergeExecutorService(
+                testThreadPool,
+                settings,
+                newNodeEnvironment(settings)
+            );
             assertThat(threadPoolMergeExecutorService.getMaxConcurrentMerges(), equalTo(mergeExecutorThreadCount));
             int totalMergeTasksCount = randomIntBetween(1, 10);
             ThreadPoolExecutor threadPoolExecutor = (ThreadPoolExecutor) testThreadPool.executor(ThreadPool.Names.MERGE);
@@ -533,7 +560,7 @@ public void testThreadPoolStatsWithBackloggedMergeTasks() throws Exception {
                     assertThat(threadPoolExecutor.getActiveCount(), is(backloggedMergeTasksList.size()));
                     assertThat(threadPoolExecutor.getQueue().size(), is(0));
                 }
-                assertThat(threadPoolMergeExecutorService.getQueuedMergeTasks().size(), is(0));
+                assertThat(threadPoolMergeExecutorService.getMergeTasksQueueLength(), is(0));
             });
             // re-enqueue backlogged merge tasks
             for (MergeTask backloggedMergeTask : backloggedMergeTasksList) {
@@ -557,7 +584,11 @@ public void testBackloggedMergeTasksExecuteExactlyOnce() throws Exception {
             .put(EsExecutors.NODE_PROCESSORS_SETTING.getKey(), mergeExecutorThreadCount)
             .build();
         try (TestThreadPool testThreadPool = new TestThreadPool("test", settings)) {
-            ThreadPoolMergeExecutorService threadPoolMergeExecutorService = getThreadPoolMergeExecutorService(testThreadPool);
+            ThreadPoolMergeExecutorService threadPoolMergeExecutorService = getThreadPoolMergeExecutorService(
+                testThreadPool,
+                settings,
+                newNodeEnvironment(settings)
+            );
             assertThat(threadPoolMergeExecutorService.getMaxConcurrentMerges(), equalTo(mergeExecutorThreadCount));
             // many merge tasks concurrently
             int mergeTaskCount = randomIntBetween(10, 100);
@@ -613,10 +644,14 @@ public void testBackloggedMergeTasksExecuteExactlyOnce() throws Exception {
         }
     }
 
-    public void testMergeTasksExecuteInSizeOrder() {
+    public void testMergeTasksExecuteInSizeOrder() throws IOException {
         DeterministicTaskQueue mergeExecutorTaskQueue = new DeterministicTaskQueue();
         ThreadPool mergeExecutorThreadPool = mergeExecutorTaskQueue.getThreadPool();
-        ThreadPoolMergeExecutorService threadPoolMergeExecutorService = getThreadPoolMergeExecutorService(mergeExecutorThreadPool);
+        ThreadPoolMergeExecutorService threadPoolMergeExecutorService = getThreadPoolMergeExecutorService(
+            mergeExecutorThreadPool,
+            Settings.EMPTY,
+            newNodeEnvironment(Settings.EMPTY)
+        );
         DeterministicTaskQueue reEnqueueBackloggedTaskQueue = new DeterministicTaskQueue();
         int mergeTaskCount = randomIntBetween(10, 100);
         // sort merge tasks available to run by size
@@ -696,13 +731,21 @@ public void onMergeAborted(OnGoingMerge merge) {
         }
     }
 
-    static ThreadPoolMergeExecutorService getThreadPoolMergeExecutorService(ThreadPool threadPool) {
+    static ThreadPoolMergeExecutorService getThreadPoolMergeExecutorService(
+        ThreadPool threadPool,
+        Settings settings,
+        NodeEnvironment nodeEnvironment
+    ) {
         ThreadPoolMergeExecutorService threadPoolMergeExecutorService = ThreadPoolMergeExecutorService
             .maybeCreateThreadPoolMergeExecutorService(
                 threadPool,
                 randomBoolean()
-                    ? Settings.EMPTY
-                    : Settings.builder().put(ThreadPoolMergeScheduler.USE_THREAD_POOL_MERGE_SCHEDULER_SETTING.getKey(), true).build()
+                    ? settings
+                    : Settings.builder()
+                        .put(settings)
+                        .put(ThreadPoolMergeScheduler.USE_THREAD_POOL_MERGE_SCHEDULER_SETTING.getKey(), true)
+                        .build(),
+                nodeEnvironment
             );
         assertNotNull(threadPoolMergeExecutorService);
         assertTrue(threadPoolMergeExecutorService.allDone());