async-search is_partial=true when any shard search fails or times out

With the recent addition of per-cluster metadata to the `_clusters` section of the response
for cross-cluster searches (see elastic#97731), the `is_partial` setting in the async-search response,
now acts as a useful summary to end-users that search/aggs data from all shards is potentially incomplete
(not all shards fully searched), which could be for one of 3 reasons:

1. at least one shard was not successfully searched (a PARTIAL search cluster state)
2. at least one cluster (marked as `skip_unavailable`=`true`) was unavailable (or all
   searches on all shards of that cluster failed), causing the cluster to be marked as SKIPPED
3. a search on at least one cluster timed out (`timed_out`=`true`, resulting in a PARTIAL cluster search status)

This commit changes local-only (non-CCS) searches to behave consistently with cross-cluster searches,
namely, if any search on any shard fails or if the search times out, the is_partial flag is set to true.

Closes elastic#98725

server/src/main/java/org/elasticsearch/action/search/SearchResponse.java

@@ -445,6 +445,21 @@ public void writeTo(StreamOutput out) throws IOException {
         }
     }
 
+    /**
+     * If this is a CCS search all the underlying Cluster search states will be evaluated.
+     * If it is a local-only search, the shards info and timedOut field of the SearchResponse will be examined.
+     *
+     * @return true if the search has partial results due to either failed shards, the search is still running
+     * or a search timed out.
+     */
+    public boolean hasPartialResults() {
+        if (clusters.hasClusterObjects()) {
+            return clusters.hasPartialResults();
+        } else {
+            return getFailedShards() > 0 || isTimedOut();
+        }
+    }
+
     @Override
     public String toString() {
         return Strings.toString(this);
@@ -729,6 +744,10 @@ public String toString() {
         }
 
         /**
+         * This method should NOT be called for local-only, non-CCS searches. In that case,
+         * the Clusters object has no useful info and the _shards section of the SearchResponse
+         * should be used. Call the {@link SearchResponse#hasPartialResults} method instead.
+         *
          * @return true if any underlying Cluster objects have PARTIAL, SKIPPED, FAILED or RUNNING status.
          *              or any Cluster is marked as timedOut.
          */

x-pack/plugin/async-search/src/main/java/org/elasticsearch/xpack/search/MutableSearchResponse.java

@@ -119,17 +119,10 @@ assert shardsInResponseMatchExpected(response, ccsMinimizeRoundtrips)
 
         this.responseHeaders = threadContext.getResponseHeaders();
         this.finalResponse = response;
-        this.isPartial = isPartialResponse(response);
+        this.isPartial = response.hasPartialResults();
         this.frozen = true;
     }
 
-    private boolean isPartialResponse(SearchResponse response) {
-        if (response.getClusters() == null) {
-            return true;
-        }
-        return response.getClusters().hasPartialResults();
-    }
-
     /**
      * Updates the response with a fatal failure. This method preserves the partial response
      * received from previous updates

x-pack/plugin/async-search/src/test/java/org/elasticsearch/xpack/search/AsyncSearchSingleNodeTests.java

@@ -99,7 +99,7 @@ public void testFetchFailuresOnlySomeShards() throws Exception {
         AsyncSearchResponse asyncSearchResponse = client().execute(SubmitAsyncSearchAction.INSTANCE, submitAsyncSearchRequest).actionGet();
 
         assertFalse(asyncSearchResponse.isRunning());
-        assertFalse(asyncSearchResponse.isPartial());
+        assertTrue(asyncSearchResponse.isPartial());
         assertNull(asyncSearchResponse.getFailure());
         SearchResponse searchResponse = asyncSearchResponse.getSearchResponse();
         assertEquals(10, searchResponse.getTotalShards());

x-pack/plugin/async-search/src/test/java/org/elasticsearch/xpack/search/AsyncSearchTaskTests.java

@@ -310,7 +310,15 @@ public void testWithFetchFailures() throws InterruptedException {
         ((AsyncSearchTask.Listener) task.getProgressListener()).onResponse(
             newSearchResponse(totalShards, totalShards - numFetchFailures, numSkippedShards, shardSearchFailures)
         );
-        assertCompletionListeners(task, totalShards, totalShards - numFetchFailures, numSkippedShards, numFetchFailures, false, false);
+        assertCompletionListeners(
+            task,
+            totalShards,
+            totalShards - numFetchFailures,
+            numSkippedShards,
+            numFetchFailures,
+            numFetchFailures > 0,
+            false
+        );
     }
 
     public void testFatalFailureDuringFetch() throws InterruptedException {