more cleanup

Signed-off-by: Atri Sharma <atri.jiit@gmail.com>

Author: atris

server/src/main/java/org/opensearch/action/search/QueryPhaseResultConsumer.java

@@ -77,7 +77,7 @@ public class QueryPhaseResultConsumer extends ArraySearchPhaseResults<SearchPhas
     private final Executor executor;
     private final CircuitBreaker circuitBreaker;
     private final SearchPhaseController controller;
-    protected final SearchProgressListener progressListener;
+    private final SearchProgressListener progressListener;
     private final ReduceContextBuilder aggReduceContextBuilder;
     private final NamedWriteableRegistry namedWriteableRegistry;
 
@@ -151,6 +151,14 @@ int getBatchReduceSize(int requestBatchedReduceSize, int minBatchReduceSize) {
         return (hasAggs || hasTopDocs) ? Math.min(requestBatchedReduceSize, minBatchReduceSize) : minBatchReduceSize;
     }
 
+    /**
+     * Protected accessor for progressListener to allow subclasses to access it.
+     * @return the search progress listener
+     */
+    protected SearchProgressListener progressListener() {
+        return this.progressListener;
+    }
+
     @Override
     public void close() {
         Releasables.close(pendingMerges);

server/src/main/java/org/opensearch/action/search/SearchRequest.java

@@ -281,9 +281,14 @@ public SearchRequest(StreamInput in) throws IOException {
         if (in.getVersion().onOrAfter(Version.V_2_12_0)) {
             phaseTook = in.readOptionalBoolean();
         }
-        // Read streaming fields
-        streamingScoring = in.readBoolean();
-        streamingSearchMode = in.readOptionalString();
+        // Read streaming fields - gated on version for BWC
+        if (in.getVersion().onOrAfter(Version.V_3_3_0)) {
+            streamingScoring = in.readBoolean();
+            streamingSearchMode = in.readOptionalString();
+        } else {
+            streamingScoring = false;
+            streamingSearchMode = null;
+        }
     }
 
     @Override
@@ -318,9 +323,11 @@ public void writeTo(StreamOutput out) throws IOException {
         if (out.getVersion().onOrAfter(Version.V_2_12_0)) {
             out.writeOptionalBoolean(phaseTook);
         }
-        // Write streaming fields
-        out.writeBoolean(streamingScoring);
-        out.writeOptionalString(streamingSearchMode);
+        // Write streaming fields - gated on version for BWC
+        if (out.getVersion().onOrAfter(Version.V_3_3_0)) {
+            out.writeBoolean(streamingScoring);
+            out.writeOptionalString(streamingSearchMode);
+        }
     }
 
     @Override

server/src/main/java/org/opensearch/search/internal/ShardSearchRequest.java

@@ -276,9 +276,14 @@ public ShardSearchRequest(StreamInput in) throws IOException {
         keepAlive = in.readOptionalTimeValue();
         originalIndices = OriginalIndices.readOriginalIndices(in);
         assert keepAlive == null || readerId != null : "readerId: " + readerId + " keepAlive: " + keepAlive;
-        // Read streaming fields
-        streamingSearch = in.readBoolean();
-        streamingSearchMode = in.readOptionalString();
+        // Read streaming fields - gated on version for BWC
+        if (in.getVersion().onOrAfter(Version.V_3_3_0)) {
+            streamingSearch = in.readBoolean();
+            streamingSearchMode = in.readOptionalString();
+        } else {
+            streamingSearch = false;
+            streamingSearchMode = null;
+        }
     }
 
     public ShardSearchRequest(ShardSearchRequest clone) {
@@ -311,9 +316,11 @@ public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
         innerWriteTo(out, false);
         OriginalIndices.writeOriginalIndices(originalIndices, out);
-        // Write streaming fields after OriginalIndices
-        out.writeBoolean(streamingSearch);
-        out.writeOptionalString(streamingSearchMode);
+        // Write streaming fields after OriginalIndices - gated on version for BWC
+        if (out.getVersion().onOrAfter(Version.V_3_3_0)) {
+            out.writeBoolean(streamingSearch);
+            out.writeOptionalString(streamingSearchMode);
+        }
     }
 
     protected final void innerWriteTo(StreamOutput out, boolean asKey) throws IOException {