Remove early phase failure in batched

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

@@ -82,6 +82,9 @@ public class SearchQueryThenFetchAsyncAction extends AbstractSearchAsyncAction<S
     private static final Logger logger = LogManager.getLogger(SearchQueryThenFetchAsyncAction.class);
 
     private static final TransportVersion BATCHED_QUERY_PHASE_VERSION = TransportVersion.fromName("batched_query_phase_version");
+    private static final TransportVersion BATCHED_RESPONSE_MIGHT_INCLUDE_REDUCTION_FAILURE = TransportVersion.fromName(
+        "batched_response_might_include_reduction_failure"
+    );
 
     private final SearchProgressListener progressListener;
 
@@ -229,8 +232,7 @@ public static final class NodeQueryResponse extends TransportResponse {
 
         public NodeQueryResponse(StreamInput in) throws IOException {
             this.results = in.readArray(i -> i.readBoolean() ? new QuerySearchResult(i) : i.readException(), Object[]::new);
-            boolean hasReductionFailure = in.readBoolean();
-            if (hasReductionFailure) {
+            if (in.getTransportVersion().supports(BATCHED_RESPONSE_MIGHT_INCLUDE_REDUCTION_FAILURE) && in.readBoolean()) {
                 this.reductionFailure = in.readException();
                 this.mergeResult = null;
                 this.topDocsStats = null;
@@ -260,10 +262,14 @@ public void writeTo(StreamOutput out) throws IOException {
                     writePerShardResult(out, (QuerySearchResult) result);
                 }
             }
-            boolean hasReductionFailure = reductionFailure != null;
-            out.writeBoolean(hasReductionFailure);
-            if (hasReductionFailure) {
-                out.writeException(reductionFailure);
+            if (out.getTransportVersion().supports(BATCHED_RESPONSE_MIGHT_INCLUDE_REDUCTION_FAILURE)) {
+                boolean hasReductionFailure = reductionFailure != null;
+                out.writeBoolean(hasReductionFailure);
+                if (hasReductionFailure) {
+                    out.writeException(reductionFailure);
+                } else {
+                    writeMergeResult(out, mergeResult, topDocsStats);
+                }
             } else {
                 writeMergeResult(out, mergeResult, topDocsStats);
             }

server/src/main/resources/transport/definitions/referable/batched_response_might_include_reduction_failure.csv

@@ -0,0 +1 @@
+9198000

server/src/main/resources/transport/upper_bounds/9.3.csv

@@ -1 +1 @@
-dimension_values,9197000
+batched_response_might_include_reduction_failure,9198000