Merge branch 'main' into rehydrate_source_vectors_nested

Author: jimczi

rest-api-spec/src/main/resources/rest-api-spec/api/inference.put_custom.json

@@ -0,0 +1,35 @@
+{
+  "inference.put_custom": {
+    "documentation": {
+      "url": "https://www.elastic.co/docs/api/doc/elasticsearch/operation/operation-inference-put-custom",
+      "description": "Configure a custom inference endpoint"
+    },
+    "stability": "stable",
+    "visibility": "public",
+    "headers": {
+      "accept": ["application/json"],
+      "content_type": ["application/json"]
+    },
+    "url": {
+      "paths": [
+        {
+          "path": "/_inference/{task_type}/{custom_inference_id}",
+          "methods": ["PUT"],
+          "parts": {
+            "task_type": {
+              "type": "string",
+              "description": "The task type"
+            },
+            "custom_inference_id": {
+              "type": "string",
+              "description": "The inference Id"
+            }
+          }
+        }
+      ]
+    },
+    "body": {
+      "description": "The inference endpoint's task and service settings"
+    }
+  }
+}

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

@@ -16,6 +16,7 @@
 import org.elasticsearch.action.admin.indices.segments.IndexShardSegments;
 import org.elasticsearch.action.admin.indices.segments.IndicesSegmentResponse;
 import org.elasticsearch.action.admin.indices.segments.ShardSegments;
+import org.elasticsearch.action.admin.indices.stats.IndicesStatsResponse;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.CollectionUtils;
 import org.elasticsearch.common.util.concurrent.ConcurrentCollections;
@@ -44,6 +45,7 @@
 import java.util.concurrent.atomic.AtomicReference;
 
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAllSuccessful;
+import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.is;
@@ -272,12 +274,11 @@ public void testMergingFallsBehindAndThenCatchesUp() throws Exception {
             assertThat(testEnginePlugin.enqueuedMergesSet.size(), is(0));
             testEnginePlugin.mergeExecutorServiceReference.get().allDone();
         }, 1, TimeUnit.MINUTES);
-        var segmentsCountAfterMergingCaughtUp = getSegmentsCountForAllShards("index");
-        // force merge should be a noop after all available merging was done
-        assertAllSuccessful(indicesAdmin().prepareForceMerge("index").get());
-        var segmentsCountAfterForceMerge = getSegmentsCountForAllShards("index");
-        assertThat(segmentsCountAfterForceMerge, is(segmentsCountAfterMergingCaughtUp));
-        // let's also run a force-merge to 1 segment
+        // indices stats says that no merge is currently running (meaning merging did catch up)
+        IndicesStatsResponse indicesStatsResponse = client().admin().indices().prepareStats("index").setMerge(true).get();
+        long currentMergeCount = indicesStatsResponse.getIndices().get("index").getPrimaries().merge.getCurrent();
+        assertThat(currentMergeCount, equalTo(0L));
+        // run a force-merge to 1 segment to make sure nothing is broken
         assertAllSuccessful(indicesAdmin().prepareForceMerge("index").setMaxNumSegments(1).get());
         assertAllSuccessful(indicesAdmin().prepareRefresh("index").get());
         // assert one segment per shard
@@ -292,20 +293,6 @@ public void testMergingFallsBehindAndThenCatchesUp() throws Exception {
         }
     }
 
-    private int getSegmentsCountForAllShards(String indexName) {
-        // refresh, otherwise we'd be still seeing the old merged-away segments
-        assertAllSuccessful(indicesAdmin().prepareRefresh(indexName).get());
-        int count = 0;
-        IndicesSegmentResponse indicesSegmentResponse = indicesAdmin().prepareSegments(indexName).get();
-        Iterator<IndexShardSegments> indexShardSegmentsIterator = indicesSegmentResponse.getIndices().get(indexName).iterator();
-        while (indexShardSegmentsIterator.hasNext()) {
-            for (ShardSegments segments : indexShardSegmentsIterator.next()) {
-                count += segments.getSegments().size();
-            }
-        }
-        return count;
-    }
-
     private TestEnginePlugin getTestEnginePlugin() {
         return getInstanceFromNode(PluginsService.class).filterPlugins(TestEnginePlugin.class).toList().get(0);
     }

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

@@ -328,6 +328,7 @@ public void testIORateIsAdjustedForAllRunningMergeTasks() throws Exception {
             ThreadPoolExecutor threadPoolExecutor = (ThreadPoolExecutor) testThreadPool.executor(ThreadPool.Names.MERGE);
             Semaphore runMergeSemaphore = new Semaphore(0);
             Set<MergeTask> currentlyRunningMergeTasksSet = ConcurrentCollections.newConcurrentSet();
+
             while (mergesStillToComplete > 0) {
                 if (mergesStillToSubmit > 0 && (currentlyRunningMergeTasksSet.isEmpty() || randomBoolean())) {
                     MergeTask mergeTask = mock(MergeTask.class);