Cancel expired async search task when a remote returns its results (#126583) (#126931)

A while ago we enabled using ccs_minimize_roundtrips in async search.
This makes it possible for users of async search to send a single search
request per remote cluster, and minimize the impact of network latency.

With non minimized roundtrips, we have pretty recurring cancellation checks:
as part of the execution, we detect that a task expired whenever each shard comes
back with its results.

In a scenario where the coord node does not hold data, or only remote data is
targeted by an async search, we have much less chance of detecting cancellation
if roundtrips are minimized. The local coordinator would do nothing other than
waiting for the minimized results from each remote cluster.
One scenario where we can check for cancellation is when each cluster comes
back with its full set of results. This commit adds such check, plus some testing
for async search cancellation with minimized roundtrips.

Author: javanna

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: []

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);