Threadpool merge scheduler

docs/changelog/120869.yaml

@@ -0,0 +1,5 @@
+pr: 120869
+summary: Threadpool merge scheduler
+area: Engine
+type: feature
+issues: []

server/src/internalClusterTest/java/org/elasticsearch/index/engine/InternalEngineMergeIT.java

@@ -8,24 +8,40 @@
  */
 package org.elasticsearch.index.engine;
 
+import org.elasticsearch.action.admin.cluster.node.stats.NodeStats;
+import org.elasticsearch.action.admin.cluster.node.stats.NodesStatsResponse;
 import org.elasticsearch.action.admin.indices.stats.IndicesStatsResponse;
 import org.elasticsearch.action.bulk.BulkRequestBuilder;
 import org.elasticsearch.action.bulk.BulkResponse;
 import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.index.query.QueryBuilders;
 import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.test.ESIntegTestCase.ClusterScope;
 import org.elasticsearch.test.ESIntegTestCase.Scope;
+import org.elasticsearch.threadpool.ThreadPool;
 
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
+import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertHitCount;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertNoFailures;
 import static org.elasticsearch.xcontent.XContentFactory.jsonBuilder;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.lessThan;
 import static org.hamcrest.Matchers.lessThanOrEqualTo;
 
-@ClusterScope(supportsDedicatedMasters = false, numDataNodes = 1, scope = Scope.SUITE)
+@ClusterScope(supportsDedicatedMasters = false, numDataNodes = 1, numClientNodes = 0, scope = Scope.TEST)
 public class InternalEngineMergeIT extends ESIntegTestCase {
 
+    private boolean useThreadPoolMerging;
+
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal, Settings otherSettings) {
+        useThreadPoolMerging = randomBoolean();
+        Settings.Builder settings = Settings.builder().put(super.nodeSettings(nodeOrdinal, otherSettings));
+        settings.put(ThreadPoolMergeScheduler.USE_THREAD_POOL_MERGE_SCHEDULER_SETTING.getKey(), useThreadPoolMerging);
+        return settings.build();
+    }
+
     public void testMergesHappening() throws Exception {
         final int numOfShards = randomIntBetween(1, 5);
         // some settings to keep num segments low
@@ -83,4 +99,60 @@ public void testMergesHappening() throws Exception {
         assertThat(count, lessThanOrEqualTo(upperNumberSegments));
     }
 
+    public void testMergesUseTheMergeThreadPool() throws Exception {
+        final String indexName = randomIdentifier();
+        createIndex(indexName, indexSettings(randomIntBetween(1, 3), 0).build());
+        long id = 0;
+        final int minMerges = randomIntBetween(1, 5);
+        long totalDocs = 0;
+
+        while (true) {
+            int docs = randomIntBetween(100, 200);
+            totalDocs += docs;
+
+            BulkRequestBuilder request = client().prepareBulk();
+            for (int j = 0; j < docs; ++j) {
+                request.add(
+                    new IndexRequest(indexName).id(Long.toString(id++))
+                        .source(jsonBuilder().startObject().field("l", randomLong()).endObject())
+                );
+            }
+            BulkResponse response = request.get();
+            assertNoFailures(response);
+            refresh(indexName);
+
+            var mergesResponse = client().admin().indices().prepareStats(indexName).clear().setMerge(true).get();
+            var primaries = mergesResponse.getIndices().get(indexName).getPrimaries();
+            if (primaries.merge.getTotal() >= minMerges) {
+                break;
+            }
+        }
+
+        forceMerge();
+        refresh(indexName);
+
+        // after a force merge there should only be 1 segment per shard
+        var shardsWithMultipleSegments = getShardSegments().stream()
+            .filter(shardSegments -> shardSegments.getSegments().size() > 1)
+            .toList();
+        assertTrue("there are shards with multiple segments " + shardsWithMultipleSegments, shardsWithMultipleSegments.isEmpty());
+
+        final long expectedTotalDocs = totalDocs;
+        assertHitCount(prepareSearch(indexName).setQuery(QueryBuilders.matchAllQuery()).setTrackTotalHits(true), expectedTotalDocs);
+
+        IndicesStatsResponse indicesStats = client().admin().indices().prepareStats(indexName).setMerge(true).get();
+        long mergeCount = indicesStats.getIndices().get(indexName).getPrimaries().merge.getTotal();
+        NodesStatsResponse nodesStatsResponse = client().admin().cluster().prepareNodesStats().setThreadPool(true).get();
+        assertThat(nodesStatsResponse.getNodes().size(), equalTo(1));
+
+        NodeStats nodeStats = nodesStatsResponse.getNodes().get(0);
+        if (useThreadPoolMerging) {
+            assertThat(
+                nodeStats.getThreadPool().stats().stream().filter(s -> ThreadPool.Names.MERGE.equals(s.name())).findAny().get().completed(),
+                equalTo(mergeCount)
+            );
+        } else {
+            assertTrue(nodeStats.getThreadPool().stats().stream().filter(s -> ThreadPool.Names.MERGE.equals(s.name())).findAny().isEmpty());
+        }
+    }
 }

server/src/internalClusterTest/java/org/elasticsearch/index/shard/IndexShardIT.java

@@ -625,6 +625,7 @@ public static final IndexShard newIndexShard(
             indexService.getIndexEventListener(),
             wrapper,
             indexService.getThreadPool(),
+            indexService.getThreadPoolMergeExecutorService(),
             indexService.getBigArrays(),
             null,
             Collections.emptyList(),

server/src/internalClusterTest/java/org/elasticsearch/indices/IndexingMemoryControllerIT.java

@@ -59,6 +59,7 @@ EngineConfig engineConfigWithLargerIndexingMemory(EngineConfig config) {
             return new EngineConfig(
                 config.getShardId(),
                 config.getThreadPool(),
+                config.getThreadPoolMergeExecutorService(),
                 indexSettings,
                 config.getWarmer(),
                 config.getStore(),

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.ThreadPoolMergeScheduler;
 import org.elasticsearch.indices.IndexingMemoryController;
 import org.elasticsearch.indices.IndicesQueryCache;
 import org.elasticsearch.indices.IndicesRequestCache;
@@ -624,6 +625,7 @@ public void apply(Settings value, Settings current, Settings previous) {
         TDigestExecutionHint.SETTING,
         MergePolicyConfig.DEFAULT_MAX_MERGED_SEGMENT_SETTING,
         MergePolicyConfig.DEFAULT_MAX_TIME_BASED_MERGED_SEGMENT_SETTING,
+        ThreadPoolMergeScheduler.USE_THREAD_POOL_MERGE_SCHEDULER_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/IndexModule.java

@@ -43,6 +43,7 @@
 import org.elasticsearch.index.cache.query.QueryCache;
 import org.elasticsearch.index.engine.Engine;
 import org.elasticsearch.index.engine.EngineFactory;
+import org.elasticsearch.index.engine.ThreadPoolMergeExecutorService;
 import org.elasticsearch.index.mapper.IdFieldMapper;
 import org.elasticsearch.index.mapper.MapperMetrics;
 import org.elasticsearch.index.mapper.MapperRegistry;
@@ -470,6 +471,7 @@ public IndexService newIndexService(
         CircuitBreakerService circuitBreakerService,
         BigArrays bigArrays,
         ThreadPool threadPool,
+        ThreadPoolMergeExecutorService threadPoolMergeExecutorService,
         ScriptService scriptService,
         ClusterService clusterService,
         Client client,
@@ -523,6 +525,7 @@ public IndexService newIndexService(
                 circuitBreakerService,
                 bigArrays,
                 threadPool,
+                threadPoolMergeExecutorService,
                 scriptService,
                 clusterService,
                 client,

server/src/main/java/org/elasticsearch/index/IndexService.java

@@ -49,6 +49,7 @@
 import org.elasticsearch.index.cache.query.QueryCache;
 import org.elasticsearch.index.engine.Engine;
 import org.elasticsearch.index.engine.EngineFactory;
+import org.elasticsearch.index.engine.ThreadPoolMergeExecutorService;
 import org.elasticsearch.index.fielddata.FieldDataContext;
 import org.elasticsearch.index.fielddata.IndexFieldData;
 import org.elasticsearch.index.fielddata.IndexFieldDataCache;
@@ -154,6 +155,8 @@ public class IndexService extends AbstractIndexComponent implements IndicesClust
 
     private final AsyncTrimTranslogTask trimTranslogTask;
     private final ThreadPool threadPool;
+    @Nullable
+    private final ThreadPoolMergeExecutorService threadPoolMergeExecutorService;
     private final BigArrays bigArrays;
     private final ScriptService scriptService;
     private final ClusterService clusterService;
@@ -178,6 +181,7 @@ public IndexService(
         CircuitBreakerService circuitBreakerService,
         BigArrays bigArrays,
         ThreadPool threadPool,
+        ThreadPoolMergeExecutorService threadPoolMergeExecutorService,
         ScriptService scriptService,
         ClusterService clusterService,
         Client client,
@@ -261,6 +265,7 @@ public IndexService(
         this.indexFoldersDeletionListener = indexFoldersDeletionListener;
         this.bigArrays = bigArrays;
         this.threadPool = threadPool;
+        this.threadPoolMergeExecutorService = threadPoolMergeExecutorService;
         this.scriptService = scriptService;
         this.clusterService = clusterService;
         this.client = client;
@@ -556,6 +561,7 @@ public synchronized IndexShard createShard(
                 eventListener,
                 readerWrapper,
                 threadPool,
+                threadPoolMergeExecutorService,
                 bigArrays,
                 engineWarmer,
                 searchOperationListeners,
@@ -820,6 +826,10 @@ public ThreadPool getThreadPool() {
         return threadPool;
     }
 
+    public @Nullable ThreadPoolMergeExecutorService getThreadPoolMergeExecutorService() {
+        return threadPoolMergeExecutorService;
+    }
+
     /**
      * The {@link BigArrays} to use for this index.
      */

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

@@ -58,6 +58,8 @@ public final class EngineConfig {
     private final MapperService mapperService;
     private final IndexStorePlugin.SnapshotCommitSupplier snapshotCommitSupplier;
     private final ThreadPool threadPool;
+    @Nullable
+    private final ThreadPoolMergeExecutorService threadPoolMergeExecutorService;
     private final Engine.Warmer warmer;
     private final Store store;
     private final MergePolicy mergePolicy;
@@ -150,6 +152,7 @@ public Supplier<RetentionLeases> retentionLeasesSupplier() {
     public EngineConfig(
         ShardId shardId,
         ThreadPool threadPool,
+        ThreadPoolMergeExecutorService threadPoolMergeExecutorService,
         IndexSettings indexSettings,
         Engine.Warmer warmer,
         Store store,
@@ -179,6 +182,7 @@ public EngineConfig(
         this.shardId = shardId;
         this.indexSettings = indexSettings;
         this.threadPool = threadPool;
+        this.threadPoolMergeExecutorService = threadPoolMergeExecutorService;
         this.warmer = warmer == null ? (a) -> {} : warmer;
         this.store = store;
         this.mergePolicy = mergePolicy;
@@ -287,6 +291,10 @@ public ThreadPool getThreadPool() {
         return threadPool;
     }
 
+    public @Nullable ThreadPoolMergeExecutorService getThreadPoolMergeExecutorService() {
+        return threadPoolMergeExecutorService;
+    }
+
     /**
      * Returns an {@link org.elasticsearch.index.engine.Engine.Warmer} used to warm new searchers before they are used for searching.
      */