Cut over SearchResponse and SearchTemplateResponse to Writeable (#41855)

Relates to #34389

Author: javanna

client/client-benchmark-noop-api-plugin/src/main/java/org/elasticsearch/plugin/noop/action/search/NoopSearchAction.java

@@ -20,17 +20,23 @@
 
 import org.elasticsearch.action.Action;
 import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.common.io.stream.Writeable;
 
 public class NoopSearchAction extends Action<SearchResponse> {
     public static final NoopSearchAction INSTANCE = new NoopSearchAction();
     public static final String NAME = "mock:data/read/search";
 
-    public NoopSearchAction() {
+    private NoopSearchAction() {
         super(NAME);
     }
 
     @Override
     public SearchResponse newResponse() {
-        return new SearchResponse();
+        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
+    }
+
+    @Override
+    public Writeable.Reader<SearchResponse> getResponseReader() {
+        return SearchResponse::new;
     }
 }

modules/lang-mustache/src/main/java/org/elasticsearch/script/mustache/MultiSearchTemplateResponse.java

@@ -49,8 +49,7 @@ public static class Item implements Writeable {
 
         private Item(StreamInput in) throws IOException {
             if (in.readBoolean()) {
-                this.response = new SearchTemplateResponse();
-                response.readFrom(in);
+                this.response = new SearchTemplateResponse(in);
                 this.exception = null;
             } else {
                 exception = in.readException();

modules/lang-mustache/src/main/java/org/elasticsearch/script/mustache/SearchTemplateAction.java

@@ -20,6 +20,7 @@
 package org.elasticsearch.script.mustache;
 
 import org.elasticsearch.action.Action;
+import org.elasticsearch.common.io.stream.Writeable;
 
 public class SearchTemplateAction extends Action<SearchTemplateResponse> {
 
@@ -32,6 +33,11 @@ private SearchTemplateAction() {
 
     @Override
     public SearchTemplateResponse newResponse() {
-        return new SearchTemplateResponse();
+        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
+    }
+
+    @Override
+    public Writeable.Reader<SearchTemplateResponse> getResponseReader() {
+        return SearchTemplateResponse::new;
     }
 }

modules/lang-mustache/src/main/java/org/elasticsearch/script/mustache/SearchTemplateResponse.java

@@ -48,6 +48,12 @@ public class SearchTemplateResponse extends ActionResponse implements StatusToXC
     SearchTemplateResponse() {
     }
 
+    SearchTemplateResponse(StreamInput in) throws IOException {
+        super(in);
+        source = in.readOptionalBytesReference();
+        response = in.readOptionalWriteable(SearchResponse::new);
+    }
+
     public BytesReference getSource() {
         return source;
     }
@@ -81,10 +87,8 @@ public void writeTo(StreamOutput out) throws IOException {
     }
 
     @Override
-    public void readFrom(StreamInput in) throws IOException {
-        super.readFrom(in);
-        source = in.readOptionalBytesReference();
-        response = in.readOptionalStreamable(SearchResponse::new);
+    public void readFrom(StreamInput in) {
+        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
     }
 
     public static SearchTemplateResponse fromXContent(XContentParser parser) throws IOException {

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

@@ -70,8 +70,7 @@ public Item(SearchResponse response, Exception exception) {
 
         Item(StreamInput in) throws IOException{
             if (in.readBoolean()) {
-                this.response = new SearchResponse();
-                this.response.readFrom(in);
+                this.response = new SearchResponse(in);
                 this.exception = null;
             } else {
                 this.exception = in.readException();

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

@@ -20,6 +20,7 @@
 package org.elasticsearch.action.search;
 
 import org.elasticsearch.action.Action;
+import org.elasticsearch.common.io.stream.Writeable;
 
 public class SearchAction extends Action<SearchResponse> {
 
@@ -32,6 +33,11 @@ private SearchAction() {
 
     @Override
     public SearchResponse newResponse() {
-        return new SearchResponse();
+        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
+    }
+
+    @Override
+    public Writeable.Reader<SearchResponse> getResponseReader() {
+        return SearchResponse::new;
     }
 }

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

@@ -66,23 +66,33 @@ public class SearchResponse extends ActionResponse implements StatusToXContentOb
     private static final ParseField TERMINATED_EARLY = new ParseField("terminated_early");
     private static final ParseField NUM_REDUCE_PHASES = new ParseField("num_reduce_phases");
 
-    private SearchResponseSections internalResponse;
-
-    private String scrollId;
-
-    private int totalShards;
-
-    private int successfulShards;
-
-    private int skippedShards;
-
-    private ShardSearchFailure[] shardFailures;
-
-    private Clusters clusters;
-
-    private long tookInMillis;
-
-    public SearchResponse() {
+    private final SearchResponseSections internalResponse;
+    private final String scrollId;
+    private final int totalShards;
+    private final int successfulShards;
+    private final int skippedShards;
+    private final ShardSearchFailure[] shardFailures;
+    private final Clusters clusters;
+    private final long tookInMillis;
+
+    public SearchResponse(StreamInput in) throws IOException {
+        super(in);
+        internalResponse = new InternalSearchResponse(in);
+        totalShards = in.readVInt();
+        successfulShards = in.readVInt();
+        int size = in.readVInt();
+        if (size == 0) {
+            shardFailures = ShardSearchFailure.EMPTY_ARRAY;
+        } else {
+            shardFailures = new ShardSearchFailure[size];
+            for (int i = 0; i < shardFailures.length; i++) {
+                shardFailures[i] = readShardSearchFailure(in);
+            }
+        }
+        clusters = new Clusters(in);
+        scrollId = in.readOptionalString();
+        tookInMillis = in.readVLong();
+        skippedShards = in.readVInt();
     }
 
     public SearchResponse(SearchResponseSections internalResponse, String scrollId, int totalShards, int successfulShards,
@@ -193,10 +203,6 @@ public String getScrollId() {
         return scrollId;
     }
 
-    public void scrollId(String scrollId) {
-        this.scrollId = scrollId;
-    }
-
     /**
      * If profiling was enabled, this returns an object containing the profile results from
      * each shard.  If profiling was not enabled, this will return null
@@ -355,24 +361,8 @@ static SearchResponse innerFromXContent(XContentParser parser) throws IOExceptio
     }
 
     @Override
-    public void readFrom(StreamInput in) throws IOException {
-        super.readFrom(in);
-        internalResponse = new InternalSearchResponse(in);
-        totalShards = in.readVInt();
-        successfulShards = in.readVInt();
-        int size = in.readVInt();
-        if (size == 0) {
-            shardFailures = ShardSearchFailure.EMPTY_ARRAY;
-        } else {
-            shardFailures = new ShardSearchFailure[size];
-            for (int i = 0; i < shardFailures.length; i++) {
-                shardFailures[i] = readShardSearchFailure(in);
-            }
-        }
-        clusters = new Clusters(in);
-        scrollId = in.readOptionalString();
-        tookInMillis = in.readVLong();
-        skippedShards = in.readVInt();
+    public void readFrom(StreamInput in) {
+        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
     }
 
     @Override

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

@@ -20,6 +20,7 @@
 package org.elasticsearch.action.search;
 
 import org.elasticsearch.action.Action;
+import org.elasticsearch.common.io.stream.Writeable;
 
 public class SearchScrollAction extends Action<SearchResponse> {
 
@@ -32,6 +33,11 @@ private SearchScrollAction() {
 
     @Override
     public SearchResponse newResponse() {
-        return new SearchResponse();
+        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
+    }
+
+    @Override
+    public Writeable.Reader<SearchResponse> getResponseReader() {
+        return SearchResponse::new;
     }
 }

server/src/test/java/org/elasticsearch/action/search/MultiSearchActionTookTests.java

@@ -32,6 +32,7 @@
 import org.elasticsearch.common.UUIDs;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.AtomicArray;
+import org.elasticsearch.search.internal.InternalSearchResponse;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.tasks.TaskManager;
 import org.elasticsearch.test.ESTestCase;
@@ -154,7 +155,8 @@ public void search(final SearchRequest request, final ActionListener<SearchRespo
                 requests.add(request);
                 commonExecutor.execute(() -> {
                     counter.decrementAndGet();
-                    listener.onResponse(new SearchResponse());
+                    listener.onResponse(new SearchResponse(InternalSearchResponse.empty(), null, 0, 0, 0, 0L,
+                        ShardSearchFailure.EMPTY_ARRAY, SearchResponse.Clusters.EMPTY));
                 });
             }
         };

server/src/test/java/org/elasticsearch/action/search/SearchAsyncActionTests.java

@@ -32,6 +32,7 @@
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.search.SearchPhaseResult;
 import org.elasticsearch.search.internal.AliasFilter;
+import org.elasticsearch.search.internal.InternalSearchResponse;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.transport.Transport;
 import org.elasticsearch.transport.TransportException;
@@ -406,13 +407,17 @@ static GroupShardsIterator<SearchShardIterator> getShardsIter(String index, Orig
     }
 
     public static class TestSearchResponse extends SearchResponse {
-        public final Set<ShardId> queried = new HashSet<>();
+        final Set<ShardId> queried = new HashSet<>();
+
+        TestSearchResponse() {
+            super(InternalSearchResponse.empty(), null, 0, 0, 0, 0L, ShardSearchFailure.EMPTY_ARRAY, Clusters.EMPTY);
+        }
     }
 
     public static class TestSearchPhaseResult extends SearchPhaseResult {
         final DiscoveryNode node;
 
-        public TestSearchPhaseResult(long id, DiscoveryNode node) {
+        TestSearchPhaseResult(long id, DiscoveryNode node) {
             this.requestId = id;
             this.node = node;
         }
@@ -427,7 +432,7 @@ public static final class MockConnection implements Transport.Connection {
 
         private final DiscoveryNode node;
 
-        public MockConnection(DiscoveryNode node) {
+        MockConnection(DiscoveryNode node) {
             this.node = node;
         }
 
@@ -438,7 +443,7 @@ public DiscoveryNode getNode() {
 
         @Override
         public void sendRequest(long requestId, String action, TransportRequest request, TransportRequestOptions options)
-            throws IOException, TransportException {
+            throws TransportException {
             throw new UnsupportedOperationException();
         }