Merge branch 'main' into mp-rest-tests

Author: nielsbauman

docs/changelog/126583.yaml

@@ -0,0 +1,5 @@
+pr: 126583
+summary: Cancel expired async search task when a remote returns its results
+area: CCS
+type: bug
+issues: []

docs/changelog/126840.yaml

@@ -0,0 +1,6 @@
+pr: 126840
+summary: Fix `PolicyStepsRegistry` cache concurrency issue
+area: ILM+SLM
+type: bug
+issues:
+ - 118406

muted-tests.yml

@@ -135,9 +135,6 @@ tests:
 - class: org.elasticsearch.xpack.test.rest.XPackRestIT
   method: test {p0=transform/transforms_unattended/Test unattended put and start}
   issue: https://github.com/elastic/elasticsearch/issues/120019
-- class: org.elasticsearch.xpack.ilm.actions.SearchableSnapshotActionIT
-  method: testUpdatePolicyToAddPhasesYieldsInvalidActionsToBeSkipped
-  issue: https://github.com/elastic/elasticsearch/issues/118406
 - class: org.elasticsearch.xpack.security.QueryableReservedRolesIT
   method: testConfiguredReservedRolesAfterClosingAndOpeningIndex
   issue: https://github.com/elastic/elasticsearch/issues/120127
@@ -414,6 +411,18 @@ tests:
 - class: org.elasticsearch.repositories.gcs.GoogleCloudStorageBlobStoreRepositoryTests
   method: testWriteReadLarge
   issue: https://github.com/elastic/elasticsearch/issues/126903
+- class: org.elasticsearch.repositories.gcs.GoogleCloudStorageBlobStoreRepositoryTests
+  method: testWriteFileMultipleOfChunkSize
+  issue: https://github.com/elastic/elasticsearch/issues/126908
+- class: org.elasticsearch.repositories.gcs.GoogleCloudStorageBlobStoreRepositoryTests
+  method: testSnapshotWithLargeSegmentFiles
+  issue: https://github.com/elastic/elasticsearch/issues/126909
+- class: org.elasticsearch.index.engine.CompletionStatsCacheTests
+  method: testCompletionStatsCache
+  issue: https://github.com/elastic/elasticsearch/issues/126910
+- class: org.elasticsearch.xpack.ml.integration.ClassificationHousePricingIT
+  method: testFeatureImportanceValues
+  issue: https://github.com/elastic/elasticsearch/issues/124341
 
 # Examples:
 #

test/framework/src/main/java/org/elasticsearch/node/MockNode.java

@@ -9,6 +9,8 @@
 
 package org.elasticsearch.node;
 
+import org.elasticsearch.action.search.OnlinePrewarmingService;
+import org.elasticsearch.action.search.OnlinePrewarmingServiceProvider;
 import org.elasticsearch.client.internal.node.NodeClient;
 import org.elasticsearch.cluster.ClusterInfoService;
 import org.elasticsearch.cluster.MockInternalClusterInfoService;
@@ -117,6 +119,10 @@ SearchService newSearchService(
                     tracer
                 );
             }
+            OnlinePrewarmingService onlinePrewarmingService = pluginsService.loadSingletonServiceProvider(
+                OnlinePrewarmingServiceProvider.class,
+                () -> OnlinePrewarmingServiceProvider.DEFAULT
+            ).create(clusterService.getSettings(), threadPool, clusterService);
             return new MockSearchService(
                 clusterService,
                 indicesService,
@@ -126,7 +132,8 @@ SearchService newSearchService(
                 fetchPhase,
                 circuitBreakerService,
                 executorSelector,
-                tracer
+                tracer,
+                onlinePrewarmingService
             );
         }
 

test/framework/src/main/java/org/elasticsearch/search/MockSearchService.java

@@ -85,7 +85,8 @@ public MockSearchService(
         FetchPhase fetchPhase,
         CircuitBreakerService circuitBreakerService,
         ExecutorSelector executorSelector,
-        Tracer tracer
+        Tracer tracer,
+        OnlinePrewarmingService onlinePrewarmingService
     ) {
         super(
             clusterService,
@@ -97,7 +98,7 @@ public MockSearchService(
             circuitBreakerService,
             executorSelector,
             tracer,
-            OnlinePrewarmingService.NOOP
+            onlinePrewarmingService
         );
     }
 

x-pack/plugin/async-search/src/internalClusterTest/java/org/elasticsearch/xpack/search/CCSUsageTelemetryAsyncSearchIT.java

@@ -87,7 +87,7 @@ public void resetSearchListenerPlugin() {
     }
 
     private SubmitAsyncSearchRequest makeSearchRequest(String... indices) {
-        CrossClusterAsyncSearchIT.SearchListenerPlugin.blockQueryPhase();
+        CrossClusterAsyncSearchIT.SearchListenerPlugin.blockLocalQueryPhase();
 
         SubmitAsyncSearchRequest request = new SubmitAsyncSearchRequest(indices);
         request.setCcsMinimizeRoundtrips(randomBoolean());
@@ -220,7 +220,8 @@ public void testCancelledSearch() throws Exception {
         String remoteIndex = (String) testClusterInfo.get("remote.index");
 
         SubmitAsyncSearchRequest searchRequest = makeSearchRequest(localIndex, REMOTE1 + ":" + remoteIndex);
-        CrossClusterAsyncSearchIT.SearchListenerPlugin.blockQueryPhase();
+        CrossClusterAsyncSearchIT.SearchListenerPlugin.blockLocalQueryPhase();
+        CrossClusterAsyncSearchIT.SearchListenerPlugin.blockRemoteQueryPhase();
 
         String nodeName = cluster(LOCAL_CLUSTER).getRandomNodeName();
         final AsyncSearchResponse response = cluster(LOCAL_CLUSTER).client(nodeName)
@@ -232,7 +233,8 @@ public void testCancelledSearch() throws Exception {
             response.decRef();
             assertTrue(response.isRunning());
         }
-        CrossClusterAsyncSearchIT.SearchListenerPlugin.waitSearchStarted();
+        CrossClusterAsyncSearchIT.SearchListenerPlugin.waitLocalSearchStarted();
+        CrossClusterAsyncSearchIT.SearchListenerPlugin.waitRemoteSearchStarted();
 
         ActionFuture<ListTasksResponse> cancelFuture;
         try {
@@ -290,7 +292,8 @@ public void testCancelledSearch() throws Exception {
                 assertTrue(taskInfo.description(), taskInfo.cancelled());
             }
         } finally {
-            CrossClusterAsyncSearchIT.SearchListenerPlugin.allowQueryPhase();
+            CrossClusterAsyncSearchIT.SearchListenerPlugin.allowLocalQueryPhase();
+            CrossClusterAsyncSearchIT.SearchListenerPlugin.allowRemoteQueryPhase();
         }
 
         assertBusy(() -> assertTrue(cancelFuture.isDone()));
@@ -314,7 +317,7 @@ private CCSTelemetrySnapshot getTelemetrySnapshot(String nodeName) {
     }
 
     private Map<String, Object> setupClusters() {
-        String localIndex = "demo";
+        String localIndex = "local";
         int numShardsLocal = randomIntBetween(2, 10);
         Settings localSettings = indexSettings(numShardsLocal, randomIntBetween(0, 1)).build();
         assertAcked(
@@ -326,7 +329,7 @@ private Map<String, Object> setupClusters() {
         );
         indexDocs(client(LOCAL_CLUSTER), localIndex);
 
-        String remoteIndex = "prod";
+        String remoteIndex = "remote";
         int numShardsRemote = randomIntBetween(2, 10);
         for (String clusterAlias : remoteClusterAlias()) {
             final InternalTestCluster remoteCluster = cluster(clusterAlias);