Refactor TransportSingleShardAction to serialize Writeable responses (elastic#41985)

Previously, TransportSingleShardAction required constructing a new
empty response object. This response object's Streamable readFrom
was used. As part of the migration to Writeable, the interface here
was updated to leverage Writeable.Reader.

relates to elastic#34389.

Author: talevy

modules/lang-painless/src/main/java/org/elasticsearch/painless/action/PainlessExecuteAction.java

@@ -101,7 +101,7 @@ private PainlessExecuteAction() {
 
     @Override
     public Response newResponse() {
-        return new Response();
+        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
     }
 
     public static class Request extends SingleShardRequest<Request> implements ToXContentObject {
@@ -381,20 +381,22 @@ public static class Response extends ActionResponse implements ToXContentObject
 
         private Object result;
 
-        Response() {}
-
         Response(Object result) {
             this.result = result;
         }
 
+        Response(StreamInput in) throws IOException {
+            super(in);
+            result = in.readGenericValue();
+        }
+
         public Object getResult() {
             return result;
         }
 
         @Override
         public void readFrom(StreamInput in) throws IOException {
-            super.readFrom(in);
-            result = in.readGenericValue();
+            throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
         }
 
         @Override
@@ -469,8 +471,8 @@ public TransportAction(ThreadPool threadPool, TransportService transportService,
         }
 
         @Override
-        protected Response newResponse() {
-            return new Response();
+        protected Writeable.Reader<Response> getResponseReader() {
+            return Response::new;
         }
 
         @Override

modules/lang-painless/src/test/java/org/elasticsearch/painless/action/PainlessExecuteResponseTests.java

@@ -18,17 +18,57 @@
  */
 package org.elasticsearch.painless.action;
 
-import org.elasticsearch.test.AbstractStreamableTestCase;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.test.AbstractSerializingTestCase;
 
-public class PainlessExecuteResponseTests extends AbstractStreamableTestCase<PainlessExecuteAction.Response> {
+import java.io.IOException;
+
+public class PainlessExecuteResponseTests extends AbstractSerializingTestCase<PainlessExecuteAction.Response> {
 
     @Override
-    protected PainlessExecuteAction.Response createBlankInstance() {
-        return new PainlessExecuteAction.Response();
+    protected Writeable.Reader<PainlessExecuteAction.Response> instanceReader() {
+        return PainlessExecuteAction.Response::new;
     }
 
     @Override
     protected PainlessExecuteAction.Response createTestInstance() {
-        return new PainlessExecuteAction.Response(randomAlphaOfLength(10));
+        Object result;
+        switch (randomIntBetween(0, 2)) {
+            case 0:
+                result = randomAlphaOfLength(10);
+                break;
+            case 1:
+                result = randomBoolean();
+                break;
+            case 2:
+                result = randomDoubleBetween(-10, 10, true);
+                break;
+            default:
+                throw new IllegalStateException("invalid branch");
+        }
+        return new PainlessExecuteAction.Response(result);
+    }
+
+    @Override
+    protected PainlessExecuteAction.Response doParseInstance(XContentParser parser) throws IOException {
+        parser.nextToken(); // START-OBJECT
+        parser.nextToken(); // FIELD-NAME
+        XContentParser.Token token = parser.nextToken(); // result value
+        Object result;
+        switch (token) {
+            case VALUE_STRING:
+                result = parser.text();
+                break;
+            case VALUE_BOOLEAN:
+                result = parser.booleanValue();
+                break;
+            case VALUE_NUMBER:
+                result = parser.doubleValue();
+                break;
+            default:
+                throw new IOException("invalid response");
+        }
+        return new PainlessExecuteAction.Response(result);
     }
 }

server/src/main/java/org/elasticsearch/action/admin/indices/analyze/AnalyzeResponse.java

@@ -71,7 +71,7 @@ public int hashCode() {
             return Objects.hash(term, startOffset, endOffset, position, positionLength, attributes, type);
         }
 
-        public AnalyzeToken(String term, int position, int startOffset, int endOffset, int positionLength,
+        AnalyzeToken(String term, int position, int startOffset, int endOffset, int positionLength,
                             String type, Map<String, Object> attributes) {
             this.term = term;
             this.position = position;
@@ -82,7 +82,7 @@ public AnalyzeToken(String term, int position, int startOffset, int endOffset, i
             this.attributes = attributes;
         }
 
-        public AnalyzeToken(StreamInput in) throws IOException {
+        AnalyzeToken(StreamInput in) throws IOException {
             term = in.readString();
             startOffset = in.readInt();
             endOffset = in.readInt();
@@ -203,7 +203,6 @@ public void writeTo(StreamOutput out) throws IOException {
     }
 
     private final DetailAnalyzeResponse detail;
-
     private final List<AnalyzeToken> tokens;
 
     public AnalyzeResponse(List<AnalyzeToken> tokens, DetailAnalyzeResponse detail) {

server/src/main/java/org/elasticsearch/action/admin/indices/analyze/TransportAnalyzeAction.java

@@ -96,11 +96,6 @@ public TransportAnalyzeAction(Settings settings, ThreadPool threadPool, ClusterS
         this.environment = environment;
     }
 
-    @Override
-    protected AnalyzeResponse newResponse() {
-        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-    }
-
     @Override
     protected Writeable.Reader<AnalyzeResponse> getResponseReader() {
         return AnalyzeResponse::new;

server/src/main/java/org/elasticsearch/action/admin/indices/mapping/get/GetFieldMappingsAction.java

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

server/src/main/java/org/elasticsearch/action/admin/indices/mapping/get/GetFieldMappingsResponse.java

@@ -92,9 +92,33 @@ public class GetFieldMappingsResponse extends ActionResponse implements ToXConte
         this.mappings = mappings;
     }
 
+
     GetFieldMappingsResponse() {
     }
 
+    GetFieldMappingsResponse(StreamInput in) throws IOException {
+        super(in);
+        int size = in.readVInt();
+        Map<String, Map<String, Map<String, FieldMappingMetaData>>> indexMapBuilder = new HashMap<>(size);
+        for (int i = 0; i < size; i++) {
+            String index = in.readString();
+            int typesSize = in.readVInt();
+            Map<String, Map<String, FieldMappingMetaData>> typeMapBuilder = new HashMap<>(typesSize);
+            for (int j = 0; j < typesSize; j++) {
+                String type = in.readString();
+                int fieldSize = in.readVInt();
+                Map<String, FieldMappingMetaData> fieldMapBuilder = new HashMap<>(fieldSize);
+                for (int k = 0; k < fieldSize; k++) {
+                    fieldMapBuilder.put(in.readString(), new FieldMappingMetaData(in.readString(), in.readBytesReference()));
+                }
+                typeMapBuilder.put(type, unmodifiableMap(fieldMapBuilder));
+            }
+            indexMapBuilder.put(index, unmodifiableMap(typeMapBuilder));
+        }
+        mappings = unmodifiableMap(indexMapBuilder);
+
+    }
+
     /** returns the retrieved field mapping. The return map keys are index, type, field (as specified in the request). */
     public Map<String, Map<String, Map<String, FieldMappingMetaData>>> mappings() {
         return mappings;
@@ -269,25 +293,7 @@ public int hashCode() {
 
     @Override
     public void readFrom(StreamInput in) throws IOException {
-        super.readFrom(in);
-        int size = in.readVInt();
-        Map<String, Map<String, Map<String, FieldMappingMetaData>>> indexMapBuilder = new HashMap<>(size);
-        for (int i = 0; i < size; i++) {
-            String index = in.readString();
-            int typesSize = in.readVInt();
-            Map<String, Map<String, FieldMappingMetaData>> typeMapBuilder = new HashMap<>(typesSize);
-            for (int j = 0; j < typesSize; j++) {
-                String type = in.readString();
-                int fieldSize = in.readVInt();
-                Map<String, FieldMappingMetaData> fieldMapBuilder = new HashMap<>(fieldSize);
-                for (int k = 0; k < fieldSize; k++) {
-                    fieldMapBuilder.put(in.readString(), new FieldMappingMetaData(in.readString(), in.readBytesReference()));
-                }
-                typeMapBuilder.put(type, unmodifiableMap(fieldMapBuilder));
-            }
-            indexMapBuilder.put(index, unmodifiableMap(typeMapBuilder));
-        }
-        mappings = unmodifiableMap(indexMapBuilder);
+        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
     }
 
     @Override

server/src/main/java/org/elasticsearch/action/admin/indices/mapping/get/TransportGetFieldMappingsIndexAction.java

@@ -32,6 +32,7 @@
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.regex.Regex;
 import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.XContentHelper;
@@ -123,8 +124,8 @@ protected GetFieldMappingsResponse shardOperation(final GetFieldMappingsIndexReq
     }
 
     @Override
-    protected GetFieldMappingsResponse newResponse() {
-        return new GetFieldMappingsResponse();
+    protected Writeable.Reader<GetFieldMappingsResponse> getResponseReader() {
+        return GetFieldMappingsResponse::new;
     }
 
     @Override

server/src/main/java/org/elasticsearch/action/explain/ExplainAction.java

@@ -20,6 +20,7 @@
 package org.elasticsearch.action.explain;
 
 import org.elasticsearch.action.Action;
+import org.elasticsearch.common.io.stream.Writeable;
 
 /**
  * Entry point for the explain feature.
@@ -35,6 +36,11 @@ private ExplainAction() {
 
     @Override
     public ExplainResponse newResponse() {
-        return new ExplainResponse();
+        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
+    }
+
+    @Override
+    public Writeable.Reader<ExplainResponse> getResponseReader() {
+        return ExplainResponse::new;
     }
 }

server/src/main/java/org/elasticsearch/action/explain/ExplainResponse.java

@@ -60,6 +60,7 @@ public class ExplainResponse extends ActionResponse implements StatusToXContentO
     private Explanation explanation;
     private GetResult getResult;
 
+    // TODO(talevy): remove dependency on empty constructor from ExplainResponseTests
     ExplainResponse() {
     }
 
@@ -80,6 +81,20 @@ public ExplainResponse(String index, String type, String id, boolean exists, Exp
         this.getResult = getResult;
     }
 
+    public ExplainResponse(StreamInput in) throws IOException {
+        super(in);
+        index = in.readString();
+        type = in.readString();
+        id = in.readString();
+        exists = in.readBoolean();
+        if (in.readBoolean()) {
+            explanation = readExplanation(in);
+        }
+        if (in.readBoolean()) {
+            getResult = GetResult.readGetResult(in);
+        }
+    }
+
     public String getIndex() {
         return index;
     }
@@ -123,17 +138,7 @@ public RestStatus status() {
 
     @Override
     public void readFrom(StreamInput in) throws IOException {
-        super.readFrom(in);
-        index = in.readString();
-        type = in.readString();
-        id = in.readString();
-        exists = in.readBoolean();
-        if (in.readBoolean()) {
-            explanation = readExplanation(in);
-        }
-        if (in.readBoolean()) {
-            getResult = GetResult.readGetResult(in);
-        }
+        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
     }
 
     @Override

server/src/main/java/org/elasticsearch/action/explain/TransportExplainAction.java

@@ -32,6 +32,7 @@
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.lease.Releasables;
 import org.elasticsearch.index.IndexService;
 import org.elasticsearch.index.engine.Engine;
@@ -152,8 +153,8 @@ protected ExplainResponse shardOperation(ExplainRequest request, ShardId shardId
     }
 
     @Override
-    protected ExplainResponse newResponse() {
-        return new ExplainResponse();
+    protected Writeable.Reader<ExplainResponse> getResponseReader() {
+        return ExplainResponse::new;
     }
 
     @Override

server/src/main/java/org/elasticsearch/action/fieldcaps/TransportFieldCapabilitiesIndexAction.java

@@ -28,6 +28,7 @@
 import org.elasticsearch.cluster.routing.ShardsIterator;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.index.mapper.MappedFieldType;
 import org.elasticsearch.index.mapper.MapperService;
 import org.elasticsearch.index.mapper.ObjectMapper;
@@ -114,8 +115,8 @@ protected FieldCapabilitiesIndexResponse shardOperation(final FieldCapabilitiesI
     }
 
     @Override
-    protected FieldCapabilitiesIndexResponse newResponse() {
-        return new FieldCapabilitiesIndexResponse();
+    protected Writeable.Reader<FieldCapabilitiesIndexResponse> getResponseReader() {
+        return FieldCapabilitiesIndexResponse::new;
     }
 
     @Override

server/src/main/java/org/elasticsearch/action/get/GetAction.java

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

server/src/main/java/org/elasticsearch/action/get/GetResponse.java

@@ -48,7 +48,9 @@ public class GetResponse extends ActionResponse implements Iterable<DocumentFiel
 
     GetResult getResult;
 
-    GetResponse() {
+    GetResponse(StreamInput in) throws IOException {
+        super(in);
+        getResult = GetResult.readGetResult(in);
     }
 
     public GetResponse(GetResult getResult) {
@@ -203,8 +205,7 @@ public static GetResponse fromXContent(XContentParser parser) throws IOException
 
     @Override
     public void readFrom(StreamInput in) throws IOException {
-        super.readFrom(in);
-        getResult = GetResult.readGetResult(in);
+        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
     }
 
     @Override