testMergesExecuteInSizeOrder

Author: albertzaharovits

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

@@ -84,6 +84,21 @@ public ThreadPoolMergeScheduler(
         this.threadPoolMergeExecutorService = threadPoolMergeExecutorService;
     }
 
+    // used for tests
+    ThreadPoolMergeScheduler(
+        ShardId shardId,
+        MergeSchedulerConfig mergeSchedulerConfig,
+        Logger logger,
+        MergeTracking mergeTracking,
+        ThreadPoolMergeExecutorService threadPoolMergeExecutorService
+    ) {
+        this.shardId = shardId;
+        this.config = mergeSchedulerConfig;
+        this.logger = logger;
+        this.mergeTracking = mergeTracking;
+        this.threadPoolMergeExecutorService = threadPoolMergeExecutorService;
+    }
+
     @Override
     public Set<OnGoingMerge> onGoingMerges() {
         return mergeTracking.onGoingMerges();
@@ -283,6 +298,7 @@ public Directory wrapForMerge(MergePolicy.OneMerge merge, Directory in) {
         // the combined IO rate per node is, roughly, 'thread_pool_size * merge_queue#targetMBPerSec', as
         // the per-thread IO rate is updated, best effort, for all running merge threads concomitantly.
         if (merge.isAborted()) {
+            // merges can theoretically be aborted at any moment
             return in;
         }
         MergeTask mergeTask = currentlyRunningMergeTasks.get(merge);
@@ -408,7 +424,7 @@ void abortOnGoingMerge() {
         }
 
         long estimatedMergeSize() {
-            return onGoingMerge.getMerge().estimatedMergeBytes;
+            return onGoingMerge.getMerge().getStoreMergeInfo().estimatedMergeBytes();
         }
 
         @Override

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

@@ -570,7 +570,7 @@ public void testMergeTasksExecuteInSizeOrder() {
         }
     }
 
-    private boolean runOneTask(DeterministicTaskQueue deterministicTaskQueue) {
+    private static boolean runOneTask(DeterministicTaskQueue deterministicTaskQueue) {
         while (deterministicTaskQueue.hasAnyTasks()) {
             if (deterministicTaskQueue.hasRunnableTasks()) {
                 deterministicTaskQueue.runRandomTask();

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

@@ -10,7 +10,7 @@
 package org.elasticsearch.index.engine;
 
 import org.apache.lucene.index.MergePolicy;
-import org.apache.lucene.index.MergeScheduler;
+import org.apache.lucene.index.MergeScheduler.MergeSource;
 import org.apache.lucene.index.MergeTrigger;
 import org.apache.lucene.store.MergeInfo;
 import org.elasticsearch.common.settings.Settings;
@@ -24,32 +24,75 @@
 import org.junit.Before;
 import org.mockito.ArgumentCaptor;
 
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.lessThanOrEqualTo;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 public class ThreadPoolMergeSchedulerTests extends ESTestCase {
 
     DeterministicTaskQueue deterministicTaskQueue;
-    ThreadPool testThreadPool;
+    ThreadPool mergesExecutorThreadPool;
     Settings settingsWithMergeScheduler;
     IndexSettings indexSettings;
     ThreadPoolMergeExecutorService threadPoolMergeExecutorService;
 
     @Before
     public void setUpThreadPool() {
         deterministicTaskQueue = new DeterministicTaskQueue();
-        testThreadPool = deterministicTaskQueue.getThreadPool();
+        mergesExecutorThreadPool = deterministicTaskQueue.getThreadPool();
         settingsWithMergeScheduler = Settings.builder()
             .put(ThreadPoolMergeScheduler.USE_THREAD_POOL_MERGE_SCHEDULER_SETTING.getKey(), true)
             .build();
         indexSettings = IndexSettingsModule.newIndexSettings("index", settingsWithMergeScheduler);
         threadPoolMergeExecutorService = ThreadPoolMergeExecutorService.maybeCreateThreadPoolMergeExecutorService(
-            testThreadPool,
+            mergesExecutorThreadPool,
             settingsWithMergeScheduler
         );
     }
 
+    public void testMergesExecuteInSizeOrder() throws IOException {
+        try (
+            ThreadPoolMergeScheduler threadPoolMergeScheduler = new ThreadPoolMergeScheduler(
+                new ShardId("index", "_na_", 1),
+                indexSettings,
+                threadPoolMergeExecutorService
+            )
+        ) {
+            List<MergePolicy.OneMerge> executedMergesList = new ArrayList<>();
+            int mergeCount = randomIntBetween(2, 10);
+            for (int i = 0; i < mergeCount; i++) {
+                MergeSource mergeSource = mock(MergeSource.class);
+                MergePolicy.OneMerge oneMerge = mock(MergePolicy.OneMerge.class);
+                when(oneMerge.getStoreMergeInfo()).thenReturn(
+                        new MergeInfo(randomNonNegativeInt(), randomLongBetween(1L, 10L), randomBoolean(), randomFrom(-1, randomNonNegativeInt()))
+                );
+                when(oneMerge.getMergeProgress()).thenReturn(new MergePolicy.OneMergeProgress());
+                when(mergeSource.getNextMerge()).thenReturn(oneMerge, (MergePolicy.OneMerge) null);
+                doAnswer(invocation -> {
+                    executedMergesList.add((MergePolicy.OneMerge) invocation.getArguments()[0]);
+                    return null;
+                }).when(mergeSource).merge(any(MergePolicy.OneMerge.class));
+                threadPoolMergeScheduler.merge(mergeSource, randomFrom(MergeTrigger.values()));
+            }
+            deterministicTaskQueue.runAllTasks();
+            assertThat(executedMergesList.size(), is(mergeCount));
+            // assert merges are executed in ascending size order
+            for (int i = 1; i < mergeCount; i++) {
+                assertThat(executedMergesList.get(i - 1).getStoreMergeInfo().estimatedMergeBytes(),
+                        lessThanOrEqualTo(executedMergesList.get(i).getStoreMergeInfo().estimatedMergeBytes()));
+            }
+        }
+        assertTrue(threadPoolMergeExecutorService.allDone());
+    }
+
     public void testAutoIOThrottleForMergeTasksWhenSchedulerDisablesIt() throws Exception {
         // merge scheduler configured with auto IO throttle disabled
         Settings settings = Settings.builder()
@@ -64,7 +107,7 @@ public void testAutoIOThrottleForMergeTasksWhenSchedulerDisablesIt() throws Exce
             new MergeInfo(randomNonNegativeInt(), randomNonNegativeLong(), randomBoolean(), randomFrom(-1, randomNonNegativeInt()))
         );
         when(oneMerge.getMergeProgress()).thenReturn(oneMergeProgress);
-        MergeScheduler.MergeSource mergeSource = mock(MergeScheduler.MergeSource.class);
+        MergeSource mergeSource = mock(MergeSource.class);
         when(mergeSource.getNextMerge()).thenReturn(oneMerge);
         try (
             ThreadPoolMergeScheduler threadPoolMergeScheduler = new ThreadPoolMergeScheduler(
@@ -99,7 +142,7 @@ public void testAutoIOThrottleForMergeTasks() throws Exception {
             new MergeInfo(randomNonNegativeInt(), randomNonNegativeLong(), randomBoolean(), randomNonNegativeInt())
         );
         when(oneMerge.getMergeProgress()).thenReturn(oneMergeProgress);
-        MergeScheduler.MergeSource mergeSource = mock(MergeScheduler.MergeSource.class);
+        MergeSource mergeSource = mock(MergeSource.class);
         when(mergeSource.getNextMerge()).thenReturn(oneMerge);
         ThreadPoolMergeExecutorService threadPoolMergeExecutorService = mock(ThreadPoolMergeExecutorService.class);
         try (