Adds the _cluster object to the AsyncStatusResponse XContent when Clusters has a non-zero count total (#96535)

Adds the _cluster object to the AsyncStatusResponse XContent when Clusters has a non-zero count total.

When CCS is performed, the _cluster object will have a non-zero `total` count and thus
is useful status information for the user.

When a local cluster only async search is done, the _cluster object will not be present in the AsyncStatusResponse (same as it is now)

A TransportVersion bump was required since Clusters is now an optionally written to the AsyncStatusResponse object.

Author: quux00

server/src/main/java/org/elasticsearch/TransportVersion.java

@@ -132,12 +132,13 @@ private static TransportVersion registerTransportVersion(int id, String uniqueId
     public static final TransportVersion V_8_500_006 = registerTransportVersion(8_500_006, "7BB5621A-80AC-425F-BA88-75543C442F23");
     public static final TransportVersion V_8_500_007 = registerTransportVersion(8_500_007, "77261d43-4149-40af-89c5-7e71e0454fce");
     public static final TransportVersion V_8_500_008 = registerTransportVersion(8_500_008, "8884ab9d-94cd-4bac-aff8-01f2c394f47c");
+    public static final TransportVersion V_8_500_009 = registerTransportVersion(8_500_009, "35091358-fd41-4106-a6e2-d2a1315494c1");
 
     /**
      * Reference to the most recent transport version.
      * This should be the transport version with the highest id.
      */
-    public static final TransportVersion CURRENT = findCurrent(V_8_500_008);
+    public static final TransportVersion CURRENT = findCurrent(V_8_500_009);
 
     /**
      * Reference to the earliest compatible transport version to this version of the codebase.

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

@@ -516,9 +516,17 @@ public Clusters(int total, int successful, int skipped, int remoteClusters, bool
          * We are not tracking number of remote clusters in this search.
          */
         public Clusters(int total, int successful, int skipped) {
-            assert total >= 0 && successful >= 0 && skipped >= 0
+            this(total, successful, skipped, true);
+        }
+
+        /**
+         * @param finalState if true, then do an assert that total = successful + skipped. This is true
+         *                   only when the cluster is in its final state, not an initial or intermediate state.
+         */
+        Clusters(int total, int successful, int skipped, boolean finalState) {
+            assert total >= 0 && successful >= 0 && skipped >= 0 && successful <= total
                 : "total: " + total + " successful: " + successful + " skipped: " + skipped;
-            assert successful <= total && skipped == total - successful
+            assert finalState == false || skipped == total - successful
                 : "total: " + total + " successful: " + successful + " skipped: " + skipped;
             this.total = total;
             this.successful = successful;
@@ -527,8 +535,9 @@ public Clusters(int total, int successful, int skipped) {
             this.ccsMinimizeRoundtrips = false;
         }
 
-        private Clusters(StreamInput in) throws IOException {
-            this(in.readVInt(), in.readVInt(), in.readVInt());
+        public Clusters(StreamInput in) throws IOException {
+            // when coming across the wire, we don't have context to know if this Cluster is in a final state, so set finalState=false
+            this(in.readVInt(), in.readVInt(), in.readVInt(), false);
         }
 
         @Override

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

@@ -241,6 +241,11 @@ synchronized AsyncSearchResponse toAsyncSearchResponse(AsyncSearchTask task, lon
      * @return response representing the status of async search
      */
     synchronized AsyncStatusResponse toStatusResponse(String asyncExecutionId, long startTime, long expirationTime) {
+        SearchResponse.Clusters clustersInStatus = null;
+        if (clusters != null && clusters.getTotal() > 0) {
+            // include clusters in the status if present and not Clusters.EMPTY (the case for local searches only)
+            clustersInStatus = clusters;
+        }
         if (finalResponse != null) {
             return new AsyncStatusResponse(
                 asyncExecutionId,
@@ -252,7 +257,8 @@ synchronized AsyncStatusResponse toStatusResponse(String asyncExecutionId, long
                 finalResponse.getSuccessfulShards(),
                 finalResponse.getSkippedShards(),
                 finalResponse.getShardFailures() != null ? finalResponse.getShardFailures().length : 0,
-                finalResponse.status()
+                finalResponse.status(),
+                clustersInStatus
             );
         }
         if (failure != null) {
@@ -266,7 +272,8 @@ synchronized AsyncStatusResponse toStatusResponse(String asyncExecutionId, long
                 successfulShards,
                 skippedShards,
                 queryFailures == null ? 0 : queryFailures.nonNullLength(),
-                ExceptionsHelper.status(ExceptionsHelper.unwrapCause(failure))
+                ExceptionsHelper.status(ExceptionsHelper.unwrapCause(failure)),
+                clustersInStatus
             );
         }
         return new AsyncStatusResponse(
@@ -279,7 +286,8 @@ synchronized AsyncStatusResponse toStatusResponse(String asyncExecutionId, long
             successfulShards,
             skippedShards,
             queryFailures == null ? 0 : queryFailures.nonNullLength(),
-            null  // for a still running search, completion status is null
+            null,  // for a still running search, completion status is null
+            clustersInStatus
         );
     }