Nits

albertzaharovits · albertzaharovits · commit b0e2da0254c5 · 2025-02-23T17:17:03.000+02:00
diff --git a/server/src/internalClusterTest/java/org/elasticsearch/index/shard/IndexShardIT.java b/server/src/internalClusterTest/java/org/elasticsearch/index/shard/IndexShardIT.java
@@ -625,7 +625,7 @@ public static final IndexShard newIndexShard(
             indexService.getIndexEventListener(),
             wrapper,
             indexService.getThreadPool(),
-            indexService.getThreadPoolMergeQueue(),
+            indexService.getThreadPoolMergeExecutorService(),
             indexService.getBigArrays(),
             null,
             Collections.emptyList(),
diff --git a/server/src/internalClusterTest/java/org/elasticsearch/indices/IndexingMemoryControllerIT.java b/server/src/internalClusterTest/java/org/elasticsearch/indices/IndexingMemoryControllerIT.java
@@ -59,7 +59,7 @@ EngineConfig engineConfigWithLargerIndexingMemory(EngineConfig config) {
             return new EngineConfig(
                 config.getShardId(),
                 config.getThreadPool(),
-                config.getThreadPoolMergeQueue(),
+                config.getThreadPoolMergeExecutorService(),
                 indexSettings,
                 config.getWarmer(),
                 config.getStore(),
diff --git a/server/src/main/java/org/elasticsearch/index/IndexService.java b/server/src/main/java/org/elasticsearch/index/IndexService.java
@@ -826,7 +826,7 @@ public ThreadPool getThreadPool() {
         return threadPool;
     }
 
-    public @Nullable ThreadPoolMergeExecutorService getThreadPoolMergeQueue() {
+    public @Nullable ThreadPoolMergeExecutorService getThreadPoolMergeExecutorService() {
         return threadPoolMergeExecutorService;
     }
 
diff --git a/server/src/main/java/org/elasticsearch/index/engine/EngineConfig.java b/server/src/main/java/org/elasticsearch/index/engine/EngineConfig.java
@@ -291,7 +291,7 @@ public ThreadPool getThreadPool() {
         return threadPool;
     }
 
-    public @Nullable ThreadPoolMergeExecutorService getThreadPoolMergeQueue() {
+    public @Nullable ThreadPoolMergeExecutorService getThreadPoolMergeExecutorService() {
         return threadPoolMergeExecutorService;
     }
 
diff --git a/server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java b/server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java
@@ -257,7 +257,7 @@ public InternalEngine(EngineConfig engineConfig) {
             mergeScheduler = createMergeScheduler(
                 engineConfig.getShardId(),
                 engineConfig.getIndexSettings(),
-                engineConfig.getThreadPoolMergeQueue()
+                engineConfig.getThreadPoolMergeExecutorService()
             );
             scheduler = mergeScheduler.getMergeScheduler();
             throttle = new IndexThrottle();
diff --git a/server/src/main/java/org/elasticsearch/index/engine/ThreadPoolMergeExecutorService.java b/server/src/main/java/org/elasticsearch/index/engine/ThreadPoolMergeExecutorService.java
@@ -36,7 +36,7 @@ public class ThreadPoolMergeExecutorService {
      * Initial value for IO write rate limit of individual merge tasks when doAutoIOThrottle is true
      */
     private static final ByteSizeValue START_IO_RATE = ByteSizeValue.ofMb(20L);
-    private final AtomicInteger submittedIOThrottledMergeTasksCount = new AtomicInteger();
+    private final AtomicInteger currentlySubmittedIOThrottledMergeTasksCount = new AtomicInteger();
     private final PriorityBlockingQueue<MergeTask> queuedMergeTasks = new PriorityBlockingQueue<>();
     // the set of all merge tasks currently being executed by merge threads from the pool,
     // in order to be able to update the IO throttle rate of merge tasks also after they have started (while executing)
@@ -49,7 +49,10 @@ public class ThreadPoolMergeExecutorService {
     private final ExecutorService executorService;
     private final int maxConcurrentMerges;
 
-    public static @Nullable ThreadPoolMergeExecutorService maybeCreateThreadPoolMergeQueue(ThreadPool threadPool, Settings settings) {
+    public static @Nullable ThreadPoolMergeExecutorService maybeCreateThreadPoolMergeExecutorService(
+        ThreadPool threadPool,
+        Settings settings
+    ) {
         if (ThreadPoolMergeScheduler.USE_THREAD_POOL_MERGE_SCHEDULER_SETTING.get(settings)) {
             return new ThreadPoolMergeExecutorService(threadPool);
         } else {
@@ -72,7 +75,7 @@ boolean submitMergeTask(MergeTask mergeTask) {
         } else {
             if (mergeTask.supportsIOThrottling()) {
                 // count enqueued merge tasks that support IO auto throttling, and maybe adjust IO rate for all
-                maybeUpdateIORateBytesPerSec(submittedIOThrottledMergeTasksCount.incrementAndGet());
+                maybeUpdateIORateBytesPerSec(currentlySubmittedIOThrottledMergeTasksCount.incrementAndGet());
             }
             // then enqueue the merge task proper
             enqueueMergeTask(mergeTask);
@@ -115,7 +118,7 @@ private boolean enqueueMergeTaskExecution() {
                     }
                 } finally {
                     if (smallestMergeTask != null && smallestMergeTask.supportsIOThrottling()) {
-                        submittedIOThrottledMergeTasksCount.decrementAndGet();
+                        currentlySubmittedIOThrottledMergeTasksCount.decrementAndGet();
                     }
                     if (interrupted) {
                         Thread.currentThread().interrupt();
@@ -199,9 +202,9 @@ private static long newTargetIORateBytesPerSec(long currentTargetIORateBytesPerS
         return ByteSizeValue.ofBytes(targetIORateBytesPerSec.get()).getMbFrac();
     }
 
-    public boolean isEmpty() {
-        boolean isEmpty = queuedMergeTasks.isEmpty() && currentlyRunningMergeTasks.isEmpty();
-        assert isEmpty == false || submittedIOThrottledMergeTasksCount.get() == 0L;
-        return isEmpty;
+    public boolean allDone() {
+        return queuedMergeTasks.isEmpty()
+            && currentlyRunningMergeTasks.isEmpty()
+            && currentlySubmittedIOThrottledMergeTasksCount.get() == 0L;
     }
 }
diff --git a/server/src/main/java/org/elasticsearch/indices/IndicesService.java b/server/src/main/java/org/elasticsearch/indices/IndicesService.java
@@ -287,7 +287,7 @@ protected void doStart() {
     IndicesService(IndicesServiceBuilder builder) {
         this.settings = builder.settings;
         this.threadPool = builder.threadPool;
-        this.threadPoolMergeExecutorService = ThreadPoolMergeExecutorService.maybeCreateThreadPoolMergeQueue(threadPool, settings);
+        this.threadPoolMergeExecutorService = ThreadPoolMergeExecutorService.maybeCreateThreadPoolMergeExecutorService(threadPool, settings);
         this.pluginsService = builder.pluginsService;
         this.nodeEnv = builder.nodeEnv;
         this.parserConfig = XContentParserConfiguration.EMPTY.withDeprecationHandler(LoggingDeprecationHandler.INSTANCE)
@@ -1910,7 +1910,7 @@ public BigArrays getBigArrays() {
     }
 
     @Nullable
-    public ThreadPoolMergeExecutorService getThreadPoolMergeQueue() {
+    public ThreadPoolMergeExecutorService getThreadPoolMergeExecutorService() {
         return threadPoolMergeExecutorService;
     }
 }
diff --git a/server/src/test/java/org/elasticsearch/index/IndexModuleTests.java b/server/src/test/java/org/elasticsearch/index/IndexModuleTests.java
@@ -190,7 +190,7 @@ public void setUp() throws Exception {
             emptyMap()
         );
         threadPool = new TestThreadPool("test");
-        threadPoolMergeExecutorService = ThreadPoolMergeExecutorService.maybeCreateThreadPoolMergeQueue(threadPool, settings);
+        threadPoolMergeExecutorService = ThreadPoolMergeExecutorService.maybeCreateThreadPoolMergeExecutorService(threadPool, settings);
         circuitBreakerService = new NoneCircuitBreakerService();
         PageCacheRecycler pageCacheRecycler = new PageCacheRecycler(settings);
         bigArrays = new BigArrays(pageCacheRecycler, circuitBreakerService, CircuitBreaker.REQUEST);
diff --git a/server/src/test/java/org/elasticsearch/index/engine/InternalEngineTests.java b/server/src/test/java/org/elasticsearch/index/engine/InternalEngineTests.java
@@ -7150,7 +7150,7 @@ public void testNotWarmUpSearcherInEngineCtor() throws Exception {
             EngineConfig configWithWarmer = new EngineConfig(
                 config.getShardId(),
                 config.getThreadPool(),
-                config.getThreadPoolMergeQueue(),
+                config.getThreadPoolMergeExecutorService(),
                 config.getIndexSettings(),
                 warmer,
                 store,
diff --git a/server/src/test/java/org/elasticsearch/index/shard/IndexShardTests.java b/server/src/test/java/org/elasticsearch/index/shard/IndexShardTests.java
@@ -5033,7 +5033,7 @@ public void testCloseShardWhileEngineIsWarming() throws Exception {
             EngineConfig configWithWarmer = new EngineConfig(
                 config.getShardId(),
                 config.getThreadPool(),
-                config.getThreadPoolMergeQueue(),
+                config.getThreadPoolMergeExecutorService(),
                 config.getIndexSettings(),
                 warmer,
                 config.getStore(),
diff --git a/server/src/test/java/org/elasticsearch/index/shard/RefreshListenersTests.java b/server/src/test/java/org/elasticsearch/index/shard/RefreshListenersTests.java
@@ -104,7 +104,7 @@ public void setupListeners() throws Exception {
             .put(ThreadPoolMergeScheduler.USE_THREAD_POOL_MERGE_SCHEDULER_SETTING.getKey(), randomBoolean())
             .build();
         IndexSettings indexSettings = IndexSettingsModule.newIndexSettings("index", settings);
-        threadPoolMergeExecutorService = ThreadPoolMergeExecutorService.maybeCreateThreadPoolMergeQueue(threadPool, settings);
+        threadPoolMergeExecutorService = ThreadPoolMergeExecutorService.maybeCreateThreadPoolMergeExecutorService(threadPool, settings);
         listeners = new RefreshListeners(
             () -> maxListeners,
             () -> engine.refresh("too-many-listeners"),
diff --git a/test/framework/src/main/java/org/elasticsearch/index/engine/EngineTestCase.java b/test/framework/src/main/java/org/elasticsearch/index/engine/EngineTestCase.java
@@ -242,7 +242,7 @@ public void setUp() throws Exception {
         }
         defaultSettings = IndexSettingsModule.newIndexSettings("index", indexSettings());
         threadPool = new TestThreadPool(getClass().getName());
-        threadPoolMergeExecutorService = ThreadPoolMergeExecutorService.maybeCreateThreadPoolMergeQueue(
+        threadPoolMergeExecutorService = ThreadPoolMergeExecutorService.maybeCreateThreadPoolMergeExecutorService(
             threadPool,
             defaultSettings.getNodeSettings()
         );
@@ -278,7 +278,7 @@ public static EngineConfig copy(EngineConfig config, LongSupplier globalCheckpoi
         return new EngineConfig(
             config.getShardId(),
             config.getThreadPool(),
-            config.getThreadPoolMergeQueue(),
+            config.getThreadPoolMergeExecutorService(),
             config.getIndexSettings(),
             config.getWarmer(),
             config.getStore(),
@@ -311,7 +311,7 @@ public EngineConfig copy(EngineConfig config, Analyzer analyzer) {
         return new EngineConfig(
             config.getShardId(),
             config.getThreadPool(),
-            config.getThreadPoolMergeQueue(),
+            config.getThreadPoolMergeExecutorService(),
             config.getIndexSettings(),
             config.getWarmer(),
             config.getStore(),
@@ -344,7 +344,7 @@ public EngineConfig copy(EngineConfig config, MergePolicy mergePolicy) {
         return new EngineConfig(
             config.getShardId(),
             config.getThreadPool(),
-            config.getThreadPoolMergeQueue(),
+            config.getThreadPoolMergeExecutorService(),
             config.getIndexSettings(),
             config.getWarmer(),
             config.getStore(),
@@ -890,7 +890,7 @@ protected EngineConfig config(EngineConfig config, Store store, Path translogPat
         return new EngineConfig(
             config.getShardId(),
             config.getThreadPool(),
-            config.getThreadPoolMergeQueue(),
+            config.getThreadPoolMergeExecutorService(),
             indexSettings,
             config.getWarmer(),
             store,
diff --git a/test/framework/src/main/java/org/elasticsearch/index/shard/IndexShardTestCase.java b/test/framework/src/main/java/org/elasticsearch/index/shard/IndexShardTestCase.java
@@ -169,7 +169,7 @@ public static void addMockCloseImplementation(IndexShard shard) throws IOExcepti
     public void setUp() throws Exception {
         super.setUp();
         threadPool = setUpThreadPool();
-        threadPoolMergeExecutorService = ThreadPoolMergeExecutorService.maybeCreateThreadPoolMergeQueue(threadPool, threadPoolSettings());
+        threadPoolMergeExecutorService = ThreadPoolMergeExecutorService.maybeCreateThreadPoolMergeExecutorService(threadPool, threadPoolSettings());
         writeExecutor = threadPool.executor(ThreadPool.Names.WRITE);
         primaryTerm = randomIntBetween(1, 100); // use random but fixed term for creating shards
         failOnShardFailures();
diff --git a/test/framework/src/main/java/org/elasticsearch/test/InternalTestCluster.java b/test/framework/src/main/java/org/elasticsearch/test/InternalTestCluster.java
@@ -13,7 +13,6 @@
 import com.carrotsearch.randomizedtesting.generators.RandomNumbers;
 import com.carrotsearch.randomizedtesting.generators.RandomPicks;
 import com.carrotsearch.randomizedtesting.generators.RandomStrings;
-
 import org.apache.logging.log4j.Level;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
@@ -1418,12 +1417,15 @@ public void assertConsistentHistoryBetweenTranslogAndLuceneIndex() throws IOExce
         }
     }
 
-    public void assertMergeQueueIsEmpty() throws Exception {
+    public void assertMergeExecutorIsDone() throws Exception {
         assertBusy(() -> {
             for (String nodeName : getNodeNames()) {
                 IndicesService indicesService = getInstance(IndicesService.class, nodeName);
-                if (indicesService.getThreadPoolMergeQueue() != null) {
-                    assertTrue("thread pool merge queue is not empty after test", indicesService.getThreadPoolMergeQueue().isEmpty());
+                if (indicesService.getThreadPoolMergeExecutorService() != null) {
+                    assertTrue(
+                        "thread pool merge executor is not done after test",
+                        indicesService.getThreadPoolMergeExecutorService().allDone()
+                    );
                 }
             }
         });
@@ -2537,7 +2539,7 @@ public synchronized void assertAfterTest() throws Exception {
         assertRequestsFinished();
         assertSearchContextsReleased();
         assertNoInFlightDocsInEngine();
-        assertMergeQueueIsEmpty();
+        assertMergeExecutorIsDone();
         awaitIndexShardCloseAsyncTasks();
         for (NodeAndClient nodeAndClient : nodes.values()) {
             NodeEnvironment env = nodeAndClient.node().getNodeEnvironment();
diff --git a/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/index/engine/FollowingEngineTests.java b/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/index/engine/FollowingEngineTests.java
@@ -94,7 +94,7 @@ public class FollowingEngineTests extends ESTestCase {
     public void setUp() throws Exception {
         super.setUp();
         threadPool = new TestThreadPool("following-engine-tests");
-        threadPoolMergeExecutorService = ThreadPoolMergeExecutorService.maybeCreateThreadPoolMergeQueue(threadPool, Settings.EMPTY);
+        threadPoolMergeExecutorService = ThreadPoolMergeExecutorService.maybeCreateThreadPoolMergeExecutorService(threadPool, Settings.EMPTY);
         index = new Index("index", "uuid");
         shardId = new ShardId(index, 0);
         primaryTerm.set(randomLongBetween(1, Long.MAX_VALUE));

Original file line number	Diff line number	Diff line change
`@@ -826,7 +826,7 @@ public ThreadPool getThreadPool() {`
`826`	`826`	`return threadPool;`
`827`	`827`	`}`
`828`	`828`
`829`		`- public @Nullable ThreadPoolMergeExecutorService getThreadPoolMergeQueue() {`
	`829`	`+ public @Nullable ThreadPoolMergeExecutorService getThreadPoolMergeExecutorService() {`
`830`	`830`	`return threadPoolMergeExecutorService;`
`831`	`831`	`}`
`832`	`832`
Original file line number	Diff line number	Diff line change
`@@ -291,7 +291,7 @@ public ThreadPool getThreadPool() {`
`291`	`291`	`return threadPool;`
`292`	`292`	`}`
`293`	`293`
`294`		`- public @Nullable ThreadPoolMergeExecutorService getThreadPoolMergeQueue() {`
	`294`	`+ public @Nullable ThreadPoolMergeExecutorService getThreadPoolMergeExecutorService() {`
`295`	`295`	`return threadPoolMergeExecutorService;`
`296`	`296`	`}`
`297`	`297`
Original file line number	Diff line number	Diff line change
`@@ -287,7 +287,7 @@ protected void doStart() {`
`287`	`287`	`IndicesService(IndicesServiceBuilder builder) {`
`288`	`288`	`this.settings = builder.settings;`
`289`	`289`	`this.threadPool = builder.threadPool;`
`290`		`- this.threadPoolMergeExecutorService = ThreadPoolMergeExecutorService.maybeCreateThreadPoolMergeQueue(threadPool, settings);`
	`290`	`+ this.threadPoolMergeExecutorService = ThreadPoolMergeExecutorService.maybeCreateThreadPoolMergeExecutorService(threadPool, settings);`
`291`	`291`	`this.pluginsService = builder.pluginsService;`
`292`	`292`	`this.nodeEnv = builder.nodeEnv;`
`293`	`293`	`this.parserConfig = XContentParserConfiguration.EMPTY.withDeprecationHandler(LoggingDeprecationHandler.INSTANCE)`
`@@ -1910,7 +1910,7 @@ public BigArrays getBigArrays() {`
`1910`	`1910`	`}`
`1911`	`1911`
`1912`	`1912`	`@Nullable`
`1913`		`- public ThreadPoolMergeExecutorService getThreadPoolMergeQueue() {`
	`1913`	`+ public ThreadPoolMergeExecutorService getThreadPoolMergeExecutorService() {`
`1914`	`1914`	`return threadPoolMergeExecutorService;`
`1915`	`1915`	`}`
`1916`	`1916`	`}`