Add StreamableResponseAction to aid in deprecation of Streamable (#43770)

The Action base class currently works for both Streamable and Writeable
response types. This commit intorduces StreamableResponseAction, for
which only the legacy Action implementions which provide newResponse()
will extend. This eliminates the need for overriding newResponse() with
an UnsupportedOperationException.

relates #34389

client/client-benchmark-noop-api-plugin/src/main/java/org/elasticsearch/plugin/noop/action/bulk/NoopBulkAction.java

@@ -18,10 +18,10 @@
  */
 package org.elasticsearch.plugin.noop.action.bulk;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 import org.elasticsearch.action.bulk.BulkResponse;
 
-public class NoopBulkAction extends Action<BulkResponse> {
+public class NoopBulkAction extends StreamableResponseAction<BulkResponse> {
     public static final String NAME = "mock:data/write/bulk";
 
     public static final NoopBulkAction INSTANCE = new NoopBulkAction();

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

@@ -30,11 +30,6 @@ public class NoopSearchAction extends Action<SearchResponse> {
         super(NAME);
     }
 
-    @Override
-    public SearchResponse newResponse() {
-        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-    }
-
     @Override
     public Writeable.Reader<SearchResponse> getResponseReader() {
         return SearchResponse::new;

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/GrokProcessorGetAction.java

@@ -18,11 +18,11 @@
  */
 package org.elasticsearch.ingest.common;
 
-import org.elasticsearch.action.Action;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionRequest;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.ActionResponse;
+import org.elasticsearch.action.StreamableResponseAction;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.client.node.NodeClient;
@@ -45,7 +45,7 @@ import java.util.Map;
 import static org.elasticsearch.ingest.common.IngestCommonPlugin.GROK_PATTERNS;
 import static org.elasticsearch.rest.RestRequest.Method.GET;
 
-public class GrokProcessorGetAction extends Action<GrokProcessorGetAction.Response> {
+public class GrokProcessorGetAction extends StreamableResponseAction<GrokProcessorGetAction.Response> {
 
     static final GrokProcessorGetAction INSTANCE = new GrokProcessorGetAction();
     static final String NAME = "cluster:admin/ingest/processor/grok/get";

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

@@ -31,11 +31,6 @@ public class MultiSearchTemplateAction extends Action<MultiSearchTemplateRespons
         super(NAME);
     }
 
-    @Override
-    public MultiSearchTemplateResponse newResponse() {
-        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-    }
-
     @Override
     public Writeable.Reader<MultiSearchTemplateResponse> getResponseReader() {
         return MultiSearchTemplateResponse::new;

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

@@ -31,11 +31,6 @@ public class SearchTemplateAction extends Action<SearchTemplateResponse> {
         super(NAME);
     }
 
-    @Override
-    public SearchTemplateResponse newResponse() {
-        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-    }
-
     @Override
     public Writeable.Reader<SearchTemplateResponse> getResponseReader() {
         return SearchTemplateResponse::new;

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

@@ -75,11 +75,6 @@ public class PainlessContextAction extends Action<PainlessContextAction.Response
         super(NAME);
     }
 
-    @Override
-    public Response newResponse() {
-        throw new UnsupportedOperationException();
-    }
-
     @Override
     public Writeable.Reader<Response> getResponseReader() {
         return Response::new;

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

@@ -99,8 +99,8 @@ public class PainlessExecuteAction extends Action<PainlessExecuteAction.Response
     }
 
     @Override
-    public Response newResponse() {
+    public Writeable.Reader<Response> getResponseReader() {
-        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
+        return Response::new;
     }
 
     public static class Request extends SingleShardRequest<Request> implements ToXContentObject {

modules/rank-eval/src/main/java/org/elasticsearch/index/rankeval/RankEvalAction.java

@@ -19,12 +19,12 @@
 
 package org.elasticsearch.index.rankeval;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
 /**
  * Action for explaining evaluating search ranking results.
  */
-public class RankEvalAction extends Action<RankEvalResponse> {
+public class RankEvalAction extends StreamableResponseAction<RankEvalResponse> {
 
     public static final RankEvalAction INSTANCE = new RankEvalAction();
     public static final String NAME = "indices:data/read/rank_eval";

modules/reindex/src/main/java/org/elasticsearch/index/reindex/RethrottleAction.java

@@ -31,11 +31,6 @@ public class RethrottleAction extends Action<ListTasksResponse> {
         super(NAME);
     }
 
-    @Override
-    public ListTasksResponse newResponse() {
-        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-    }
-
     @Override
     public Writeable.Reader<ListTasksResponse> getResponseReader() {
         return ListTasksResponse::new;

server/src/main/java/org/elasticsearch/action/Action.java

@@ -26,15 +26,27 @@ import org.elasticsearch.transport.TransportRequestOptions;
 /**
  * A generic action. Should strive to make it a singleton.
  */
-public abstract class Action<Response extends ActionResponse> {
+public class Action<Response extends ActionResponse> {
 
     private final String name;
+    private final Writeable.Reader<Response> responseReader;
 
     /**
      * @param name The name of the action, must be unique across actions.
+     * @deprecated Pass a {@link Writeable.Reader} with {@link }
      */
+    @Deprecated
     protected Action(String name) {
+        this(name, null);
+    }
+
+    /**
+     * @param name The name of the action, must be unique across actions.
+     * @param responseReader A reader for the response type
+     */
+    public Action(String name, Writeable.Reader<Response> responseReader) {
         this.name = name;
+        this.responseReader = responseReader;
     }
 
     /**
@@ -44,23 +56,11 @@ public abstract class Action<Response extends ActionResponse> {
         return this.name;
     }
 
-    /**
-     * Creates a new response instance.
-     * @deprecated Implement {@link #getResponseReader()} instead and make this method throw an
-     *             {@link UnsupportedOperationException}
-     */
-    @Deprecated
-    public abstract Response newResponse();
-
     /**
      * Get a reader that can create a new instance of the class from a {@link org.elasticsearch.common.io.stream.StreamInput}
      */
     public Writeable.Reader<Response> getResponseReader() {
-        return in -> {
+        return responseReader;
-            Response response = newResponse();
-            response.readFrom(in);
-            return response;
-        };
     }
 
     /**

server/src/main/java/org/elasticsearch/action/StreamableResponseAction.java

@@ -19,29 +19,33 @@
 
 package org.elasticsearch.action;
 
-
 import org.elasticsearch.common.io.stream.Writeable;
 
 /**
- * An action for which the response class implements {@link org.elasticsearch.common.io.stream.Writeable}.
+ * An action for with the response type implements {@link org.elasticsearch.common.io.stream.Streamable}.
+ * @deprecated Use {@link Action} directly and provide a {@link Writeable.Reader}
  */
-public class Action2<Response extends ActionResponse> extends Action<Response> {
+@Deprecated
-    private final Writeable.Reader<Response> responseReader;
+public abstract class StreamableResponseAction<Response extends ActionResponse> extends Action<Response> {
 
-    public Action2(String name, Writeable.Reader<Response> responseReader) {
+    protected StreamableResponseAction(String name) {
         super(name);
-        this.responseReader = responseReader;
-    }
-
-    @Override
-    public Response newResponse() {
-        throw new UnsupportedOperationException();
     }
 
     /**
-     * Get a reader that can create a new instance of the class from a {@link org.elasticsearch.common.io.stream.StreamInput}
+     * Creates a new response instance.
+     * @deprecated Implement {@link #getResponseReader()} instead and make this method throw an
+     *             {@link UnsupportedOperationException}
      */
-    public Writeable.Reader<Response> getResponseReader() {
+    @Deprecated
-        return responseReader;
+    public abstract Response newResponse();
+
+    @Override
+    public final Writeable.Reader<Response> getResponseReader() {
+        return in -> {
+            Response response = newResponse();
+            response.readFrom(in);
+            return response;
+        };
     }
 }

server/src/main/java/org/elasticsearch/action/admin/cluster/allocation/ClusterAllocationExplainAction.java

@@ -19,12 +19,12 @@
 
 package org.elasticsearch.action.admin.cluster.allocation;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
 /**
  * Action for explaining shard allocation for a shard in the cluster
  */
-public class ClusterAllocationExplainAction extends Action<ClusterAllocationExplainResponse> {
+public class ClusterAllocationExplainAction extends StreamableResponseAction<ClusterAllocationExplainResponse> {
 
     public static final ClusterAllocationExplainAction INSTANCE = new ClusterAllocationExplainAction();
     public static final String NAME = "cluster:monitor/allocation/explain";

server/src/main/java/org/elasticsearch/action/admin/cluster/configuration/AddVotingConfigExclusionsAction.java

@@ -29,11 +29,6 @@ public class AddVotingConfigExclusionsAction extends Action<AddVotingConfigExclu
         super(NAME);
     }
 
-    @Override
-    public AddVotingConfigExclusionsResponse newResponse() {
-        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-    }
-
     @Override
     public Reader<AddVotingConfigExclusionsResponse> getResponseReader() {
         return AddVotingConfigExclusionsResponse::new;

server/src/main/java/org/elasticsearch/action/admin/cluster/configuration/ClearVotingConfigExclusionsAction.java

@@ -29,11 +29,6 @@ public class ClearVotingConfigExclusionsAction extends Action<ClearVotingConfigE
         super(NAME);
     }
 
-    @Override
-    public ClearVotingConfigExclusionsResponse newResponse() {
-        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-    }
-
     @Override
     public Reader<ClearVotingConfigExclusionsResponse> getResponseReader() {
         return ClearVotingConfigExclusionsResponse::new;

server/src/main/java/org/elasticsearch/action/admin/cluster/health/ClusterHealthAction.java

@@ -19,9 +19,9 @@
 
 package org.elasticsearch.action.admin.cluster.health;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
-public class ClusterHealthAction extends Action<ClusterHealthResponse> {
+public class ClusterHealthAction extends StreamableResponseAction<ClusterHealthResponse> {
 
     public static final ClusterHealthAction INSTANCE = new ClusterHealthAction();
     public static final String NAME = "cluster:monitor/health";

server/src/main/java/org/elasticsearch/action/admin/cluster/node/hotthreads/NodesHotThreadsAction.java

@@ -19,9 +19,9 @@
 
 package org.elasticsearch.action.admin.cluster.node.hotthreads;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
-public class NodesHotThreadsAction extends Action<NodesHotThreadsResponse> {
+public class NodesHotThreadsAction extends StreamableResponseAction<NodesHotThreadsResponse> {
 
     public static final NodesHotThreadsAction INSTANCE = new NodesHotThreadsAction();
     public static final String NAME = "cluster:monitor/nodes/hot_threads";

server/src/main/java/org/elasticsearch/action/admin/cluster/node/info/NodesInfoAction.java

@@ -19,9 +19,9 @@
 
 package org.elasticsearch.action.admin.cluster.node.info;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
-public class NodesInfoAction extends Action<NodesInfoResponse> {
+public class NodesInfoAction extends StreamableResponseAction<NodesInfoResponse> {
 
     public static final NodesInfoAction INSTANCE = new NodesInfoAction();
     public static final String NAME = "cluster:monitor/nodes/info";

server/src/main/java/org/elasticsearch/action/admin/cluster/node/reload/NodesReloadSecureSettingsAction.java

@@ -19,10 +19,10 @@
 
 package org.elasticsearch.action.admin.cluster.node.reload;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
 public class NodesReloadSecureSettingsAction
-        extends Action<NodesReloadSecureSettingsResponse> {
+        extends StreamableResponseAction<NodesReloadSecureSettingsResponse> {
 
     public static final NodesReloadSecureSettingsAction INSTANCE = new NodesReloadSecureSettingsAction();
     public static final String NAME = "cluster:admin/nodes/reload_secure_settings";

server/src/main/java/org/elasticsearch/action/admin/cluster/node/stats/NodesStatsAction.java

@@ -19,9 +19,9 @@
 
 package org.elasticsearch.action.admin.cluster.node.stats;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
-public class NodesStatsAction extends Action<NodesStatsResponse> {
+public class NodesStatsAction extends StreamableResponseAction<NodesStatsResponse> {
 
     public static final NodesStatsAction INSTANCE = new NodesStatsAction();
     public static final String NAME = "cluster:monitor/nodes/stats";

server/src/main/java/org/elasticsearch/action/admin/cluster/node/tasks/cancel/CancelTasksAction.java

@@ -34,11 +34,6 @@ public class CancelTasksAction extends Action<CancelTasksResponse> {
         super(NAME);
     }
 
-    @Override
-    public CancelTasksResponse newResponse() {
-        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-    }
-
     @Override
     public Writeable.Reader<CancelTasksResponse> getResponseReader() {
         return CancelTasksResponse::new;

server/src/main/java/org/elasticsearch/action/admin/cluster/node/tasks/get/GetTaskAction.java

@@ -19,12 +19,12 @@
 
 package org.elasticsearch.action.admin.cluster.node.tasks.get;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
 /**
  * Action for retrieving a list of currently running tasks
  */
-public class GetTaskAction extends Action<GetTaskResponse> {
+public class GetTaskAction extends StreamableResponseAction<GetTaskResponse> {
     public static final String TASKS_ORIGIN = "tasks";
 
     public static final GetTaskAction INSTANCE = new GetTaskAction();

server/src/main/java/org/elasticsearch/action/admin/cluster/node/tasks/list/ListTasksAction.java

@@ -34,11 +34,6 @@ public class ListTasksAction extends Action<ListTasksResponse> {
         super(NAME);
     }
 
-    @Override
-    public ListTasksResponse newResponse() {
-        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-    }
-
     @Override
     public Writeable.Reader<ListTasksResponse> getResponseReader() {
         return ListTasksResponse::new;

server/src/main/java/org/elasticsearch/action/admin/cluster/node/usage/NodesUsageAction.java

@@ -19,9 +19,9 @@
 
 package org.elasticsearch.action.admin.cluster.node.usage;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
-public class NodesUsageAction extends Action<NodesUsageResponse> {
+public class NodesUsageAction extends StreamableResponseAction<NodesUsageResponse> {
 
     public static final NodesUsageAction INSTANCE = new NodesUsageAction();
     public static final String NAME = "cluster:monitor/nodes/usage";

server/src/main/java/org/elasticsearch/action/admin/cluster/remote/RemoteInfoAction.java

@@ -19,9 +19,9 @@
 
 package org.elasticsearch.action.admin.cluster.remote;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
-public final class RemoteInfoAction extends Action<RemoteInfoResponse> {
+public final class RemoteInfoAction extends StreamableResponseAction<RemoteInfoResponse> {
 
     public static final String NAME = "cluster:monitor/remote/info";
     public static final RemoteInfoAction INSTANCE = new RemoteInfoAction();

server/src/main/java/org/elasticsearch/action/admin/cluster/repositories/delete/DeleteRepositoryAction.java

@@ -35,11 +35,6 @@ public class DeleteRepositoryAction extends Action<AcknowledgedResponse> {
         super(NAME);
     }
 
-    @Override
-    public AcknowledgedResponse newResponse() {
-        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-    }
-
     @Override
     public Writeable.Reader<AcknowledgedResponse> getResponseReader() {
         return AcknowledgedResponse::new;

server/src/main/java/org/elasticsearch/action/admin/cluster/repositories/get/GetRepositoriesAction.java

@@ -34,11 +34,6 @@ public class GetRepositoriesAction extends Action<GetRepositoriesResponse> {
         super(NAME);
     }
 
-    @Override
-    public GetRepositoriesResponse newResponse() {
-        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-    }
-
     @Override
     public Writeable.Reader<GetRepositoriesResponse> getResponseReader() {
         return GetRepositoriesResponse::new;

server/src/main/java/org/elasticsearch/action/admin/cluster/repositories/put/PutRepositoryAction.java

@@ -35,11 +35,6 @@ public class PutRepositoryAction extends Action<AcknowledgedResponse> {
         super(NAME);
     }
 
-    @Override
-    public AcknowledgedResponse newResponse() {
-        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-    }
-
     @Override
     public Writeable.Reader<AcknowledgedResponse> getResponseReader() {
         return AcknowledgedResponse::new;

server/src/main/java/org/elasticsearch/action/admin/cluster/repositories/verify/VerifyRepositoryAction.java

@@ -19,12 +19,12 @@
 
 package org.elasticsearch.action.admin.cluster.repositories.verify;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
 /**
  * Unregister repository action
  */
-public class VerifyRepositoryAction extends Action<VerifyRepositoryResponse> {
+public class VerifyRepositoryAction extends StreamableResponseAction<VerifyRepositoryResponse> {
 
     public static final VerifyRepositoryAction INSTANCE = new VerifyRepositoryAction();
     public static final String NAME = "cluster:admin/repository/verify";

server/src/main/java/org/elasticsearch/action/admin/cluster/reroute/ClusterRerouteAction.java

@@ -31,11 +31,6 @@ public class ClusterRerouteAction extends Action<ClusterRerouteResponse> {
         super(NAME);
     }
 
-    @Override
-    public ClusterRerouteResponse newResponse() {
-        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-    }
-
     @Override
     public Writeable.Reader<ClusterRerouteResponse> getResponseReader() {
         return ClusterRerouteResponse::new;

server/src/main/java/org/elasticsearch/action/admin/cluster/settings/ClusterUpdateSettingsAction.java

@@ -31,11 +31,6 @@ public class ClusterUpdateSettingsAction extends Action<ClusterUpdateSettingsRes
         super(NAME);
     }
 
-    @Override
-    public ClusterUpdateSettingsResponse newResponse() {
-        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-    }
-
     @Override
     public Writeable.Reader<ClusterUpdateSettingsResponse> getResponseReader() {
         return ClusterUpdateSettingsResponse::new;

server/src/main/java/org/elasticsearch/action/admin/cluster/shards/ClusterSearchShardsAction.java

@@ -31,11 +31,6 @@ public class ClusterSearchShardsAction extends Action<ClusterSearchShardsRespons
         super(NAME);
     }
 
-    @Override
-    public ClusterSearchShardsResponse newResponse() {
-        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-    }
-
     @Override
     public Writeable.Reader<ClusterSearchShardsResponse> getResponseReader() {
         return ClusterSearchShardsResponse::new;

server/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/create/CreateSnapshotAction.java

@@ -19,12 +19,12 @@
 
 package org.elasticsearch.action.admin.cluster.snapshots.create;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
 /**
  * Create snapshot action
  */
-public class CreateSnapshotAction extends Action<CreateSnapshotResponse> {
+public class CreateSnapshotAction extends StreamableResponseAction<CreateSnapshotResponse> {
 
     public static final CreateSnapshotAction INSTANCE = new CreateSnapshotAction();
     public static final String NAME = "cluster:admin/snapshot/create";

server/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/delete/DeleteSnapshotAction.java

@@ -35,11 +35,6 @@ public class DeleteSnapshotAction extends Action<AcknowledgedResponse> {
         super(NAME);
     }
 
-    @Override
-    public AcknowledgedResponse newResponse() {
-        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-    }
-
     @Override
     public Writeable.Reader<AcknowledgedResponse> getResponseReader() {
         return AcknowledgedResponse::new;

server/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/get/GetSnapshotsAction.java

@@ -34,11 +34,6 @@ public class GetSnapshotsAction extends Action<GetSnapshotsResponse> {
         super(NAME);
     }
 
-    @Override
-    public GetSnapshotsResponse newResponse() {
-        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-    }
-
     @Override
     public Writeable.Reader<GetSnapshotsResponse> getResponseReader() {
         return GetSnapshotsResponse::new;

server/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/restore/RestoreSnapshotAction.java

@@ -19,12 +19,12 @@
 
 package org.elasticsearch.action.admin.cluster.snapshots.restore;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
 /**
  * Restore snapshot action
  */
-public class RestoreSnapshotAction extends Action<RestoreSnapshotResponse> {
+public class RestoreSnapshotAction extends StreamableResponseAction<RestoreSnapshotResponse> {
 
     public static final RestoreSnapshotAction INSTANCE = new RestoreSnapshotAction();
     public static final String NAME = "cluster:admin/snapshot/restore";

server/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/status/SnapshotsStatusAction.java

@@ -19,12 +19,12 @@
 
 package org.elasticsearch.action.admin.cluster.snapshots.status;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
 /**
  * Snapshots status action
  */
-public class SnapshotsStatusAction extends Action<SnapshotsStatusResponse> {
+public class SnapshotsStatusAction extends StreamableResponseAction<SnapshotsStatusResponse> {
 
     public static final SnapshotsStatusAction INSTANCE = new SnapshotsStatusAction();
     public static final String NAME = "cluster:admin/snapshot/status";

server/src/main/java/org/elasticsearch/action/admin/cluster/state/ClusterStateAction.java

@@ -19,9 +19,9 @@
 
 package org.elasticsearch.action.admin.cluster.state;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
-public class ClusterStateAction extends Action<ClusterStateResponse> {
+public class ClusterStateAction extends StreamableResponseAction<ClusterStateResponse> {
 
     public static final ClusterStateAction INSTANCE = new ClusterStateAction();
     public static final String NAME = "cluster:monitor/state";

server/src/main/java/org/elasticsearch/action/admin/cluster/stats/ClusterStatsAction.java

@@ -19,9 +19,9 @@
 
 package org.elasticsearch.action.admin.cluster.stats;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
-public class ClusterStatsAction extends Action<ClusterStatsResponse> {
+public class ClusterStatsAction extends StreamableResponseAction<ClusterStatsResponse> {
 
     public static final ClusterStatsAction INSTANCE = new ClusterStatsAction();
     public static final String NAME = "cluster:monitor/stats";

server/src/main/java/org/elasticsearch/action/admin/cluster/storedscripts/DeleteStoredScriptAction.java

@@ -32,11 +32,6 @@ public class DeleteStoredScriptAction extends Action<AcknowledgedResponse> {
         super(NAME);
     }
 
-    @Override
-    public AcknowledgedResponse newResponse() {
-        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-    }
-
     @Override
     public Writeable.Reader<AcknowledgedResponse> getResponseReader() {
         return AcknowledgedResponse::new;

server/src/main/java/org/elasticsearch/action/admin/cluster/storedscripts/GetStoredScriptAction.java

@@ -19,9 +19,9 @@
 
 package org.elasticsearch.action.admin.cluster.storedscripts;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
-public class GetStoredScriptAction extends Action<GetStoredScriptResponse> {
+public class GetStoredScriptAction extends StreamableResponseAction<GetStoredScriptResponse> {
 
     public static final GetStoredScriptAction INSTANCE = new GetStoredScriptAction();
     public static final String NAME = "cluster:admin/script/get";

server/src/main/java/org/elasticsearch/action/admin/cluster/storedscripts/PutStoredScriptAction.java

@@ -33,11 +33,6 @@ public class PutStoredScriptAction extends Action<AcknowledgedResponse> {
         super(NAME);
     }
 
-    @Override
-    public AcknowledgedResponse newResponse() {
-        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-    }
-
     @Override
     public Writeable.Reader<AcknowledgedResponse> getResponseReader() {
         return AcknowledgedResponse::new;

server/src/main/java/org/elasticsearch/action/admin/cluster/tasks/PendingClusterTasksAction.java

@@ -19,9 +19,9 @@
 
 package org.elasticsearch.action.admin.cluster.tasks;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
-public class PendingClusterTasksAction extends Action<PendingClusterTasksResponse> {
+public class PendingClusterTasksAction extends StreamableResponseAction<PendingClusterTasksResponse> {
 
     public static final PendingClusterTasksAction INSTANCE = new PendingClusterTasksAction();
     public static final String NAME = "cluster:monitor/task";

server/src/main/java/org/elasticsearch/action/admin/indices/alias/IndicesAliasesAction.java

@@ -32,11 +32,6 @@ public class IndicesAliasesAction extends Action<AcknowledgedResponse> {
         super(NAME);
     }
 
-    @Override
-    public AcknowledgedResponse newResponse() {
-        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-    }
-
     @Override
     public Writeable.Reader<AcknowledgedResponse> getResponseReader() {
         return AcknowledgedResponse::new;

server/src/main/java/org/elasticsearch/action/admin/indices/alias/get/GetAliasesAction.java

@@ -19,9 +19,9 @@
 
 package org.elasticsearch.action.admin.indices.alias.get;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
-public class GetAliasesAction extends Action<GetAliasesResponse> {
+public class GetAliasesAction extends StreamableResponseAction<GetAliasesResponse> {
 
     public static final GetAliasesAction INSTANCE = new GetAliasesAction();
     public static final String NAME = "indices:admin/aliases/get";

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

@@ -59,11 +59,6 @@ public class AnalyzeAction extends Action<AnalyzeAction.Response> {
         return Response::new;
     }
 
-    @Override
-    public Response newResponse() {
-        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-    }
-
     /**
      * A request to analyze a text associated with a specific index. Allow to provide
      * the actual analyzer name to perform the analysis with.

server/src/main/java/org/elasticsearch/action/admin/indices/cache/clear/ClearIndicesCacheAction.java

@@ -19,9 +19,9 @@
 
 package org.elasticsearch.action.admin.indices.cache.clear;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
-public class ClearIndicesCacheAction extends Action<ClearIndicesCacheResponse> {
+public class ClearIndicesCacheAction extends StreamableResponseAction<ClearIndicesCacheResponse> {
 
     public static final ClearIndicesCacheAction INSTANCE = new ClearIndicesCacheAction();
     public static final String NAME = "indices:admin/cache/clear";

server/src/main/java/org/elasticsearch/action/admin/indices/close/CloseIndexAction.java

@@ -31,11 +31,6 @@ public class CloseIndexAction extends Action<CloseIndexResponse> {
         super(NAME);
     }
 
-    @Override
-    public CloseIndexResponse newResponse() {
-        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-    }
-
     @Override
     public Writeable.Reader<CloseIndexResponse> getResponseReader() {
         return CloseIndexResponse::new;

server/src/main/java/org/elasticsearch/action/admin/indices/create/CreateIndexAction.java

@@ -31,11 +31,6 @@ public class CreateIndexAction extends Action<CreateIndexResponse> {
         super(NAME);
     }
 
-    @Override
-    public CreateIndexResponse newResponse() {
-        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-    }
-
     @Override
     public Writeable.Reader<CreateIndexResponse> getResponseReader() {
         return CreateIndexResponse::new;

server/src/main/java/org/elasticsearch/action/admin/indices/delete/DeleteIndexAction.java

@@ -32,11 +32,6 @@ public class DeleteIndexAction extends Action<AcknowledgedResponse> {
         super(NAME);
     }
 
-    @Override
-    public AcknowledgedResponse newResponse() {
-        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-    }
-
     @Override
     public Writeable.Reader<AcknowledgedResponse> getResponseReader() {
         return AcknowledgedResponse::new;

server/src/main/java/org/elasticsearch/action/admin/indices/flush/FlushAction.java

@@ -19,9 +19,9 @@
 
 package org.elasticsearch.action.admin.indices.flush;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
-public class FlushAction extends Action<FlushResponse> {
+public class FlushAction extends StreamableResponseAction<FlushResponse> {
 
     public static final FlushAction INSTANCE = new FlushAction();
     public static final String NAME = "indices:admin/flush";

server/src/main/java/org/elasticsearch/action/admin/indices/flush/SyncedFlushAction.java

@@ -19,10 +19,10 @@
 
 package org.elasticsearch.action.admin.indices.flush;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
 
-public class SyncedFlushAction extends Action<SyncedFlushResponse> {
+public class SyncedFlushAction extends StreamableResponseAction<SyncedFlushResponse> {
 
     public static final SyncedFlushAction INSTANCE = new SyncedFlushAction();
     public static final String NAME = "indices:admin/synced_flush";

server/src/main/java/org/elasticsearch/action/admin/indices/forcemerge/ForceMergeAction.java

@@ -19,9 +19,9 @@
 
 package org.elasticsearch.action.admin.indices.forcemerge;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
-public class ForceMergeAction extends Action<ForceMergeResponse> {
+public class ForceMergeAction extends StreamableResponseAction<ForceMergeResponse> {
 
     public static final ForceMergeAction INSTANCE = new ForceMergeAction();
     public static final String NAME = "indices:admin/forcemerge";

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

@@ -19,9 +19,9 @@
 
 package org.elasticsearch.action.admin.indices.get;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
-public class GetIndexAction extends Action<GetIndexResponse> {
+public class GetIndexAction extends StreamableResponseAction<GetIndexResponse> {
 
     public static final GetIndexAction INSTANCE = new GetIndexAction();
     public static final String NAME = "indices:admin/get";

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

@@ -31,11 +31,6 @@ public class GetFieldMappingsAction extends Action<GetFieldMappingsResponse> {
         super(NAME);
     }
 
-    @Override
-    public GetFieldMappingsResponse newResponse() {
-        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/GetMappingsAction.java

@@ -19,9 +19,9 @@
 
 package org.elasticsearch.action.admin.indices.mapping.get;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
-public class GetMappingsAction extends Action<GetMappingsResponse> {
+public class GetMappingsAction extends StreamableResponseAction<GetMappingsResponse> {
 
     public static final GetMappingsAction INSTANCE = new GetMappingsAction();
     public static final String NAME = "indices:admin/mappings/get";

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

@@ -22,7 +22,6 @@ package org.elasticsearch.action.admin.indices.mapping.get;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.Version;
 import org.elasticsearch.action.Action;
-import org.elasticsearch.action.Action2;
 import org.elasticsearch.action.admin.indices.mapping.get.GetFieldMappingsResponse.FieldMappingMetaData;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.single.shard.TransportSingleShardAction;
@@ -65,7 +64,7 @@ public class TransportGetFieldMappingsIndexAction
         extends TransportSingleShardAction<GetFieldMappingsIndexRequest, GetFieldMappingsResponse> {
 
     private static final String ACTION_NAME = GetFieldMappingsAction.NAME + "[index]";
-    public static final Action<GetFieldMappingsResponse> ACTION_INSTANCE = new Action2<>(ACTION_NAME, GetFieldMappingsResponse::new);
+    public static final Action<GetFieldMappingsResponse> ACTION_INSTANCE = new Action<>(ACTION_NAME, GetFieldMappingsResponse::new);
 
     protected final ClusterService clusterService;
     private final IndicesService indicesService;

server/src/main/java/org/elasticsearch/action/admin/indices/mapping/put/PutMappingAction.java

@@ -32,11 +32,6 @@ public class PutMappingAction extends Action<AcknowledgedResponse> {
         super(NAME);
     }
 
-    @Override
-    public AcknowledgedResponse newResponse() {
-        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-    }
-
     @Override
     public Writeable.Reader<AcknowledgedResponse> getResponseReader() {
         return AcknowledgedResponse::new;

server/src/main/java/org/elasticsearch/action/admin/indices/open/OpenIndexAction.java

@@ -31,11 +31,6 @@ public class OpenIndexAction extends Action<OpenIndexResponse> {
         super(NAME);
     }
 
-    @Override
-    public OpenIndexResponse newResponse() {
-        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-    }
-
     @Override
     public Writeable.Reader<OpenIndexResponse> getResponseReader() {
         return OpenIndexResponse::new;

server/src/main/java/org/elasticsearch/action/admin/indices/recovery/RecoveryAction.java

@@ -19,12 +19,12 @@
 
 package org.elasticsearch.action.admin.indices.recovery;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
 /**
  * Recovery information action
  */
-public class RecoveryAction extends Action<RecoveryResponse> {
+public class RecoveryAction extends StreamableResponseAction<RecoveryResponse> {
 
     public static final RecoveryAction INSTANCE = new RecoveryAction();
     public static final String NAME = "indices:monitor/recovery";

server/src/main/java/org/elasticsearch/action/admin/indices/refresh/RefreshAction.java

@@ -19,9 +19,9 @@
 
 package org.elasticsearch.action.admin.indices.refresh;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
-public class RefreshAction extends Action<RefreshResponse> {
+public class RefreshAction extends StreamableResponseAction<RefreshResponse> {
 
     public static final RefreshAction INSTANCE = new RefreshAction();
     public static final String NAME = "indices:admin/refresh";

server/src/main/java/org/elasticsearch/action/admin/indices/rollover/RolloverAction.java

@@ -31,11 +31,6 @@ public class RolloverAction extends Action<RolloverResponse> {
         super(NAME);
     }
 
-    @Override
-    public RolloverResponse newResponse() {
-        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-    }
-
     @Override
     public Writeable.Reader<RolloverResponse> getResponseReader() {
         return RolloverResponse::new;

server/src/main/java/org/elasticsearch/action/admin/indices/segments/IndicesSegmentsAction.java

@@ -19,9 +19,9 @@
 
 package org.elasticsearch.action.admin.indices.segments;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
-public class IndicesSegmentsAction extends Action<IndicesSegmentResponse> {
+public class IndicesSegmentsAction extends StreamableResponseAction<IndicesSegmentResponse> {
 
     public static final IndicesSegmentsAction INSTANCE = new IndicesSegmentsAction();
     public static final String NAME = "indices:monitor/segments";

server/src/main/java/org/elasticsearch/action/admin/indices/settings/get/GetSettingsAction.java

@@ -19,9 +19,9 @@
 
 package org.elasticsearch.action.admin.indices.settings.get;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
-public class GetSettingsAction extends Action<GetSettingsResponse> {
+public class GetSettingsAction extends StreamableResponseAction<GetSettingsResponse> {
 
     public static final GetSettingsAction INSTANCE = new GetSettingsAction();
     public static final String NAME = "indices:monitor/settings/get";

server/src/main/java/org/elasticsearch/action/admin/indices/settings/put/UpdateSettingsAction.java

@@ -32,11 +32,6 @@ public class UpdateSettingsAction extends Action<AcknowledgedResponse> {
         super(NAME);
     }
 
-    @Override
-    public AcknowledgedResponse newResponse() {
-        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-    }
-
     @Override
     public Writeable.Reader<AcknowledgedResponse> getResponseReader() {
         return AcknowledgedResponse::new;

server/src/main/java/org/elasticsearch/action/admin/indices/shards/IndicesShardStoresAction.java

@@ -19,7 +19,7 @@
 
 package org.elasticsearch.action.admin.indices.shards;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
 /**
  * Action for {@link TransportIndicesShardStoresAction}
@@ -28,7 +28,7 @@ import org.elasticsearch.action.Action;
  * Shard store information reports which nodes hold shard copies, how recent they are
  * and any exceptions on opening the shard index or from previous engine failures
  */
-public class IndicesShardStoresAction extends Action<IndicesShardStoresResponse> {
+public class IndicesShardStoresAction extends StreamableResponseAction<IndicesShardStoresResponse> {
 
     public static final IndicesShardStoresAction INSTANCE = new IndicesShardStoresAction();
     public static final String NAME = "indices:monitor/shard_stores";

server/src/main/java/org/elasticsearch/action/admin/indices/shrink/ResizeAction.java

@@ -31,11 +31,6 @@ public class ResizeAction extends Action<ResizeResponse> {
         super(NAME);
     }
 
-    @Override
-    public ResizeResponse newResponse() {
-        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-    }
-
     @Override
     public Writeable.Reader<ResizeResponse> getResponseReader() {
         return ResizeResponse::new;

server/src/main/java/org/elasticsearch/action/admin/indices/shrink/ShrinkAction.java

@@ -31,11 +31,6 @@ public class ShrinkAction extends Action<ResizeResponse> {
         super(NAME);
     }
 
-    @Override
-    public ResizeResponse newResponse() {
-        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-    }
-
     @Override
     public Writeable.Reader<ResizeResponse> getResponseReader() {
         return ResizeResponse::new;

server/src/main/java/org/elasticsearch/action/admin/indices/stats/IndicesStatsAction.java

@@ -19,9 +19,9 @@
 
 package org.elasticsearch.action.admin.indices.stats;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
-public class IndicesStatsAction extends Action<IndicesStatsResponse> {
+public class IndicesStatsAction extends StreamableResponseAction<IndicesStatsResponse> {
 
     public static final IndicesStatsAction INSTANCE = new IndicesStatsAction();
     public static final String NAME = "indices:monitor/stats";

server/src/main/java/org/elasticsearch/action/admin/indices/template/delete/DeleteIndexTemplateAction.java

@@ -32,11 +32,6 @@ public class DeleteIndexTemplateAction extends Action<AcknowledgedResponse> {
         super(NAME);
     }
 
-    @Override
-    public AcknowledgedResponse newResponse() {
-        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-    }
-
     @Override
     public Writeable.Reader<AcknowledgedResponse> getResponseReader() {
         return AcknowledgedResponse::new;

server/src/main/java/org/elasticsearch/action/admin/indices/template/get/GetIndexTemplatesAction.java

@@ -18,9 +18,9 @@
  */
 package org.elasticsearch.action.admin.indices.template.get;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
-public class GetIndexTemplatesAction extends Action<GetIndexTemplatesResponse> {
+public class GetIndexTemplatesAction extends StreamableResponseAction<GetIndexTemplatesResponse> {
 
     public static final GetIndexTemplatesAction INSTANCE = new GetIndexTemplatesAction();
     public static final String NAME = "indices:admin/template/get";

server/src/main/java/org/elasticsearch/action/admin/indices/template/put/PutIndexTemplateAction.java

@@ -32,11 +32,6 @@ public class PutIndexTemplateAction extends Action<AcknowledgedResponse> {
         super(NAME);
     }
 
-    @Override
-    public AcknowledgedResponse newResponse() {
-        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-    }
-
     @Override
     public Writeable.Reader<AcknowledgedResponse> getResponseReader() {
         return AcknowledgedResponse::new;

server/src/main/java/org/elasticsearch/action/admin/indices/upgrade/get/UpgradeStatusAction.java

@@ -19,9 +19,9 @@
 
 package org.elasticsearch.action.admin.indices.upgrade.get;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
-public class UpgradeStatusAction extends Action<UpgradeStatusResponse> {
+public class UpgradeStatusAction extends StreamableResponseAction<UpgradeStatusResponse> {
 
     public static final UpgradeStatusAction INSTANCE = new UpgradeStatusAction();
     public static final String NAME = "indices:monitor/upgrade";

server/src/main/java/org/elasticsearch/action/admin/indices/upgrade/post/UpgradeAction.java

@@ -19,12 +19,12 @@
 
 package org.elasticsearch.action.admin.indices.upgrade.post;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
 /**
  * Upgrade index/indices action.
  */
-public class UpgradeAction extends Action<UpgradeResponse> {
+public class UpgradeAction extends StreamableResponseAction<UpgradeResponse> {
 
     public static final UpgradeAction INSTANCE = new UpgradeAction();
     public static final String NAME = "indices:admin/upgrade";

server/src/main/java/org/elasticsearch/action/admin/indices/upgrade/post/UpgradeSettingsAction.java

@@ -32,11 +32,6 @@ public class UpgradeSettingsAction extends Action<AcknowledgedResponse> {
         super(NAME);
     }
 
-    @Override
-    public AcknowledgedResponse newResponse() {
-        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-    }
-
     @Override
     public Writeable.Reader<AcknowledgedResponse> getResponseReader() {
         return AcknowledgedResponse::new;

server/src/main/java/org/elasticsearch/action/admin/indices/validate/query/ValidateQueryAction.java

@@ -19,9 +19,9 @@
 
 package org.elasticsearch.action.admin.indices.validate.query;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
-public class ValidateQueryAction extends Action<ValidateQueryResponse> {
+public class ValidateQueryAction extends StreamableResponseAction<ValidateQueryResponse> {
 
     public static final ValidateQueryAction INSTANCE = new ValidateQueryAction();
     public static final String NAME = "indices:admin/validate/query";

server/src/main/java/org/elasticsearch/action/bulk/BulkAction.java

@@ -19,11 +19,11 @@
 
 package org.elasticsearch.action.bulk;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.transport.TransportRequestOptions;
 
-public class BulkAction extends Action<BulkResponse> {
+public class BulkAction extends StreamableResponseAction<BulkResponse> {
 
     public static final BulkAction INSTANCE = new BulkAction();
     public static final String NAME = "indices:data/write/bulk";

server/src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java

@@ -29,6 +29,7 @@ import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionRunnable;
 import org.elasticsearch.action.DocWriteRequest;
 import org.elasticsearch.action.DocWriteResponse;
+import org.elasticsearch.action.StreamableResponseAction;
 import org.elasticsearch.action.delete.DeleteRequest;
 import org.elasticsearch.action.delete.DeleteResponse;
 import org.elasticsearch.action.index.IndexRequest;
@@ -78,7 +79,7 @@ import java.util.function.LongSupplier;
 public class TransportShardBulkAction extends TransportWriteAction<BulkShardRequest, BulkShardRequest, BulkShardResponse> {
 
     public static final String ACTION_NAME = BulkAction.NAME + "[s]";
-    public static final Action<BulkShardResponse> ACTION_INSTANCE = new Action<>(ACTION_NAME) {
+    public static final Action<BulkShardResponse> ACTION_INSTANCE = new StreamableResponseAction<>(ACTION_NAME) {
         @Override
         public BulkShardResponse newResponse() {
             return new BulkShardResponse();

server/src/main/java/org/elasticsearch/action/delete/DeleteAction.java

@@ -19,9 +19,9 @@
 
 package org.elasticsearch.action.delete;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
-public class DeleteAction extends Action<DeleteResponse> {
+public class DeleteAction extends StreamableResponseAction<DeleteResponse> {
 
     public static final DeleteAction INSTANCE = new DeleteAction();
     public static final String NAME = "indices:data/write/delete";

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

@@ -34,11 +34,6 @@ public class ExplainAction extends Action<ExplainResponse> {
         super(NAME);
     }
 
-    @Override
-    public ExplainResponse newResponse() {
-        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/fieldcaps/FieldCapabilitiesAction.java

@@ -19,9 +19,9 @@
 
 package org.elasticsearch.action.fieldcaps;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
-public class FieldCapabilitiesAction extends Action<FieldCapabilitiesResponse> {
+public class FieldCapabilitiesAction extends StreamableResponseAction<FieldCapabilitiesResponse> {
 
     public static final FieldCapabilitiesAction INSTANCE = new FieldCapabilitiesAction();
     public static final String NAME = "indices:data/read/field_caps";

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

@@ -20,7 +20,6 @@
 package org.elasticsearch.action.fieldcaps;
 
 import org.elasticsearch.action.Action;
-import org.elasticsearch.action.Action2;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.single.shard.TransportSingleShardAction;
 import org.elasticsearch.cluster.ClusterState;
@@ -50,7 +49,7 @@ public class TransportFieldCapabilitiesIndexAction extends TransportSingleShardA
 
     private static final String ACTION_NAME = FieldCapabilitiesAction.NAME + "[index]";
     public static final Action<FieldCapabilitiesIndexResponse> ACTION_INSTANCE =
-        new Action2<>(ACTION_NAME, FieldCapabilitiesIndexResponse::new);
+        new Action<>(ACTION_NAME, FieldCapabilitiesIndexResponse::new);
 
     private final IndicesService indicesService;
 

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

@@ -31,11 +31,6 @@ public class GetAction extends Action<GetResponse> {
         super(NAME);
     }
 
-    @Override
-    public GetResponse newResponse() {
-        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/MultiGetAction.java

@@ -19,9 +19,9 @@
 
 package org.elasticsearch.action.get;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
-public class MultiGetAction extends Action<MultiGetResponse> {
+public class MultiGetAction extends StreamableResponseAction<MultiGetResponse> {
 
     public static final MultiGetAction INSTANCE = new MultiGetAction();
     public static final String NAME = "indices:data/read/mget";

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

@@ -21,7 +21,6 @@ package org.elasticsearch.action.get;
 
 import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.elasticsearch.action.Action;
-import org.elasticsearch.action.Action2;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.TransportActions;
 import org.elasticsearch.action.support.single.shard.TransportSingleShardAction;
@@ -42,7 +41,7 @@ import org.elasticsearch.transport.TransportService;
 public class TransportShardMultiGetAction extends TransportSingleShardAction<MultiGetShardRequest, MultiGetShardResponse> {
 
     private static final String ACTION_NAME = MultiGetAction.NAME + "[shard]";
-    public static final Action<MultiGetShardResponse> ACTION_INSTANCE = new Action2<>(ACTION_NAME, MultiGetShardResponse::new);
+    public static final Action<MultiGetShardResponse> ACTION_INSTANCE = new Action<>(ACTION_NAME, MultiGetShardResponse::new);
 
     private final IndicesService indicesService;
 

server/src/main/java/org/elasticsearch/action/index/IndexAction.java

@@ -19,9 +19,9 @@
 
 package org.elasticsearch.action.index;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
-public class IndexAction extends Action<IndexResponse> {
+public class IndexAction extends StreamableResponseAction<IndexResponse> {
 
     public static final IndexAction INSTANCE = new IndexAction();
     public static final String NAME = "indices:data/write/index";

server/src/main/java/org/elasticsearch/action/ingest/DeletePipelineAction.java

@@ -32,11 +32,6 @@ public class DeletePipelineAction extends Action<AcknowledgedResponse> {
         super(NAME);
     }
 
-    @Override
-    public AcknowledgedResponse newResponse() {
-        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-    }
-
     @Override
     public Writeable.Reader<AcknowledgedResponse> getResponseReader() {
         return AcknowledgedResponse::new;

server/src/main/java/org/elasticsearch/action/ingest/GetPipelineAction.java

@@ -19,9 +19,9 @@
 
 package org.elasticsearch.action.ingest;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
-public class GetPipelineAction extends Action<GetPipelineResponse> {
+public class GetPipelineAction extends StreamableResponseAction<GetPipelineResponse> {
 
     public static final GetPipelineAction INSTANCE = new GetPipelineAction();
     public static final String NAME = "cluster:admin/ingest/pipeline/get";

server/src/main/java/org/elasticsearch/action/ingest/PutPipelineAction.java

@@ -32,11 +32,6 @@ public class PutPipelineAction extends Action<AcknowledgedResponse> {
         super(NAME);
     }
 
-    @Override
-    public AcknowledgedResponse newResponse() {
-        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-    }
-
     @Override
     public Writeable.Reader<AcknowledgedResponse> getResponseReader() {
         return AcknowledgedResponse::new;

server/src/main/java/org/elasticsearch/action/ingest/SimulatePipelineAction.java

@@ -19,9 +19,9 @@
 
 package org.elasticsearch.action.ingest;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
-public class SimulatePipelineAction extends Action<SimulatePipelineResponse> {
+public class SimulatePipelineAction extends StreamableResponseAction<SimulatePipelineResponse> {
 
     public static final SimulatePipelineAction INSTANCE = new SimulatePipelineAction();
     public static final String NAME = "cluster:admin/ingest/pipeline/simulate";

server/src/main/java/org/elasticsearch/action/main/MainAction.java

@@ -19,9 +19,9 @@
 
 package org.elasticsearch.action.main;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
-public class MainAction extends Action<MainResponse> {
+public class MainAction extends StreamableResponseAction<MainResponse> {
 
     public static final String NAME = "cluster:monitor/main";
     public static final MainAction INSTANCE = new MainAction();

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

@@ -19,9 +19,9 @@
 
 package org.elasticsearch.action.search;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
-public class ClearScrollAction extends Action<ClearScrollResponse> {
+public class ClearScrollAction extends StreamableResponseAction<ClearScrollResponse> {
 
     public static final ClearScrollAction INSTANCE = new ClearScrollAction();
     public static final String NAME = "indices:data/read/scroll/clear";

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

@@ -31,11 +31,6 @@ public class MultiSearchAction extends Action<MultiSearchResponse> {
         super(NAME);
     }
 
-    @Override
-    public MultiSearchResponse newResponse() {
-        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-    }
-
     @Override
     public Writeable.Reader<MultiSearchResponse> getResponseReader() {
         return MultiSearchResponse::new;

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

@@ -31,11 +31,6 @@ public class SearchAction extends Action<SearchResponse> {
         super(NAME);
     }
 
-    @Override
-    public SearchResponse newResponse() {
-        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/SearchScrollAction.java

@@ -31,11 +31,6 @@ public class SearchScrollAction extends Action<SearchResponse> {
         super(NAME);
     }
 
-    @Override
-    public SearchResponse newResponse() {
-        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/termvectors/MultiTermVectorsAction.java

@@ -19,9 +19,9 @@
 
 package org.elasticsearch.action.termvectors;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
-public class MultiTermVectorsAction extends Action<MultiTermVectorsResponse> {
+public class MultiTermVectorsAction extends StreamableResponseAction<MultiTermVectorsResponse> {
 
     public static final MultiTermVectorsAction INSTANCE = new MultiTermVectorsAction();
     public static final String NAME = "indices:data/read/mtv";

server/src/main/java/org/elasticsearch/action/termvectors/TermVectorsAction.java

@@ -31,11 +31,6 @@ public class TermVectorsAction extends Action<TermVectorsResponse> {
         super(NAME);
     }
 
-    @Override
-    public TermVectorsResponse newResponse() {
-        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-    }
-
     @Override
     public Writeable.Reader<TermVectorsResponse> getResponseReader() {
         return TermVectorsResponse::new;

server/src/main/java/org/elasticsearch/action/termvectors/TransportShardMultiTermsVectorAction.java

@@ -21,6 +21,7 @@ package org.elasticsearch.action.termvectors;
 
 import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.TransportActions;
 import org.elasticsearch.action.support.single.shard.TransportSingleShardAction;
@@ -44,7 +45,7 @@ public class TransportShardMultiTermsVectorAction extends
     private final IndicesService indicesService;
 
     private static final String ACTION_NAME = MultiTermVectorsAction.NAME + "[shard]";
-    public static final Action<MultiTermVectorsShardResponse> ACTION_INSTANCE = new Action<>(ACTION_NAME) {
+    public static final Action<MultiTermVectorsShardResponse> ACTION_INSTANCE = new StreamableResponseAction<>(ACTION_NAME) {
         @Override
         public MultiTermVectorsShardResponse newResponse() {
             return new MultiTermVectorsShardResponse();

server/src/main/java/org/elasticsearch/action/update/UpdateAction.java

@@ -19,9 +19,9 @@
 
 package org.elasticsearch.action.update;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
-public class UpdateAction extends Action<UpdateResponse> {
+public class UpdateAction extends StreamableResponseAction<UpdateResponse> {
 
     public static final UpdateAction INSTANCE = new UpdateAction();
     public static final String NAME = "indices:data/write/update";

server/src/main/java/org/elasticsearch/index/reindex/DeleteByQueryAction.java

@@ -19,9 +19,9 @@
 
 package org.elasticsearch.index.reindex;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
-public class DeleteByQueryAction extends Action<BulkByScrollResponse> {
+public class DeleteByQueryAction extends StreamableResponseAction<BulkByScrollResponse> {
 
     public static final DeleteByQueryAction INSTANCE = new DeleteByQueryAction();
     public static final String NAME = "indices:data/write/delete/byquery";

server/src/main/java/org/elasticsearch/index/reindex/ReindexAction.java

@@ -19,9 +19,9 @@
 
 package org.elasticsearch.index.reindex;
 
-import org.elasticsearch.action.Action;
+import org.elasticsearch.action.StreamableResponseAction;
 
-public class ReindexAction extends Action<BulkByScrollResponse> {
+public class ReindexAction extends StreamableResponseAction<BulkByScrollResponse> {
     public static final ReindexAction INSTANCE = new ReindexAction();
     public static final String NAME = "indices:data/write/reindex";