Merge branch 'master' into dont_cache_scripts

148 files changed, 1191 insertions, 770 deletions

diff --git a/client/client-benchmark-noop-api-plugin/src/main/java/org/elasticsearch/plugin/noop/action/bulk/RestNoopBulkAction.java b/client/client-benchmark-noop-api-plugin/src/main/java/org/elasticsearch/plugin/noop/action/bulk/RestNoopBulkAction.java
index 2c6b88954a..b9f7e25614 100644
--- a/client/client-benchmark-noop-api-plugin/src/main/java/org/elasticsearch/plugin/noop/action/bulk/RestNoopBulkAction.java
+++ b/client/client-benchmark-noop-api-plugin/src/main/java/org/elasticsearch/plugin/noop/action/bulk/RestNoopBulkAction.java
@@ -39,6 +39,8 @@ import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
 import org.elasticsearch.rest.action.RestBuilderListener;
 
+import java.io.IOException;
+
 import static org.elasticsearch.rest.RestRequest.Method.POST;
 import static org.elasticsearch.rest.RestRequest.Method.PUT;
 import static org.elasticsearch.rest.RestStatus.OK;
@@ -57,7 +59,7 @@ public class RestNoopBulkAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) throws Exception {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         BulkRequest bulkRequest = Requests.bulkRequest();
         String defaultIndex = request.param("index");
         String defaultType = request.param("type");
@@ -75,9 +77,10 @@ public class RestNoopBulkAction extends BaseRestHandler {
         bulkRequest.add(request.content(), defaultIndex, defaultType, defaultRouting, defaultFields, null, defaultPipeline, null, true);
 
         // short circuit the call to the transport layer
-        BulkRestBuilderListener listener = new BulkRestBuilderListener(channel, request);
-        listener.onResponse(bulkRequest);
-
+        return channel -> {
+            BulkRestBuilderListener listener = new BulkRestBuilderListener(channel, request);
+            listener.onResponse(bulkRequest);
+        };
     }
 
     private static class BulkRestBuilderListener extends RestBuilderListener<BulkRequest> {
diff --git a/client/client-benchmark-noop-api-plugin/src/main/java/org/elasticsearch/plugin/noop/action/search/RestNoopSearchAction.java b/client/client-benchmark-noop-api-plugin/src/main/java/org/elasticsearch/plugin/noop/action/search/RestNoopSearchAction.java
index 3520876af0..9bcde4ca39 100644
--- a/client/client-benchmark-noop-api-plugin/src/main/java/org/elasticsearch/plugin/noop/action/search/RestNoopSearchAction.java
+++ b/client/client-benchmark-noop-api-plugin/src/main/java/org/elasticsearch/plugin/noop/action/search/RestNoopSearchAction.java
@@ -23,7 +23,6 @@ import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.RestStatusToXContentListener;
@@ -47,8 +46,8 @@ public class RestNoopSearchAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) throws IOException {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         SearchRequest searchRequest = new SearchRequest();
-        client.execute(NoopSearchAction.INSTANCE, searchRequest, new RestStatusToXContentListener<>(channel));
+        return channel -> client.execute(NoopSearchAction.INSTANCE, searchRequest, new RestStatusToXContentListener<>(channel));
     }
 }
diff --git a/core/src/main/java/org/elasticsearch/action/bulk/BulkRequest.java b/core/src/main/java/org/elasticsearch/action/bulk/BulkRequest.java
index f766c2897f..b9c615c260 100644
--- a/core/src/main/java/org/elasticsearch/action/bulk/BulkRequest.java
+++ b/core/src/main/java/org/elasticsearch/action/bulk/BulkRequest.java
@@ -247,32 +247,32 @@ public class BulkRequest extends ActionRequest<BulkRequest> implements Composite
     /**
      * Adds a framed data in binary format
      */
-    public BulkRequest add(byte[] data, int from, int length) throws Exception {
+    public BulkRequest add(byte[] data, int from, int length) throws IOException {
         return add(data, from, length, null, null);
     }
 
     /**
      * Adds a framed data in binary format
      */
-    public BulkRequest add(byte[] data, int from, int length, @Nullable String defaultIndex, @Nullable String defaultType) throws Exception {
+    public BulkRequest add(byte[] data, int from, int length, @Nullable String defaultIndex, @Nullable String defaultType) throws IOException {
         return add(new BytesArray(data, from, length), defaultIndex, defaultType);
     }
 
     /**
      * Adds a framed data in binary format
      */
-    public BulkRequest add(BytesReference data, @Nullable String defaultIndex, @Nullable String defaultType) throws Exception {
+    public BulkRequest add(BytesReference data, @Nullable String defaultIndex, @Nullable String defaultType) throws IOException {
         return add(data, defaultIndex, defaultType, null, null, null, null, null, true);
     }
 
     /**
      * Adds a framed data in binary format
      */
-    public BulkRequest add(BytesReference data, @Nullable String defaultIndex, @Nullable String defaultType, boolean allowExplicitIndex) throws Exception {
+    public BulkRequest add(BytesReference data, @Nullable String defaultIndex, @Nullable String defaultType, boolean allowExplicitIndex) throws IOException {
         return add(data, defaultIndex, defaultType, null, null, null, null, null, allowExplicitIndex);
     }
 
-    public BulkRequest add(BytesReference data, @Nullable String defaultIndex, @Nullable String defaultType, @Nullable String defaultRouting, @Nullable String[] defaultFields, @Nullable FetchSourceContext defaultFetchSourceContext, @Nullable String defaultPipeline, @Nullable Object payload, boolean allowExplicitIndex) throws Exception {
+    public BulkRequest add(BytesReference data, @Nullable String defaultIndex, @Nullable String defaultType, @Nullable String defaultRouting, @Nullable String[] defaultFields, @Nullable FetchSourceContext defaultFetchSourceContext, @Nullable String defaultPipeline, @Nullable Object payload, boolean allowExplicitIndex) throws IOException {
         XContent xContent = XContentFactory.xContent(data);
         int line = 0;
         int from = 0;
diff --git a/core/src/main/java/org/elasticsearch/action/get/MultiGetRequest.java b/core/src/main/java/org/elasticsearch/action/get/MultiGetRequest.java
index 1decd8ce94..a0fa4025ec 100644
--- a/core/src/main/java/org/elasticsearch/action/get/MultiGetRequest.java
+++ b/core/src/main/java/org/elasticsearch/action/get/MultiGetRequest.java
@@ -335,7 +335,7 @@ public class MultiGetRequest extends ActionRequest<MultiGetRequest> implements I
         return add(defaultIndex, defaultType, defaultFields, defaultFetchSource, null, data, allowExplicitIndex);
     }
 
-    public MultiGetRequest add(@Nullable String defaultIndex, @Nullable String defaultType, @Nullable String[] defaultFields, @Nullable FetchSourceContext defaultFetchSource, @Nullable String defaultRouting, BytesReference data, boolean allowExplicitIndex) throws Exception {
+    public MultiGetRequest add(@Nullable String defaultIndex, @Nullable String defaultType, @Nullable String[] defaultFields, @Nullable FetchSourceContext defaultFetchSource, @Nullable String defaultRouting, BytesReference data, boolean allowExplicitIndex) throws IOException {
         try (XContentParser parser = XContentFactory.xContent(data).createParser(data)) {
             XContentParser.Token token;
             String currentFieldName = null;
diff --git a/core/src/main/java/org/elasticsearch/action/support/tasks/BaseTasksRequest.java b/core/src/main/java/org/elasticsearch/action/support/tasks/BaseTasksRequest.java
index abc081dbc0..18eea411b6 100644
--- a/core/src/main/java/org/elasticsearch/action/support/tasks/BaseTasksRequest.java
+++ b/core/src/main/java/org/elasticsearch/action/support/tasks/BaseTasksRequest.java
@@ -42,7 +42,7 @@ public class BaseTasksRequest<Request extends BaseTasksRequest<Request>> extends
 
     public static final String[] ALL_NODES = Strings.EMPTY_ARRAY;
 
-    private String[] nodesIds = ALL_NODES;
+    private String[] nodes = ALL_NODES;
 
     private TimeValue timeout;
 
@@ -58,7 +58,7 @@ public class BaseTasksRequest<Request extends BaseTasksRequest<Request>> extends
     @Override
     public ActionRequestValidationException validate() {
         ActionRequestValidationException validationException = null;
-        if (taskId.isSet() && nodesIds.length > 0) {
+        if (taskId.isSet() && nodes.length > 0) {
             validationException = addValidationError("task id cannot be used together with node ids",
                 validationException);
         }
@@ -81,13 +81,13 @@ public class BaseTasksRequest<Request extends BaseTasksRequest<Request>> extends
         return actions;
     }
 
-    public final String[] getNodesIds() {
-        return nodesIds;
+    public final String[] getNodes() {
+        return nodes;
     }
 
     @SuppressWarnings("unchecked")
-    public final Request setNodesIds(String... nodesIds) {
-        this.nodesIds = nodesIds;
+    public final Request setNodes(String... nodes) {
+        this.nodes = nodes;
         return (Request) this;
     }
 
@@ -142,7 +142,7 @@ public class BaseTasksRequest<Request extends BaseTasksRequest<Request>> extends
         super.readFrom(in);
         taskId = TaskId.readFromStream(in);
         parentTaskId = TaskId.readFromStream(in);
-        nodesIds = in.readStringArray();
+        nodes = in.readStringArray();
         actions = in.readStringArray();
         timeout = in.readOptionalWriteable(TimeValue::new);
     }
@@ -152,7 +152,7 @@ public class BaseTasksRequest<Request extends BaseTasksRequest<Request>> extends
         super.writeTo(out);
         taskId.writeTo(out);
         parentTaskId.writeTo(out);
-        out.writeStringArrayNullable(nodesIds);
+        out.writeStringArrayNullable(nodes);
         out.writeStringArrayNullable(actions);
         out.writeOptionalWriteable(timeout);
     }
diff --git a/core/src/main/java/org/elasticsearch/action/support/tasks/TasksRequestBuilder.java b/core/src/main/java/org/elasticsearch/action/support/tasks/TasksRequestBuilder.java
index a3528cb75c..6ba44c3135 100644
--- a/core/src/main/java/org/elasticsearch/action/support/tasks/TasksRequestBuilder.java
+++ b/core/src/main/java/org/elasticsearch/action/support/tasks/TasksRequestBuilder.java
@@ -48,7 +48,7 @@ public class TasksRequestBuilder<
 
     @SuppressWarnings("unchecked")
     public final RequestBuilder setNodesIds(String... nodesIds) {
-        request.setNodesIds(nodesIds);
+        request.setNodes(nodesIds);
         return (RequestBuilder) this;
     }
 
diff --git a/core/src/main/java/org/elasticsearch/action/support/tasks/TransportTasksAction.java b/core/src/main/java/org/elasticsearch/action/support/tasks/TransportTasksAction.java
index 6752ccd729..d9f035dcc4 100644
--- a/core/src/main/java/org/elasticsearch/action/support/tasks/TransportTasksAction.java
+++ b/core/src/main/java/org/elasticsearch/action/support/tasks/TransportTasksAction.java
@@ -125,7 +125,7 @@ public abstract class TransportTasksAction<
         if (request.getTaskId().isSet()) {
             return new String[]{request.getTaskId().getNodeId()};
         } else {
-            return clusterState.nodes().resolveNodes(request.getNodesIds());
+            return clusterState.nodes().resolveNodes(request.getNodes());
         }
     }
 
diff --git a/core/src/main/java/org/elasticsearch/action/termvectors/MultiTermVectorsRequest.java b/core/src/main/java/org/elasticsearch/action/termvectors/MultiTermVectorsRequest.java
index af192cea60..3cd73226e7 100644
--- a/core/src/main/java/org/elasticsearch/action/termvectors/MultiTermVectorsRequest.java
+++ b/core/src/main/java/org/elasticsearch/action/termvectors/MultiTermVectorsRequest.java
@@ -94,7 +94,7 @@ public class MultiTermVectorsRequest extends ActionRequest<MultiTermVectorsReque
         return requests;
     }
 
-    public void add(TermVectorsRequest template, BytesReference data) throws Exception {
+    public void add(TermVectorsRequest template, BytesReference data) throws IOException {
         XContentParser.Token token;
         String currentFieldName = null;
         if (data.length() > 0) {
diff --git a/core/src/main/java/org/elasticsearch/action/update/UpdateRequest.java b/core/src/main/java/org/elasticsearch/action/update/UpdateRequest.java
index 9fe5cd892d..3ec2398d2f 100644
--- a/core/src/main/java/org/elasticsearch/action/update/UpdateRequest.java
+++ b/core/src/main/java/org/elasticsearch/action/update/UpdateRequest.java
@@ -715,7 +715,7 @@ public class UpdateRequest extends InstanceShardOperationRequest<UpdateRequest>
         return detectNoop;
     }
 
-    public UpdateRequest fromXContent(BytesReference source) throws Exception {
+    public UpdateRequest fromXContent(BytesReference source) throws IOException {
         Script script = null;
         try (XContentParser parser = XContentFactory.xContent(source).createParser(source)) {
             XContentParser.Token token = parser.nextToken();
diff --git a/core/src/main/java/org/elasticsearch/common/settings/Settings.java b/core/src/main/java/org/elasticsearch/common/settings/Settings.java
index 9e5dd0efbe..819edc246a 100644
--- a/core/src/main/java/org/elasticsearch/common/settings/Settings.java
+++ b/core/src/main/java/org/elasticsearch/common/settings/Settings.java
@@ -582,6 +582,9 @@ public final class Settings implements ToXContent {
         return builder;
     }
 
+    public static final Set<String> FORMAT_PARAMS =
+        Collections.unmodifiableSet(new HashSet<>(Arrays.asList("settings_filter", "flat_settings")));
+
     /**
      * Returns <tt>true</tt> if this settings object contains no settings
      * @return <tt>true</tt> if this settings object contains no settings
diff --git a/core/src/main/java/org/elasticsearch/index/mapper/TTLFieldMapper.java b/core/src/main/java/org/elasticsearch/index/mapper/TTLFieldMapper.java
index f95f42156e..bb2da75dad 100644
--- a/core/src/main/java/org/elasticsearch/index/mapper/TTLFieldMapper.java
+++ b/core/src/main/java/org/elasticsearch/index/mapper/TTLFieldMapper.java
@@ -273,4 +273,5 @@ public class TTLFieldMapper extends MetadataFieldMapper {
             }
         }
     }
+
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/AbstractRestChannel.java b/core/src/main/java/org/elasticsearch/rest/AbstractRestChannel.java
index f146267c9b..4b3505e97e 100644
--- a/core/src/main/java/org/elasticsearch/rest/AbstractRestChannel.java
+++ b/core/src/main/java/org/elasticsearch/rest/AbstractRestChannel.java
@@ -40,12 +40,20 @@ public abstract class AbstractRestChannel implements RestChannel {
 
     protected final RestRequest request;
     protected final boolean detailedErrorsEnabled;
+    private final String format;
+    private final String filterPath;
+    private final boolean pretty;
+    private final boolean human;
 
     private BytesStreamOutput bytesOut;
 
     protected AbstractRestChannel(RestRequest request, boolean detailedErrorsEnabled) {
         this.request = request;
         this.detailedErrorsEnabled = detailedErrorsEnabled;
+        this.format = request.param("format", request.header("Accept"));
+        this.filterPath = request.param("filter_path", null);
+        this.pretty = request.paramAsBoolean("pretty", false);
+        this.human = request.paramAsBoolean("human", false);
     }
 
     @Override
@@ -61,7 +69,7 @@ public abstract class AbstractRestChannel implements RestChannel {
 
     @Override
     public XContentBuilder newBuilder(@Nullable BytesReference autoDetectSource, boolean useFiltering) throws IOException {
-        XContentType contentType = XContentType.fromMediaTypeOrFormat(request.param("format", request.header("Accept")));
+        XContentType contentType = XContentType.fromMediaTypeOrFormat(format);
         if (contentType == null) {
             // try and guess it from the auto detect source
             if (autoDetectSource != null) {
@@ -76,17 +84,17 @@ public abstract class AbstractRestChannel implements RestChannel {
         Set<String> includes = Collections.emptySet();
         Set<String> excludes = Collections.emptySet();
         if (useFiltering) {
-            Set<String> filters = Strings.splitStringByCommaToSet(request.param("filter_path", null));
+            Set<String> filters = Strings.splitStringByCommaToSet(filterPath);
             includes = filters.stream().filter(INCLUDE_FILTER).collect(toSet());
             excludes = filters.stream().filter(EXCLUDE_FILTER).map(f -> f.substring(1)).collect(toSet());
         }
 
         XContentBuilder builder = new XContentBuilder(XContentFactory.xContent(contentType), bytesOutput(), includes, excludes);
-        if (request.paramAsBoolean("pretty", false)) {
+        if (pretty) {
             builder.prettyPrint().lfAtEnd();
         }
 
-        builder.humanReadable(request.paramAsBoolean("human", builder.humanReadable()));
+        builder.humanReadable(human);
         return builder;
     }
 
diff --git a/core/src/main/java/org/elasticsearch/rest/BaseRestHandler.java b/core/src/main/java/org/elasticsearch/rest/BaseRestHandler.java
index 31e09d6706..10f5359803 100644
--- a/core/src/main/java/org/elasticsearch/rest/BaseRestHandler.java
+++ b/core/src/main/java/org/elasticsearch/rest/BaseRestHandler.java
@@ -19,6 +19,7 @@
 
 package org.elasticsearch.rest;
 
+import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.ParseFieldMatcher;
 import org.elasticsearch.common.component.AbstractComponent;
 import org.elasticsearch.common.settings.Setting;
@@ -26,6 +27,12 @@ import org.elasticsearch.common.settings.Setting.Property;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.plugins.ActionPlugin;
 
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
 /**
  * Base handler for REST requests.
  * <p>
@@ -35,6 +42,7 @@ import org.elasticsearch.plugins.ActionPlugin;
  * {@link ActionPlugin#getRestHeaders()}.
  */
 public abstract class BaseRestHandler extends AbstractComponent implements RestHandler {
+
     public static final Setting<Boolean> MULTI_ALLOW_EXPLICIT_INDEX =
         Setting.boolSetting("rest.action.multi.allow_explicit_index", true, Property.NodeScope);
     protected final ParseFieldMatcher parseFieldMatcher;
@@ -43,4 +51,65 @@ public abstract class BaseRestHandler extends AbstractComponent implements RestH
         super(settings);
         this.parseFieldMatcher = new ParseFieldMatcher(settings);
     }
+
+    @Override
+    public final void handleRequest(RestRequest request, RestChannel channel, NodeClient client) throws Exception {
+        // prepare the request for execution; has the side effect of touching the request parameters
+        final RestChannelConsumer action = prepareRequest(request, client);
+
+        // validate unconsumed params, but we must exclude params used to format the response
+        final List<String> unconsumedParams =
+            request.unconsumedParams().stream().filter(p -> !responseParams().contains(p)).collect(Collectors.toList());
+
+        // validate the non-response params
+        if (!unconsumedParams.isEmpty()) {
+            throw new IllegalArgumentException("request [" + request.path() + "] contains unused params: " + unconsumedParams.toString());
+        }
+
+        // execute the action
+        action.accept(channel);
+    }
+
+    /**
+     * REST requests are handled by preparing a channel consumer that represents the execution of
+     * the request against a channel.
+     */
+    @FunctionalInterface
+    protected interface RestChannelConsumer {
+        /**
+         * Executes a request against the given channel.
+         *
+         * @param channel the channel for sending the response
+         * @throws Exception if an exception occurred executing the request
+         */
+        void accept(RestChannel channel) throws Exception;
+    }
+
+    /**
+     * Prepare the request for execution. Implementations should consume all request params before
+     * returning the runnable for actual execution. Unconsumed params will immediately terminate
+     * execution of the request. However, some params are only used in processing the response;
+     * implementations can override {@link BaseRestHandler#responseParams()} to indicate such
+     * params.
+     *
+     * @param request the request to execute
+     * @param client  client for executing actions on the local node
+     * @return the action to execute
+     * @throws IOException if an I/O exception occurred parsing the request and preparing for
+     *                     execution
+     */
+    protected abstract RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException;
+
+    /**
+     * Parameters used for controlling the response and thus might not be consumed during
+     * preparation of the request execution in
+     * {@link BaseRestHandler#prepareRequest(RestRequest, NodeClient)}.
+     *
+     * @return a set of parameters used to control the response and thus should not trip strict
+     * URL parameter checks.
+     */
+    protected Set<String> responseParams() {
+        return Collections.emptySet();
+    }
+
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/RestController.java b/core/src/main/java/org/elasticsearch/rest/RestController.java
index e63f35884e..6a3a399705 100644
--- a/core/src/main/java/org/elasticsearch/rest/RestController.java
+++ b/core/src/main/java/org/elasticsearch/rest/RestController.java
@@ -220,10 +220,11 @@ public class RestController extends AbstractLifecycleComponent {
      */
     boolean checkRequestParameters(final RestRequest request, final RestChannel channel) {
         // error_trace cannot be used when we disable detailed errors
-        if (channel.detailedErrorsEnabled() == false && request.paramAsBoolean("error_trace", false)) {
+        // we consume the error_trace parameter first to ensure that it is always consumed
+        if (request.paramAsBoolean("error_trace", false) && channel.detailedErrorsEnabled() == false) {
             try {
                 XContentBuilder builder = channel.newErrorBuilder();
-                builder.startObject().field("error","error traces in responses are disabled.").endObject().string();
+                builder.startObject().field("error", "error traces in responses are disabled.").endObject().string();
                 RestResponse response = new BytesRestResponse(BAD_REQUEST, builder);
                 response.addHeader("Content-Type", "application/json");
                 channel.sendResponse(response);
diff --git a/core/src/main/java/org/elasticsearch/rest/RestRequest.java b/core/src/main/java/org/elasticsearch/rest/RestRequest.java
index 2db917dacf..5960fc8979 100644
--- a/core/src/main/java/org/elasticsearch/rest/RestRequest.java
+++ b/core/src/main/java/org/elasticsearch/rest/RestRequest.java
@@ -30,7 +30,11 @@ import org.elasticsearch.common.xcontent.ToXContent;
 import java.net.SocketAddress;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
 import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
 
 import static org.elasticsearch.common.unit.ByteSizeValue.parseBytesSizeValue;
 import static org.elasticsearch.common.unit.TimeValue.parseTimeValue;
@@ -39,6 +43,7 @@ public abstract class RestRequest implements ToXContent.Params {
 
     private final Map<String, String> params;
     private final String rawPath;
+    private final Set<String> consumedParams = new HashSet<>();
 
     public RestRequest(String uri) {
         final Map<String, String> params = new HashMap<>();
@@ -106,11 +111,13 @@ public abstract class RestRequest implements ToXContent.Params {
 
     @Override
     public final String param(String key) {
+        consumedParams.add(key);
         return params.get(key);
     }
 
     @Override
     public final String param(String key, String defaultValue) {
+        consumedParams.add(key);
         String value = params.get(key);
         if (value == null) {
             return defaultValue;
@@ -122,6 +129,20 @@ public abstract class RestRequest implements ToXContent.Params {
         return params;
     }
 
+    /**
+     * Returns a list of parameters that have not yet been consumed. This method returns a copy,
+     * callers are free to modify the returned list.
+     *
+     * @return the list of currently unconsumed parameters.
+     */
+    List<String> unconsumedParams() {
+        return params
+            .keySet()
+            .stream()
+            .filter(p -> !consumedParams.contains(p))
+            .collect(Collectors.toList());
+    }
+
     public float paramAsFloat(String key, float defaultValue) {
         String sValue = param(key);
         if (sValue == null) {
diff --git a/core/src/main/java/org/elasticsearch/rest/action/RestFieldStatsAction.java b/core/src/main/java/org/elasticsearch/rest/action/RestFieldStatsAction.java
index e6ef620db1..080fbbfb7a 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/RestFieldStatsAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/RestFieldStatsAction.java
@@ -30,20 +30,18 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
 import org.elasticsearch.rest.RestStatus;
 
+import java.io.IOException;
 import java.util.Map;
 
 import static org.elasticsearch.rest.RestRequest.Method.GET;
 import static org.elasticsearch.rest.RestRequest.Method.POST;
 import static org.elasticsearch.rest.action.RestActions.buildBroadcastShardsHeader;
 
-/**
- */
 public class RestFieldStatsAction extends BaseRestHandler {
 
     @Inject
@@ -56,8 +54,8 @@ public class RestFieldStatsAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request,
-                              final RestChannel channel, final NodeClient client) throws Exception {
+    public RestChannelConsumer prepareRequest(final RestRequest request,
+                                              final NodeClient client) throws IOException {
         if (RestActions.hasBodyContent(request) && request.hasParam("fields")) {
             throw new IllegalArgumentException("can't specify a request body and [fields] request parameter, " +
                 "either specify a request body or the [fields] request parameter");
@@ -73,7 +71,7 @@ public class RestFieldStatsAction extends BaseRestHandler {
             fieldStatsRequest.setFields(Strings.splitStringByCommaToArray(request.param("fields")));
         }
 
-        client.fieldStats(fieldStatsRequest, new RestBuilderListener<FieldStatsResponse>(channel) {
+        return channel -> client.fieldStats(fieldStatsRequest, new RestBuilderListener<FieldStatsResponse>(channel) {
             @Override
             public RestResponse buildResponse(FieldStatsResponse response, XContentBuilder builder) throws Exception {
                 builder.startObject();
@@ -81,7 +79,7 @@ public class RestFieldStatsAction extends BaseRestHandler {
 
                 builder.startObject("indices");
                 for (Map.Entry<String, Map<String, FieldStats>> entry1 :
-                    response.getIndicesMergedFieldStats().entrySet()) {
+                        response.getIndicesMergedFieldStats().entrySet()) {
                     builder.startObject(entry1.getKey());
                     builder.startObject("fields");
                     for (Map.Entry<String, FieldStats> entry2 : entry1.getValue().entrySet()) {
diff --git a/core/src/main/java/org/elasticsearch/rest/action/RestMainAction.java b/core/src/main/java/org/elasticsearch/rest/action/RestMainAction.java
index 56053f414b..dca1c2c86d 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/RestMainAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/RestMainAction.java
@@ -28,7 +28,6 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
@@ -39,9 +38,6 @@ import java.io.IOException;
 import static org.elasticsearch.rest.RestRequest.Method.GET;
 import static org.elasticsearch.rest.RestRequest.Method.HEAD;
 
-/**
- *
- */
 public class RestMainAction extends BaseRestHandler {
 
     @Inject
@@ -52,8 +48,8 @@ public class RestMainAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) throws Exception {
-        client.execute(MainAction.INSTANCE, new MainRequest(), new RestBuilderListener<MainResponse>(channel) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
+        return channel -> client.execute(MainAction.INSTANCE, new MainRequest(), new RestBuilderListener<MainResponse>(channel) {
             @Override
             public RestResponse buildResponse(MainResponse mainResponse, XContentBuilder builder) throws Exception {
                 return convertMainResponse(mainResponse, request, builder);
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestCancelTasksAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestCancelTasksAction.java
index 3c558fba93..b6a4e743d1 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestCancelTasksAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestCancelTasksAction.java
@@ -26,11 +26,12 @@ import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.tasks.TaskId;
 
+import java.io.IOException;
+
 import static org.elasticsearch.rest.RestRequest.Method.POST;
 import static org.elasticsearch.rest.action.admin.cluster.RestListTasksAction.listTasksResponseListener;
 
@@ -43,26 +44,29 @@ public class RestCancelTasksAction extends BaseRestHandler {
         super(settings);
         this.clusterService = clusterService;
         controller.registerHandler(POST, "/_tasks/_cancel", this);
-        controller.registerHandler(POST, "/_tasks/{taskId}/_cancel", this);
+        controller.registerHandler(POST, "/_tasks/{task_id}/_cancel", this);
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
-        String[] nodesIds = Strings.splitStringByCommaToArray(request.param("nodeId"));
-        TaskId taskId = new TaskId(request.param("taskId"));
-        String[] actions = Strings.splitStringByCommaToArray(request.param("actions"));
-        TaskId parentTaskId = new TaskId(request.param("parent_task_id"));
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
+        final String[] nodesIds = Strings.splitStringByCommaToArray(request.param("nodes"));
+        final TaskId taskId = new TaskId(request.param("task_id"));
+        final String[] actions = Strings.splitStringByCommaToArray(request.param("actions"));
+        final TaskId parentTaskId = new TaskId(request.param("parent_task_id"));
+        final String groupBy = request.param("group_by", "nodes");
 
         CancelTasksRequest cancelTasksRequest = new CancelTasksRequest();
         cancelTasksRequest.setTaskId(taskId);
-        cancelTasksRequest.setNodesIds(nodesIds);
+        cancelTasksRequest.setNodes(nodesIds);
         cancelTasksRequest.setActions(actions);
         cancelTasksRequest.setParentTaskId(parentTaskId);
-        client.admin().cluster().cancelTasks(cancelTasksRequest, listTasksResponseListener(clusterService, channel));
+        return channel ->
+            client.admin().cluster().cancelTasks(cancelTasksRequest, listTasksResponseListener(clusterService, groupBy, channel));
     }
 
     @Override
     public boolean canTripCircuitBreaker() {
         return false;
     }
+
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestClusterAllocationExplainAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestClusterAllocationExplainAction.java
index bae1d1b671..3add5cb039 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestClusterAllocationExplainAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestClusterAllocationExplainAction.java
@@ -19,8 +19,6 @@
 
 package org.elasticsearch.rest.action.admin.cluster;
 
-import java.io.IOException;
-
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.action.admin.cluster.allocation.ClusterAllocationExplainRequest;
 import org.elasticsearch.action.admin.cluster.allocation.ClusterAllocationExplainResponse;
@@ -35,7 +33,6 @@ import org.elasticsearch.common.xcontent.XContentFactory;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
@@ -43,6 +40,8 @@ import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.rest.action.RestActions;
 import org.elasticsearch.rest.action.RestBuilderListener;
 
+import java.io.IOException;
+
 /**
  * Class handling cluster allocation explanation at the REST level
  */
@@ -56,7 +55,7 @@ public class RestClusterAllocationExplainAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         ClusterAllocationExplainRequest req;
         if (RestActions.hasBodyContent(request) == false) {
             // Empty request signals "explain the first unassigned shard you find"
@@ -67,16 +66,16 @@ public class RestClusterAllocationExplainAction extends BaseRestHandler {
                 req = ClusterAllocationExplainRequest.parse(parser);
             } catch (IOException e) {
                 logger.debug("failed to parse allocation explain request", e);
-                channel.sendResponse(
-                    new BytesRestResponse(ExceptionsHelper.status(e), BytesRestResponse.TEXT_CONTENT_TYPE, BytesArray.EMPTY));
-                return;
+                return channel -> channel.sendResponse(
+                        new BytesRestResponse(ExceptionsHelper.status(e), BytesRestResponse.TEXT_CONTENT_TYPE, BytesArray.EMPTY));
             }
         }
 
         try {
             req.includeYesDecisions(request.paramAsBoolean("include_yes_decisions", false));
             req.includeDiskInfo(request.paramAsBoolean("include_disk_info", false));
-            client.admin().cluster().allocationExplain(req, new RestBuilderListener<ClusterAllocationExplainResponse>(channel) {
+            return channel ->
+                    client.admin().cluster().allocationExplain(req, new RestBuilderListener<ClusterAllocationExplainResponse>(channel) {
                 @Override
                 public RestResponse buildResponse(ClusterAllocationExplainResponse response, XContentBuilder builder) throws Exception {
                     response.getExplanation().toXContent(builder, ToXContent.EMPTY_PARAMS);
@@ -85,7 +84,9 @@ public class RestClusterAllocationExplainAction extends BaseRestHandler {
             });
         } catch (Exception e) {
             logger.error("failed to explain allocation", e);
-            channel.sendResponse(new BytesRestResponse(ExceptionsHelper.status(e), BytesRestResponse.TEXT_CONTENT_TYPE, BytesArray.EMPTY));
+            return channel ->
+                    channel.sendResponse(
+                            new BytesRestResponse(ExceptionsHelper.status(e), BytesRestResponse.TEXT_CONTENT_TYPE, BytesArray.EMPTY));
         }
     }
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestClusterGetSettingsAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestClusterGetSettingsAction.java
index ca2cbaf79f..e55603a19b 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestClusterGetSettingsAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestClusterGetSettingsAction.java
@@ -32,7 +32,6 @@ import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
@@ -40,6 +39,7 @@ import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.rest.action.RestBuilderListener;
 
 import java.io.IOException;
+import java.util.Set;
 
 public class RestClusterGetSettingsAction extends BaseRestHandler {
 
@@ -56,13 +56,13 @@ public class RestClusterGetSettingsAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         ClusterStateRequest clusterStateRequest = Requests.clusterStateRequest()
                 .routingTable(false)
                 .nodes(false);
         final boolean renderDefaults = request.paramAsBoolean("include_defaults", false);
         clusterStateRequest.local(request.paramAsBoolean("local", clusterStateRequest.local()));
-        client.admin().cluster().state(clusterStateRequest, new RestBuilderListener<ClusterStateResponse>(channel) {
+        return channel -> client.admin().cluster().state(clusterStateRequest, new RestBuilderListener<ClusterStateResponse>(channel) {
             @Override
             public RestResponse buildResponse(ClusterStateResponse response, XContentBuilder builder) throws Exception {
                 return new BytesRestResponse(RestStatus.OK, renderResponse(response.getState(), renderDefaults, builder, request));
@@ -71,6 +71,11 @@ public class RestClusterGetSettingsAction extends BaseRestHandler {
     }
 
     @Override
+    protected Set<String> responseParams() {
+        return Settings.FORMAT_PARAMS;
+    }
+
+    @Override
     public boolean canTripCircuitBreaker() {
         return false;
     }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestClusterHealthAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestClusterHealthAction.java
index 5f64bcf8aa..ad51c9f1e6 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestClusterHealthAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestClusterHealthAction.java
@@ -20,7 +20,6 @@
 package org.elasticsearch.rest.action.admin.cluster;
 
 import org.elasticsearch.action.admin.cluster.health.ClusterHealthRequest;
-import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
 import org.elasticsearch.action.support.ActiveShardCount;
 import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.cluster.health.ClusterHealthStatus;
@@ -29,12 +28,14 @@ import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.RestStatusToXContentListener;
 
+import java.io.IOException;
+import java.util.Collections;
 import java.util.Locale;
+import java.util.Set;
 
 import static org.elasticsearch.client.Requests.clusterHealthRequest;
 
@@ -49,7 +50,7 @@ public class RestClusterHealthAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         ClusterHealthRequest clusterHealthRequest = clusterHealthRequest(Strings.splitStringByCommaToArray(request.param("index")));
         clusterHealthRequest.local(request.paramAsBoolean("local", clusterHealthRequest.local()));
         clusterHealthRequest.masterNodeTimeout(request.paramAsTime("master_timeout", clusterHealthRequest.masterNodeTimeout()));
@@ -73,11 +74,19 @@ public class RestClusterHealthAction extends BaseRestHandler {
         if (request.param("wait_for_events") != null) {
             clusterHealthRequest.waitForEvents(Priority.valueOf(request.param("wait_for_events").toUpperCase(Locale.ROOT)));
         }
-        client.admin().cluster().health(clusterHealthRequest, new RestStatusToXContentListener<ClusterHealthResponse>(channel));
+        return channel -> client.admin().cluster().health(clusterHealthRequest, new RestStatusToXContentListener<>(channel));
+    }
+
+    private static final Set<String> RESPONSE_PARAMS = Collections.singleton("level");
+
+    @Override
+    protected Set<String> responseParams() {
+        return RESPONSE_PARAMS;
     }
 
     @Override
     public boolean canTripCircuitBreaker() {
         return false;
     }
+
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestClusterRerouteAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestClusterRerouteAction.java
index 1a1e78b172..bd86f41ac2 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestClusterRerouteAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestClusterRerouteAction.java
@@ -21,8 +21,8 @@ package org.elasticsearch.rest.action.admin.cluster;
 
 import org.elasticsearch.action.admin.cluster.reroute.ClusterRerouteRequest;
 import org.elasticsearch.action.admin.cluster.reroute.ClusterRerouteResponse;
-import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.client.Requests;
+import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.routing.allocation.command.AllocationCommandRegistry;
 import org.elasticsearch.cluster.routing.allocation.command.AllocationCommands;
@@ -40,13 +40,15 @@ import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.AcknowledgedRestListener;
 
 import java.io.IOException;
+import java.util.Collections;
 import java.util.EnumSet;
+import java.util.HashSet;
+import java.util.Set;
 
 /**
  */
@@ -74,29 +76,47 @@ public class RestClusterRerouteAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) throws Exception {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         ClusterRerouteRequest clusterRerouteRequest = createRequest(request, registry, parseFieldMatcher);
-        client.admin().cluster().reroute(clusterRerouteRequest, new AcknowledgedRestListener<ClusterRerouteResponse>(channel) {
-            @Override
-            protected void addCustomFields(XContentBuilder builder, ClusterRerouteResponse response) throws IOException {
-                builder.startObject("state");
-                // by default, return everything but metadata
-                if (request.param("metric") == null) {
-                    request.params().put("metric", DEFAULT_METRICS);
-                }
-                settingsFilter.addFilterSettingParams(request);
-                response.getState().toXContent(builder, request);
-                builder.endObject();
-                if (clusterRerouteRequest.explain()) {
-                    assert response.getExplanations() != null;
-                    response.getExplanations().toXContent(builder, ToXContent.EMPTY_PARAMS);
+
+        // by default, return everything but metadata
+        final String metric = request.param("metric");
+        if (metric == null) {
+            request.params().put("metric", DEFAULT_METRICS);
+        }
+
+        return channel ->
+            client.admin().cluster().reroute(clusterRerouteRequest, new AcknowledgedRestListener<ClusterRerouteResponse>(channel) {
+                @Override
+                protected void addCustomFields(XContentBuilder builder, ClusterRerouteResponse response) throws IOException {
+                    builder.startObject("state");
+                    settingsFilter.addFilterSettingParams(request);
+                    response.getState().toXContent(builder, request);
+                    builder.endObject();
+                    if (clusterRerouteRequest.explain()) {
+                        assert response.getExplanations() != null;
+                        response.getExplanations().toXContent(builder, ToXContent.EMPTY_PARAMS);
+                    }
                 }
-            }
         });
     }
 
+    private static final Set<String> RESPONSE_PARAMS;
+
+    static {
+        final Set<String> responseParams = new HashSet<>();
+        responseParams.add("metric");
+        responseParams.addAll(Settings.FORMAT_PARAMS);
+        RESPONSE_PARAMS = Collections.unmodifiableSet(responseParams);
+    }
+
+    @Override
+    protected Set<String> responseParams() {
+        return RESPONSE_PARAMS;
+    }
+
     public static ClusterRerouteRequest createRequest(RestRequest request, AllocationCommandRegistry registry,
-            ParseFieldMatcher parseFieldMatcher) throws IOException {
+                                                      ParseFieldMatcher parseFieldMatcher) throws IOException {
         ClusterRerouteRequest clusterRerouteRequest = Requests.clusterRerouteRequest();
         clusterRerouteRequest.dryRun(request.paramAsBoolean("dry_run", clusterRerouteRequest.dryRun()));
         clusterRerouteRequest.explain(request.paramAsBoolean("explain", clusterRerouteRequest.explain()));
@@ -125,4 +145,5 @@ public class RestClusterRerouteAction extends BaseRestHandler {
             return parseFieldMatcher;
         }
     }
+
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestClusterSearchShardsAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestClusterSearchShardsAction.java
index 754b9b0d63..459ccf5fa4 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestClusterSearchShardsAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestClusterSearchShardsAction.java
@@ -20,24 +20,22 @@
 package org.elasticsearch.rest.action.admin.cluster;
 
 import org.elasticsearch.action.admin.cluster.shards.ClusterSearchShardsRequest;
-import org.elasticsearch.action.admin.cluster.shards.ClusterSearchShardsResponse;
 import org.elasticsearch.action.support.IndicesOptions;
-import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.client.Requests;
+import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.RestToXContentListener;
 
+import java.io.IOException;
+
 import static org.elasticsearch.rest.RestRequest.Method.GET;
 import static org.elasticsearch.rest.RestRequest.Method.POST;
 
-/**
- */
 public class RestClusterSearchShardsAction extends BaseRestHandler {
 
     @Inject
@@ -52,7 +50,7 @@ public class RestClusterSearchShardsAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         String[] indices = Strings.splitStringByCommaToArray(request.param("index"));
         final ClusterSearchShardsRequest clusterSearchShardsRequest = Requests.clusterSearchShardsRequest(indices);
         clusterSearchShardsRequest.local(request.paramAsBoolean("local", clusterSearchShardsRequest.local()));
@@ -62,6 +60,6 @@ public class RestClusterSearchShardsAction extends BaseRestHandler {
         clusterSearchShardsRequest.preference(request.param("preference"));
         clusterSearchShardsRequest.indicesOptions(IndicesOptions.fromRequest(request, clusterSearchShardsRequest.indicesOptions()));
 
-        client.admin().cluster().searchShards(clusterSearchShardsRequest, new RestToXContentListener<ClusterSearchShardsResponse>(channel));
+        return channel -> client.admin().cluster().searchShards(clusterSearchShardsRequest, new RestToXContentListener<>(channel));
     }
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestClusterStateAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestClusterStateAction.java
index fab2ee0062..54a55a0ba0 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestClusterStateAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestClusterStateAction.java
@@ -22,8 +22,8 @@ package org.elasticsearch.rest.action.admin.cluster;
 import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
 import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
 import org.elasticsearch.action.support.IndicesOptions;
-import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.client.Requests;
+import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.inject.Inject;
@@ -32,14 +32,17 @@ import org.elasticsearch.common.settings.SettingsFilter;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
 import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.rest.action.RestBuilderListener;
 
+import java.io.IOException;
+import java.util.Collections;
 import java.util.EnumSet;
+import java.util.HashSet;
+import java.util.Set;
 
 public class RestClusterStateAction extends BaseRestHandler {
 
@@ -56,7 +59,7 @@ public class RestClusterStateAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         final ClusterStateRequest clusterStateRequest = Requests.clusterStateRequest();
         clusterStateRequest.indicesOptions(IndicesOptions.fromRequest(request, clusterStateRequest.indicesOptions()));
         clusterStateRequest.local(request.paramAsBoolean("local", clusterStateRequest.local()));
@@ -84,7 +87,7 @@ public class RestClusterStateAction extends BaseRestHandler {
         }
         settingsFilter.addFilterSettingParams(request);
 
-        client.admin().cluster().state(clusterStateRequest, new RestBuilderListener<ClusterStateResponse>(channel) {
+        return channel -> client.admin().cluster().state(clusterStateRequest, new RestBuilderListener<ClusterStateResponse>(channel) {
             @Override
             public RestResponse buildResponse(ClusterStateResponse response, XContentBuilder builder) throws Exception {
                 builder.startObject();
@@ -96,6 +99,20 @@ public class RestClusterStateAction extends BaseRestHandler {
         });
     }
 
+    private static final Set<String> RESPONSE_PARAMS;
+
+    static {
+        final Set<String> responseParams = new HashSet<>();
+        responseParams.add("metric");
+        responseParams.addAll(Settings.FORMAT_PARAMS);
+        RESPONSE_PARAMS = Collections.unmodifiableSet(responseParams);
+    }
+
+    @Override
+    protected Set<String> responseParams() {
+        return RESPONSE_PARAMS;
+    }
+
     @Override
     public boolean canTripCircuitBreaker() {
         return false;
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestClusterStatsAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestClusterStatsAction.java
index 7ef05d0455..ad0ead8b28 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestClusterStatsAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestClusterStatsAction.java
@@ -24,11 +24,12 @@ import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.RestActions.NodesResponseRestListener;
 
+import java.io.IOException;
+
 /**
  *
  */
@@ -42,10 +43,10 @@ public class RestClusterStatsAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         ClusterStatsRequest clusterStatsRequest = new ClusterStatsRequest().nodesIds(request.paramAsStringArray("nodeId", null));
         clusterStatsRequest.timeout(request.param("timeout"));
-        client.admin().cluster().clusterStats(clusterStatsRequest, new NodesResponseRestListener<>(channel));
+        return channel -> client.admin().cluster().clusterStats(clusterStatsRequest, new NodesResponseRestListener<>(channel));
     }
 
     @Override
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestClusterUpdateSettingsAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestClusterUpdateSettingsAction.java
index 8de725dbe7..b041ae4c3a 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestClusterUpdateSettingsAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestClusterUpdateSettingsAction.java
@@ -21,21 +21,21 @@ package org.elasticsearch.rest.action.admin.cluster;
 
 import org.elasticsearch.action.admin.cluster.settings.ClusterUpdateSettingsRequest;
 import org.elasticsearch.action.admin.cluster.settings.ClusterUpdateSettingsResponse;
-import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.client.Requests;
+import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentFactory;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.AcknowledgedRestListener;
 
 import java.io.IOException;
 import java.util.Map;
+import java.util.Set;
 
 public class RestClusterUpdateSettingsAction extends BaseRestHandler {
 
@@ -46,7 +46,7 @@ public class RestClusterUpdateSettingsAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) throws Exception {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         final ClusterUpdateSettingsRequest clusterUpdateSettingsRequest = Requests.clusterUpdateSettingsRequest();
         clusterUpdateSettingsRequest.timeout(request.paramAsTime("timeout", clusterUpdateSettingsRequest.timeout()));
         clusterUpdateSettingsRequest.masterNodeTimeout(
@@ -62,7 +62,7 @@ public class RestClusterUpdateSettingsAction extends BaseRestHandler {
             clusterUpdateSettingsRequest.persistentSettings((Map) source.get("persistent"));
         }
 
-        client.admin().cluster().updateSettings(clusterUpdateSettingsRequest,
+        return channel -> client.admin().cluster().updateSettings(clusterUpdateSettingsRequest,
                 new AcknowledgedRestListener<ClusterUpdateSettingsResponse>(channel) {
                     @Override
                     protected void addCustomFields(XContentBuilder builder, ClusterUpdateSettingsResponse response) throws IOException {
@@ -78,6 +78,11 @@ public class RestClusterUpdateSettingsAction extends BaseRestHandler {
     }
 
     @Override
+    protected Set<String> responseParams() {
+        return Settings.FORMAT_PARAMS;
+    }
+
+    @Override
     public boolean canTripCircuitBreaker() {
         return false;
     }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestCreateSnapshotAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestCreateSnapshotAction.java
index 96449131a6..84ca88d292 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestCreateSnapshotAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestCreateSnapshotAction.java
@@ -20,16 +20,16 @@
 package org.elasticsearch.rest.action.admin.cluster;
 
 import org.elasticsearch.action.admin.cluster.snapshots.create.CreateSnapshotRequest;
-import org.elasticsearch.action.admin.cluster.snapshots.create.CreateSnapshotResponse;
 import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.RestToXContentListener;
 
+import java.io.IOException;
+
 import static org.elasticsearch.client.Requests.createSnapshotRequest;
 import static org.elasticsearch.rest.RestRequest.Method.POST;
 import static org.elasticsearch.rest.RestRequest.Method.PUT;
@@ -47,11 +47,11 @@ public class RestCreateSnapshotAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         CreateSnapshotRequest createSnapshotRequest = createSnapshotRequest(request.param("repository"), request.param("snapshot"));
         createSnapshotRequest.source(request.content().utf8ToString());
         createSnapshotRequest.masterNodeTimeout(request.paramAsTime("master_timeout", createSnapshotRequest.masterNodeTimeout()));
         createSnapshotRequest.waitForCompletion(request.paramAsBoolean("wait_for_completion", false));
-        client.admin().cluster().createSnapshot(createSnapshotRequest, new RestToXContentListener<CreateSnapshotResponse>(channel));
+        return channel -> client.admin().cluster().createSnapshot(createSnapshotRequest, new RestToXContentListener<>(channel));
     }
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestDeleteRepositoryAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestDeleteRepositoryAction.java
index 78d063bae0..780368dfd3 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestDeleteRepositoryAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestDeleteRepositoryAction.java
@@ -20,16 +20,16 @@
 package org.elasticsearch.rest.action.admin.cluster;
 
 import org.elasticsearch.action.admin.cluster.repositories.delete.DeleteRepositoryRequest;
-import org.elasticsearch.action.admin.cluster.repositories.delete.DeleteRepositoryResponse;
 import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.AcknowledgedRestListener;
 
+import java.io.IOException;
+
 import static org.elasticsearch.client.Requests.deleteRepositoryRequest;
 import static org.elasticsearch.rest.RestRequest.Method.DELETE;
 
@@ -45,11 +45,11 @@ public class RestDeleteRepositoryAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         DeleteRepositoryRequest deleteRepositoryRequest = deleteRepositoryRequest(request.param("repository"));
         deleteRepositoryRequest.masterNodeTimeout(request.paramAsTime("master_timeout", deleteRepositoryRequest.masterNodeTimeout()));
         deleteRepositoryRequest.timeout(request.paramAsTime("timeout", deleteRepositoryRequest.timeout()));
         deleteRepositoryRequest.masterNodeTimeout(request.paramAsTime("master_timeout", deleteRepositoryRequest.masterNodeTimeout()));
-        client.admin().cluster().deleteRepository(deleteRepositoryRequest, new AcknowledgedRestListener<DeleteRepositoryResponse>(channel));
+        return channel -> client.admin().cluster().deleteRepository(deleteRepositoryRequest, new AcknowledgedRestListener<>(channel));
     }
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestDeleteSnapshotAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestDeleteSnapshotAction.java
index d001a1e90e..29ad89acda 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestDeleteSnapshotAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestDeleteSnapshotAction.java
@@ -20,16 +20,16 @@
 package org.elasticsearch.rest.action.admin.cluster;
 
 import org.elasticsearch.action.admin.cluster.snapshots.delete.DeleteSnapshotRequest;
-import org.elasticsearch.action.admin.cluster.snapshots.delete.DeleteSnapshotResponse;
 import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.AcknowledgedRestListener;
 
+import java.io.IOException;
+
 import static org.elasticsearch.client.Requests.deleteSnapshotRequest;
 import static org.elasticsearch.rest.RestRequest.Method.DELETE;
 
@@ -45,9 +45,9 @@ public class RestDeleteSnapshotAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         DeleteSnapshotRequest deleteSnapshotRequest = deleteSnapshotRequest(request.param("repository"), request.param("snapshot"));
         deleteSnapshotRequest.masterNodeTimeout(request.paramAsTime("master_timeout", deleteSnapshotRequest.masterNodeTimeout()));
-        client.admin().cluster().deleteSnapshot(deleteSnapshotRequest, new AcknowledgedRestListener<DeleteSnapshotResponse>(channel));
+        return channel -> client.admin().cluster().deleteSnapshot(deleteSnapshotRequest, new AcknowledgedRestListener<>(channel));
     }
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestDeleteStoredScriptAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestDeleteStoredScriptAction.java
index 212b42135e..31a0c70b67 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestDeleteStoredScriptAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestDeleteStoredScriptAction.java
@@ -23,11 +23,12 @@ import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.AcknowledgedRestListener;
 
+import java.io.IOException;
+
 import static org.elasticsearch.rest.RestRequest.Method.DELETE;
 
 public class RestDeleteStoredScriptAction extends BaseRestHandler {
@@ -49,9 +50,9 @@ public class RestDeleteStoredScriptAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, NodeClient client) throws IOException {
         DeleteStoredScriptRequest deleteStoredScriptRequest = new DeleteStoredScriptRequest(getScriptLang(request), request.param("id"));
-        client.admin().cluster().deleteStoredScript(deleteStoredScriptRequest, new AcknowledgedRestListener<>(channel));
+        return channel -> client.admin().cluster().deleteStoredScript(deleteStoredScriptRequest, new AcknowledgedRestListener<>(channel));
     }
 
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestGetRepositoriesAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestGetRepositoriesAction.java
index 802af3cb5b..6e5d06f374 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestGetRepositoriesAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestGetRepositoriesAction.java
@@ -31,12 +31,14 @@ import org.elasticsearch.common.settings.SettingsFilter;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
 import org.elasticsearch.rest.action.RestBuilderListener;
 
+import java.io.IOException;
+import java.util.Set;
+
 import static org.elasticsearch.client.Requests.getRepositoryRequest;
 import static org.elasticsearch.rest.RestRequest.Method.GET;
 import static org.elasticsearch.rest.RestStatus.OK;
@@ -57,23 +59,30 @@ public class RestGetRepositoriesAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         final String[] repositories = request.paramAsStringArray("repository", Strings.EMPTY_ARRAY);
         GetRepositoriesRequest getRepositoriesRequest = getRepositoryRequest(repositories);
         getRepositoriesRequest.masterNodeTimeout(request.paramAsTime("master_timeout", getRepositoriesRequest.masterNodeTimeout()));
         getRepositoriesRequest.local(request.paramAsBoolean("local", getRepositoriesRequest.local()));
         settingsFilter.addFilterSettingParams(request);
-        client.admin().cluster().getRepositories(getRepositoriesRequest, new RestBuilderListener<GetRepositoriesResponse>(channel) {
-            @Override
-            public RestResponse buildResponse(GetRepositoriesResponse response, XContentBuilder builder) throws Exception {
-                builder.startObject();
-                for (RepositoryMetaData repositoryMetaData : response.repositories()) {
-                    RepositoriesMetaData.toXContent(repositoryMetaData, builder, request);
-                }
-                builder.endObject();
+        return channel ->
+                client.admin().cluster().getRepositories(getRepositoriesRequest, new RestBuilderListener<GetRepositoriesResponse>(channel) {
+                    @Override
+                    public RestResponse buildResponse(GetRepositoriesResponse response, XContentBuilder builder) throws Exception {
+                        builder.startObject();
+                        for (RepositoryMetaData repositoryMetaData : response.repositories()) {
+                            RepositoriesMetaData.toXContent(repositoryMetaData, builder, request);
+                        }
+                        builder.endObject();
 
-                return new BytesRestResponse(OK, builder);
-            }
+                        return new BytesRestResponse(OK, builder);
+                    }
         });
     }
+
+    @Override
+    protected Set<String> responseParams() {
+        return Settings.FORMAT_PARAMS;
+    }
+
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestGetSnapshotsAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestGetSnapshotsAction.java
index 9e10a87bc0..515ce2d68d 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestGetSnapshotsAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestGetSnapshotsAction.java
@@ -20,17 +20,17 @@
 package org.elasticsearch.rest.action.admin.cluster;
 
 import org.elasticsearch.action.admin.cluster.snapshots.get.GetSnapshotsRequest;
-import org.elasticsearch.action.admin.cluster.snapshots.get.GetSnapshotsResponse;
 import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.RestToXContentListener;
 
+import java.io.IOException;
+
 import static org.elasticsearch.client.Requests.getSnapshotsRequest;
 import static org.elasticsearch.rest.RestRequest.Method.GET;
 
@@ -47,7 +47,7 @@ public class RestGetSnapshotsAction extends BaseRestHandler {
 
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         String repository = request.param("repository");
         String[] snapshots = request.paramAsStringArray("snapshot", Strings.EMPTY_ARRAY);
 
@@ -55,6 +55,6 @@ public class RestGetSnapshotsAction extends BaseRestHandler {
         getSnapshotsRequest.ignoreUnavailable(request.paramAsBoolean("ignore_unavailable", getSnapshotsRequest.ignoreUnavailable()));
 
         getSnapshotsRequest.masterNodeTimeout(request.paramAsTime("master_timeout", getSnapshotsRequest.masterNodeTimeout()));
-        client.admin().cluster().getSnapshots(getSnapshotsRequest, new RestToXContentListener<GetSnapshotsResponse>(channel));
+        return channel -> client.admin().cluster().getSnapshots(getSnapshotsRequest, new RestToXContentListener<>(channel));
     }
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestGetStoredScriptAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestGetStoredScriptAction.java
index 1185685c49..a2d11e75a6 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestGetStoredScriptAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestGetStoredScriptAction.java
@@ -26,13 +26,14 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
 import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.rest.action.RestBuilderListener;
 
+import java.io.IOException;
+
 import static org.elasticsearch.rest.RestRequest.Method.GET;
 
 public class RestGetStoredScriptAction extends BaseRestHandler {
@@ -58,9 +59,9 @@ public class RestGetStoredScriptAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, NodeClient client) throws IOException {
         final GetStoredScriptRequest getRequest = new GetStoredScriptRequest(getScriptLang(request), request.param("id"));
-        client.admin().cluster().getStoredScript(getRequest, new RestBuilderListener<GetStoredScriptResponse>(channel) {
+        return channel -> client.admin().cluster().getStoredScript(getRequest, new RestBuilderListener<GetStoredScriptResponse>(channel) {
             @Override
             public RestResponse buildResponse(GetStoredScriptResponse response, XContentBuilder builder) throws Exception {
                 builder.startObject();
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestGetTaskAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestGetTaskAction.java
index f1edf67201..df0945d99b 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestGetTaskAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestGetTaskAction.java
@@ -25,12 +25,13 @@ import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.RestToXContentListener;
 import org.elasticsearch.tasks.TaskId;
 
+import java.io.IOException;
+
 import static org.elasticsearch.rest.RestRequest.Method.GET;
 
 public class RestGetTaskAction extends BaseRestHandler {
@@ -41,7 +42,7 @@ public class RestGetTaskAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         TaskId taskId = new TaskId(request.param("taskId"));
         boolean waitForCompletion = request.paramAsBoolean("wait_for_completion", false);
         TimeValue timeout = request.paramAsTime("timeout", null);
@@ -50,6 +51,6 @@ public class RestGetTaskAction extends BaseRestHandler {
         getTaskRequest.setTaskId(taskId);
         getTaskRequest.setWaitForCompletion(waitForCompletion);
         getTaskRequest.setTimeout(timeout);
-        client.admin().cluster().getTask(getTaskRequest, new RestToXContentListener<>(channel));
+        return channel -> client.admin().cluster().getTask(getTaskRequest, new RestToXContentListener<>(channel));
     }
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestListTasksAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestListTasksAction.java
index d5ff427e3d..66a3c9cf6d 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestListTasksAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestListTasksAction.java
@@ -40,10 +40,13 @@ import org.elasticsearch.rest.action.RestBuilderListener;
 import org.elasticsearch.rest.action.RestToXContentListener;
 import org.elasticsearch.tasks.TaskId;
 
+import java.io.IOException;
+
 import static org.elasticsearch.rest.RestRequest.Method.GET;
 
 
 public class RestListTasksAction extends BaseRestHandler {
+
     private final ClusterService clusterService;
 
     @Inject
@@ -53,16 +56,23 @@ public class RestListTasksAction extends BaseRestHandler {
         controller.registerHandler(GET, "/_tasks", this);
     }
 
+    @Override
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
+        final ListTasksRequest listTasksRequest = generateListTasksRequest(request);
+        final String groupBy = request.param("group_by", "nodes");
+        return channel -> client.admin().cluster().listTasks(listTasksRequest, listTasksResponseListener(clusterService, groupBy, channel));
+    }
+
     public static ListTasksRequest generateListTasksRequest(RestRequest request) {
         boolean detailed = request.paramAsBoolean("detailed", false);
-        String[] nodesIds = Strings.splitStringByCommaToArray(request.param("node_id"));
+        String[] nodes = Strings.splitStringByCommaToArray(request.param("nodes"));
         String[] actions = Strings.splitStringByCommaToArray(request.param("actions"));
         TaskId parentTaskId = new TaskId(request.param("parent_task_id"));
         boolean waitForCompletion = request.paramAsBoolean("wait_for_completion", false);
         TimeValue timeout = request.paramAsTime("timeout", null);
 
         ListTasksRequest listTasksRequest = new ListTasksRequest();
-        listTasksRequest.setNodesIds(nodesIds);
+        listTasksRequest.setNodes(nodes);
         listTasksRequest.setDetailed(detailed);
         listTasksRequest.setActions(actions);
         listTasksRequest.setParentTaskId(parentTaskId);
@@ -71,17 +81,13 @@ public class RestListTasksAction extends BaseRestHandler {
         return listTasksRequest;
     }
 
-    @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
-        client.admin().cluster().listTasks(generateListTasksRequest(request), listTasksResponseListener(clusterService, channel));
-    }
-
     /**
      * Standard listener for extensions of {@link ListTasksResponse} that supports {@code group_by=nodes}.
      */
-    public static <T extends ListTasksResponse> ActionListener<T> listTasksResponseListener(ClusterService clusterService,
-            RestChannel channel) {
-        String groupBy = channel.request().param("group_by", "nodes");
+    public static <T extends ListTasksResponse> ActionListener<T> listTasksResponseListener(
+        ClusterService clusterService,
+        String groupBy,
+        final RestChannel channel) {
         if ("nodes".equals(groupBy)) {
             return new RestBuilderListener<T>(channel) {
                 @Override
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestNodesHotThreadsAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestNodesHotThreadsAction.java
index 87af57276c..363a89111f 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestNodesHotThreadsAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestNodesHotThreadsAction.java
@@ -29,13 +29,14 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
 import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.rest.action.RestResponseListener;
 
+import java.io.IOException;
+
 
 /**
  */
@@ -56,7 +57,7 @@ public class RestNodesHotThreadsAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         String[] nodesIds = Strings.splitStringByCommaToArray(request.param("nodeId"));
         NodesHotThreadsRequest nodesHotThreadsRequest = new NodesHotThreadsRequest(nodesIds);
         nodesHotThreadsRequest.threads(request.paramAsInt("threads", nodesHotThreadsRequest.threads()));
@@ -65,18 +66,20 @@ public class RestNodesHotThreadsAction extends BaseRestHandler {
         nodesHotThreadsRequest.interval(TimeValue.parseTimeValue(request.param("interval"), nodesHotThreadsRequest.interval(), "interval"));
         nodesHotThreadsRequest.snapshots(request.paramAsInt("snapshots", nodesHotThreadsRequest.snapshots()));
         nodesHotThreadsRequest.timeout(request.param("timeout"));
-        client.admin().cluster().nodesHotThreads(nodesHotThreadsRequest, new RestResponseListener<NodesHotThreadsResponse>(channel) {
-            @Override
-            public RestResponse buildResponse(NodesHotThreadsResponse response) throws Exception {
-                StringBuilder sb = new StringBuilder();
-                for (NodeHotThreads node : response.getNodes()) {
-                    sb.append("::: ").append(node.getNode().toString()).append("\n");
-                    Strings.spaceify(3, node.getHotThreads(), sb);
-                    sb.append('\n');
-                }
-                return new BytesRestResponse(RestStatus.OK, sb.toString());
-            }
-        });
+        return channel -> client.admin().cluster().nodesHotThreads(
+                nodesHotThreadsRequest,
+                new RestResponseListener<NodesHotThreadsResponse>(channel) {
+                    @Override
+                    public RestResponse buildResponse(NodesHotThreadsResponse response) throws Exception {
+                        StringBuilder sb = new StringBuilder();
+                        for (NodeHotThreads node : response.getNodes()) {
+                            sb.append("::: ").append(node.getNode().toString()).append("\n");
+                            Strings.spaceify(3, node.getHotThreads(), sb);
+                            sb.append('\n');
+                        }
+                        return new BytesRestResponse(RestStatus.OK, sb.toString());
+                    }
+                });
     }
 
     @Override
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestNodesInfoAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestNodesInfoAction.java
index c45709d07d..40cfc6372a 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestNodesInfoAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestNodesInfoAction.java
@@ -27,11 +27,11 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.settings.SettingsFilter;
 import org.elasticsearch.common.util.set.Sets;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.RestActions.NodesResponseRestListener;
 
+import java.io.IOException;
 import java.util.Set;
 
 import static org.elasticsearch.rest.RestRequest.Method.GET;
@@ -65,7 +65,7 @@ public class RestNodesInfoAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         String[] nodeIds;
         Set<String> metrics;
 
@@ -108,7 +108,12 @@ public class RestNodesInfoAction extends BaseRestHandler {
 
         settingsFilter.addFilterSettingParams(request);
 
-        client.admin().cluster().nodesInfo(nodesInfoRequest, new NodesResponseRestListener<>(channel));
+        return channel -> client.admin().cluster().nodesInfo(nodesInfoRequest, new NodesResponseRestListener<>(channel));
+    }
+
+    @Override
+    protected Set<String> responseParams() {
+        return Settings.FORMAT_PARAMS;
     }
 
     @Override
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestNodesStatsAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestNodesStatsAction.java
index be6847f1b5..6b353fc6e4 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestNodesStatsAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestNodesStatsAction.java
@@ -27,11 +27,11 @@ import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.RestActions.NodesResponseRestListener;
 
+import java.io.IOException;
 import java.util.Set;
 
 import static org.elasticsearch.rest.RestRequest.Method.GET;
@@ -53,7 +53,7 @@ public class RestNodesStatsAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         String[] nodesIds = Strings.splitStringByCommaToArray(request.param("nodeId"));
         Set<String> metrics = Strings.splitStringByCommaToSet(request.param("metric", "_all"));
 
@@ -111,7 +111,7 @@ public class RestNodesStatsAction extends BaseRestHandler {
             nodesStatsRequest.indices().includeSegmentFileSizes(true);
         }
 
-        client.admin().cluster().nodesStats(nodesStatsRequest, new NodesResponseRestListener<>(channel));
+        return channel -> client.admin().cluster().nodesStats(nodesStatsRequest, new NodesResponseRestListener<>(channel));
     }
 
     @Override
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestPendingClusterTasksAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestPendingClusterTasksAction.java
index d1cb65092c..e3441d00d0 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestPendingClusterTasksAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestPendingClusterTasksAction.java
@@ -24,11 +24,12 @@ import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.RestToXContentListener;
 
+import java.io.IOException;
+
 public class RestPendingClusterTasksAction extends BaseRestHandler {
 
     @Inject
@@ -38,10 +39,10 @@ public class RestPendingClusterTasksAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         PendingClusterTasksRequest pendingClusterTasksRequest = new PendingClusterTasksRequest();
         pendingClusterTasksRequest.masterNodeTimeout(request.paramAsTime("master_timeout", pendingClusterTasksRequest.masterNodeTimeout()));
         pendingClusterTasksRequest.local(request.paramAsBoolean("local", pendingClusterTasksRequest.local()));
-        client.admin().cluster().pendingClusterTasks(pendingClusterTasksRequest, new RestToXContentListener<>(channel));
+        return channel -> client.admin().cluster().pendingClusterTasks(pendingClusterTasksRequest, new RestToXContentListener<>(channel));
     }
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestPutRepositoryAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestPutRepositoryAction.java
index 002e1bfdc9..c142230925 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestPutRepositoryAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestPutRepositoryAction.java
@@ -20,16 +20,16 @@
 package org.elasticsearch.rest.action.admin.cluster;
 
 import org.elasticsearch.action.admin.cluster.repositories.put.PutRepositoryRequest;
-import org.elasticsearch.action.admin.cluster.repositories.put.PutRepositoryResponse;
 import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.AcknowledgedRestListener;
 
+import java.io.IOException;
+
 import static org.elasticsearch.client.Requests.putRepositoryRequest;
 import static org.elasticsearch.rest.RestRequest.Method.POST;
 import static org.elasticsearch.rest.RestRequest.Method.PUT;
@@ -48,12 +48,12 @@ public class RestPutRepositoryAction extends BaseRestHandler {
 
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         PutRepositoryRequest putRepositoryRequest = putRepositoryRequest(request.param("repository"));
         putRepositoryRequest.source(request.content().utf8ToString());
         putRepositoryRequest.verify(request.paramAsBoolean("verify", true));
         putRepositoryRequest.masterNodeTimeout(request.paramAsTime("master_timeout", putRepositoryRequest.masterNodeTimeout()));
         putRepositoryRequest.timeout(request.paramAsTime("timeout", putRepositoryRequest.timeout()));
-        client.admin().cluster().putRepository(putRepositoryRequest, new AcknowledgedRestListener<PutRepositoryResponse>(channel));
+        return channel -> client.admin().cluster().putRepository(putRepositoryRequest, new AcknowledgedRestListener<>(channel));
     }
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestPutStoredScriptAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestPutStoredScriptAction.java
index c5156c4cd0..aec998e6f9 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestPutStoredScriptAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestPutStoredScriptAction.java
@@ -23,11 +23,12 @@ import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.AcknowledgedRestListener;
 
+import java.io.IOException;
+
 import static org.elasticsearch.rest.RestRequest.Method.POST;
 import static org.elasticsearch.rest.RestRequest.Method.PUT;
 
@@ -51,9 +52,9 @@ public class RestPutStoredScriptAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, NodeClient client) throws IOException {
         PutStoredScriptRequest putRequest = new PutStoredScriptRequest(getScriptLang(request), request.param("id"));
         putRequest.script(request.content());
-        client.admin().cluster().putStoredScript(putRequest, new AcknowledgedRestListener<>(channel));
+        return channel -> client.admin().cluster().putStoredScript(putRequest, new AcknowledgedRestListener<>(channel));
     }
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestRestoreSnapshotAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestRestoreSnapshotAction.java
index 100866e02d..f94990e1ae 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestRestoreSnapshotAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestRestoreSnapshotAction.java
@@ -20,16 +20,16 @@
 package org.elasticsearch.rest.action.admin.cluster;
 
 import org.elasticsearch.action.admin.cluster.snapshots.restore.RestoreSnapshotRequest;
-import org.elasticsearch.action.admin.cluster.snapshots.restore.RestoreSnapshotResponse;
 import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.RestToXContentListener;
 
+import java.io.IOException;
+
 import static org.elasticsearch.client.Requests.restoreSnapshotRequest;
 import static org.elasticsearch.rest.RestRequest.Method.POST;
 
@@ -45,11 +45,11 @@ public class RestRestoreSnapshotAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         RestoreSnapshotRequest restoreSnapshotRequest = restoreSnapshotRequest(request.param("repository"), request.param("snapshot"));
         restoreSnapshotRequest.masterNodeTimeout(request.paramAsTime("master_timeout", restoreSnapshotRequest.masterNodeTimeout()));
         restoreSnapshotRequest.waitForCompletion(request.paramAsBoolean("wait_for_completion", false));
         restoreSnapshotRequest.source(request.content().utf8ToString());
-        client.admin().cluster().restoreSnapshot(restoreSnapshotRequest, new RestToXContentListener<RestoreSnapshotResponse>(channel));
+        return channel -> client.admin().cluster().restoreSnapshot(restoreSnapshotRequest, new RestToXContentListener<>(channel));
     }
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestSnapshotsStatusAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestSnapshotsStatusAction.java
index 4333dfc027..b517e32a3d 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestSnapshotsStatusAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestSnapshotsStatusAction.java
@@ -25,11 +25,12 @@ import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.RestToXContentListener;
 
+import java.io.IOException;
+
 import static org.elasticsearch.client.Requests.snapshotsStatusRequest;
 import static org.elasticsearch.rest.RestRequest.Method.GET;
 
@@ -47,7 +48,7 @@ public class RestSnapshotsStatusAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         String repository = request.param("repository", "_all");
         String[] snapshots = request.paramAsStringArray("snapshot", Strings.EMPTY_ARRAY);
         if (snapshots.length == 1 && "_all".equalsIgnoreCase(snapshots[0])) {
@@ -57,6 +58,6 @@ public class RestSnapshotsStatusAction extends BaseRestHandler {
         snapshotsStatusRequest.ignoreUnavailable(request.paramAsBoolean("ignore_unavailable", snapshotsStatusRequest.ignoreUnavailable()));
 
         snapshotsStatusRequest.masterNodeTimeout(request.paramAsTime("master_timeout", snapshotsStatusRequest.masterNodeTimeout()));
-        client.admin().cluster().snapshotsStatus(snapshotsStatusRequest, new RestToXContentListener<>(channel));
+        return channel -> client.admin().cluster().snapshotsStatus(snapshotsStatusRequest, new RestToXContentListener<>(channel));
     }
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestVerifyRepositoryAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestVerifyRepositoryAction.java
index 85aac84077..8442752446 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestVerifyRepositoryAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestVerifyRepositoryAction.java
@@ -24,11 +24,12 @@ import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.RestToXContentListener;
 
+import java.io.IOException;
+
 import static org.elasticsearch.client.Requests.verifyRepositoryRequest;
 import static org.elasticsearch.rest.RestRequest.Method.POST;
 
@@ -41,10 +42,10 @@ public class RestVerifyRepositoryAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         VerifyRepositoryRequest verifyRepositoryRequest = verifyRepositoryRequest(request.param("repository"));
         verifyRepositoryRequest.masterNodeTimeout(request.paramAsTime("master_timeout", verifyRepositoryRequest.masterNodeTimeout()));
         verifyRepositoryRequest.timeout(request.paramAsTime("timeout", verifyRepositoryRequest.timeout()));
-        client.admin().cluster().verifyRepository(verifyRepositoryRequest, new RestToXContentListener<>(channel));
+        return channel -> client.admin().cluster().verifyRepository(verifyRepositoryRequest, new RestToXContentListener<>(channel));
     }
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestAliasesExistAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestAliasesExistAction.java
index dbb8ddde9d..59aafde593 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestAliasesExistAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestAliasesExistAction.java
@@ -31,16 +31,15 @@ import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 
+import java.io.IOException;
+
 import static org.elasticsearch.rest.RestRequest.Method.HEAD;
 import static org.elasticsearch.rest.RestStatus.NOT_FOUND;
 import static org.elasticsearch.rest.RestStatus.OK;
 
-/**
- */
 public class RestAliasesExistAction extends BaseRestHandler {
 
     @Inject
@@ -52,7 +51,7 @@ public class RestAliasesExistAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         String[] aliases = request.paramAsStringArray("name", Strings.EMPTY_ARRAY);
         final String[] indices = Strings.splitStringByCommaToArray(request.param("index"));
         GetAliasesRequest getAliasesRequest = new GetAliasesRequest(aliases);
@@ -60,7 +59,7 @@ public class RestAliasesExistAction extends BaseRestHandler {
         getAliasesRequest.indicesOptions(IndicesOptions.fromRequest(request, getAliasesRequest.indicesOptions()));
         getAliasesRequest.local(request.paramAsBoolean("local", getAliasesRequest.local()));
 
-        client.admin().indices().aliasesExist(getAliasesRequest, new ActionListener<AliasesExistResponse>() {
+        return channel -> client.admin().indices().aliasesExist(getAliasesRequest, new ActionListener<AliasesExistResponse>() {
 
             @Override
             public void onResponse(AliasesExistResponse response) {
@@ -85,6 +84,7 @@ public class RestAliasesExistAction extends BaseRestHandler {
                     logger.error("Failed to send failure response", inner);
                 }
             }
+
         });
     }
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestAnalyzeAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestAnalyzeAction.java
index 04d0bf5761..4b3035e08b 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestAnalyzeAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestAnalyzeAction.java
@@ -30,7 +30,6 @@ import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.RestActions;
@@ -66,7 +65,7 @@ public class RestAnalyzeAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
 
         String[] texts = request.paramAsStringArrayOrEmptyIfAll("text");
 
@@ -99,7 +98,7 @@ public class RestAnalyzeAction extends BaseRestHandler {
             }
         }
 
-        client.admin().indices().analyze(analyzeRequest, new RestToXContentListener<>(channel));
+        return channel -> client.admin().indices().analyze(analyzeRequest, new RestToXContentListener<>(channel));
     }
 
     public static void buildFromContent(BytesReference content, AnalyzeRequest analyzeRequest, ParseFieldMatcher parseFieldMatcher) {
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestClearIndicesCacheAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestClearIndicesCacheAction.java
index 391eaa64d5..942da1f092 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestClearIndicesCacheAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestClearIndicesCacheAction.java
@@ -31,12 +31,12 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
 import org.elasticsearch.rest.action.RestBuilderListener;
 
+import java.io.IOException;
 import java.util.Map;
 
 import static org.elasticsearch.rest.RestRequest.Method.GET;
@@ -57,12 +57,13 @@ public class RestClearIndicesCacheAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         ClearIndicesCacheRequest clearIndicesCacheRequest = new ClearIndicesCacheRequest(
                 Strings.splitStringByCommaToArray(request.param("index")));
         clearIndicesCacheRequest.indicesOptions(IndicesOptions.fromRequest(request, clearIndicesCacheRequest.indicesOptions()));
         fromRequest(request, clearIndicesCacheRequest, parseFieldMatcher);
-        client.admin().indices().clearCache(clearIndicesCacheRequest, new RestBuilderListener<ClearIndicesCacheResponse>(channel) {
+        return channel ->
+                client.admin().indices().clearCache(clearIndicesCacheRequest, new RestBuilderListener<ClearIndicesCacheResponse>(channel) {
             @Override
             public RestResponse buildResponse(ClearIndicesCacheResponse response, XContentBuilder builder) throws Exception {
                 builder.startObject();
@@ -85,6 +86,9 @@ public class RestClearIndicesCacheAction extends BaseRestHandler {
             if (parseFieldMatcher.match(entry.getKey(), Fields.QUERY)) {
                 clearIndicesCacheRequest.queryCache(request.paramAsBoolean(entry.getKey(), clearIndicesCacheRequest.queryCache()));
             }
+            if (parseFieldMatcher.match(entry.getKey(), Fields.REQUEST_CACHE)) {
+                clearIndicesCacheRequest.requestCache(request.paramAsBoolean(entry.getKey(), clearIndicesCacheRequest.requestCache()));
+            }
             if (parseFieldMatcher.match(entry.getKey(), Fields.FIELD_DATA)) {
                 clearIndicesCacheRequest.fieldDataCache(request.paramAsBoolean(entry.getKey(), clearIndicesCacheRequest.fieldDataCache()));
             }
@@ -101,6 +105,7 @@ public class RestClearIndicesCacheAction extends BaseRestHandler {
 
     public static class Fields {
         public static final ParseField QUERY = new ParseField("query", "filter", "filter_cache");
+        public static final ParseField REQUEST_CACHE = new ParseField("request_cache");
         public static final ParseField FIELD_DATA = new ParseField("field_data", "fielddata");
         public static final ParseField RECYCLER = new ParseField("recycler");
         public static final ParseField FIELDS = new ParseField("fields");
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestCloseIndexAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestCloseIndexAction.java
index e5baa27f4e..68929fda4e 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestCloseIndexAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestCloseIndexAction.java
@@ -20,21 +20,18 @@
 package org.elasticsearch.rest.action.admin.indices;
 
 import org.elasticsearch.action.admin.indices.close.CloseIndexRequest;
-import org.elasticsearch.action.admin.indices.close.CloseIndexResponse;
 import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.AcknowledgedRestListener;
 
-/**
- *
- */
+import java.io.IOException;
+
 public class RestCloseIndexAction extends BaseRestHandler {
 
     @Inject
@@ -45,11 +42,12 @@ public class RestCloseIndexAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         CloseIndexRequest closeIndexRequest = new CloseIndexRequest(Strings.splitStringByCommaToArray(request.param("index")));
         closeIndexRequest.masterNodeTimeout(request.paramAsTime("master_timeout", closeIndexRequest.masterNodeTimeout()));
         closeIndexRequest.timeout(request.paramAsTime("timeout", closeIndexRequest.timeout()));
         closeIndexRequest.indicesOptions(IndicesOptions.fromRequest(request, closeIndexRequest.indicesOptions()));
-        client.admin().indices().close(closeIndexRequest, new AcknowledgedRestListener<CloseIndexResponse>(channel));
+        return channel -> client.admin().indices().close(closeIndexRequest, new AcknowledgedRestListener<>(channel));
     }
+
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestCreateIndexAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestCreateIndexAction.java
index 2a7f2a629a..7aabdfd576 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestCreateIndexAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestCreateIndexAction.java
@@ -21,22 +21,18 @@ package org.elasticsearch.rest.action.admin.indices;
 
 import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
 import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
-import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.action.support.ActiveShardCount;
+import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.AcknowledgedRestListener;
 
 import java.io.IOException;
 
-/**
- *
- */
 public class RestCreateIndexAction extends BaseRestHandler {
 
     @Inject
@@ -47,7 +43,7 @@ public class RestCreateIndexAction extends BaseRestHandler {
 
     @SuppressWarnings({"unchecked"})
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         CreateIndexRequest createIndexRequest = new CreateIndexRequest(request.param("index"));
         if (request.hasContent()) {
             createIndexRequest.source(request.content());
@@ -56,7 +52,7 @@ public class RestCreateIndexAction extends BaseRestHandler {
         createIndexRequest.timeout(request.paramAsTime("timeout", createIndexRequest.timeout()));
         createIndexRequest.masterNodeTimeout(request.paramAsTime("master_timeout", createIndexRequest.masterNodeTimeout()));
         createIndexRequest.waitForActiveShards(ActiveShardCount.parseString(request.param("wait_for_active_shards")));
-        client.admin().indices().create(createIndexRequest, new AcknowledgedRestListener<CreateIndexResponse>(channel) {
+        return channel -> client.admin().indices().create(createIndexRequest, new AcknowledgedRestListener<CreateIndexResponse>(channel) {
             @Override
             public void addCustomFields(XContentBuilder builder, CreateIndexResponse response) throws IOException {
                 response.addCustomFields(builder);
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestDeleteIndexAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestDeleteIndexAction.java
index d3e07effc1..94ffc7e989 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestDeleteIndexAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestDeleteIndexAction.java
@@ -20,21 +20,18 @@
 package org.elasticsearch.rest.action.admin.indices;
 
 import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
-import org.elasticsearch.action.admin.indices.delete.DeleteIndexResponse;
 import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.AcknowledgedRestListener;
 
-/**
- *
- */
+import java.io.IOException;
+
 public class RestDeleteIndexAction extends BaseRestHandler {
 
     @Inject
@@ -45,11 +42,11 @@ public class RestDeleteIndexAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest(Strings.splitStringByCommaToArray(request.param("index")));
         deleteIndexRequest.timeout(request.paramAsTime("timeout", deleteIndexRequest.timeout()));
         deleteIndexRequest.masterNodeTimeout(request.paramAsTime("master_timeout", deleteIndexRequest.masterNodeTimeout()));
         deleteIndexRequest.indicesOptions(IndicesOptions.fromRequest(request, deleteIndexRequest.indicesOptions()));
-        client.admin().indices().delete(deleteIndexRequest, new AcknowledgedRestListener<DeleteIndexResponse>(channel));
+        return channel -> client.admin().indices().delete(deleteIndexRequest, new AcknowledgedRestListener<>(channel));
     }
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestDeleteIndexTemplateAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestDeleteIndexTemplateAction.java
index 425581fe92..3d721d2748 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestDeleteIndexTemplateAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestDeleteIndexTemplateAction.java
@@ -23,11 +23,12 @@ import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.AcknowledgedRestListener;
 
+import java.io.IOException;
+
 public class RestDeleteIndexTemplateAction extends BaseRestHandler {
 
     @Inject
@@ -37,9 +38,9 @@ public class RestDeleteIndexTemplateAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         DeleteIndexTemplateRequest deleteIndexTemplateRequest = new DeleteIndexTemplateRequest(request.param("name"));
         deleteIndexTemplateRequest.masterNodeTimeout(request.paramAsTime("master_timeout", deleteIndexTemplateRequest.masterNodeTimeout()));
-        client.admin().indices().deleteTemplate(deleteIndexTemplateRequest, new AcknowledgedRestListener<>(channel));
+        return channel -> client.admin().indices().deleteTemplate(deleteIndexTemplateRequest, new AcknowledgedRestListener<>(channel));
     }
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestFlushAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestFlushAction.java
index b963a80593..ac58dc1f6e 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestFlushAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestFlushAction.java
@@ -29,12 +29,13 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
 import org.elasticsearch.rest.action.RestBuilderListener;
 
+import java.io.IOException;
+
 import static org.elasticsearch.rest.RestRequest.Method.GET;
 import static org.elasticsearch.rest.RestRequest.Method.POST;
 import static org.elasticsearch.rest.RestStatus.OK;
@@ -56,12 +57,12 @@ public class RestFlushAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         FlushRequest flushRequest = new FlushRequest(Strings.splitStringByCommaToArray(request.param("index")));
         flushRequest.indicesOptions(IndicesOptions.fromRequest(request, flushRequest.indicesOptions()));
         flushRequest.force(request.paramAsBoolean("force", flushRequest.force()));
         flushRequest.waitIfOngoing(request.paramAsBoolean("wait_if_ongoing", flushRequest.waitIfOngoing()));
-        client.admin().indices().flush(flushRequest, new RestBuilderListener<FlushResponse>(channel) {
+        return channel -> client.admin().indices().flush(flushRequest, new RestBuilderListener<FlushResponse>(channel) {
             @Override
             public RestResponse buildResponse(FlushResponse response, XContentBuilder builder) throws Exception {
                 builder.startObject();
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestForceMergeAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestForceMergeAction.java
index c376866ad1..d253b02bbd 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestForceMergeAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestForceMergeAction.java
@@ -29,12 +29,13 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
 import org.elasticsearch.rest.action.RestBuilderListener;
 
+import java.io.IOException;
+
 import static org.elasticsearch.rest.RestRequest.Method.POST;
 import static org.elasticsearch.rest.RestStatus.OK;
 import static org.elasticsearch.rest.action.RestActions.buildBroadcastShardsHeader;
@@ -52,13 +53,13 @@ public class RestForceMergeAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         ForceMergeRequest mergeRequest = new ForceMergeRequest(Strings.splitStringByCommaToArray(request.param("index")));
         mergeRequest.indicesOptions(IndicesOptions.fromRequest(request, mergeRequest.indicesOptions()));
         mergeRequest.maxNumSegments(request.paramAsInt("max_num_segments", mergeRequest.maxNumSegments()));
         mergeRequest.onlyExpungeDeletes(request.paramAsBoolean("only_expunge_deletes", mergeRequest.onlyExpungeDeletes()));
         mergeRequest.flush(request.paramAsBoolean("flush", mergeRequest.flush()));
-        client.admin().indices().forceMerge(mergeRequest, new RestBuilderListener<ForceMergeResponse>(channel) {
+        return channel -> client.admin().indices().forceMerge(mergeRequest, new RestBuilderListener<ForceMergeResponse>(channel) {
             @Override
             public RestResponse buildResponse(ForceMergeResponse response, XContentBuilder builder) throws Exception {
                 builder.startObject();
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetAliasesAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetAliasesAction.java
index 4b58fd0f16..e740875789 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetAliasesAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetAliasesAction.java
@@ -32,13 +32,13 @@ import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
 import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.rest.action.RestBuilderListener;
 
+import java.io.IOException;
 import java.util.List;
 import java.util.Locale;
 
@@ -57,7 +57,7 @@ public class RestGetAliasesAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         final String[] aliases = request.paramAsStringArrayOrEmptyIfAll("name");
         final String[] indices = Strings.splitStringByCommaToArray(request.param("index"));
         final GetAliasesRequest getAliasesRequest = new GetAliasesRequest(aliases);
@@ -65,7 +65,7 @@ public class RestGetAliasesAction extends BaseRestHandler {
         getAliasesRequest.indicesOptions(IndicesOptions.fromRequest(request, getAliasesRequest.indicesOptions()));
         getAliasesRequest.local(request.paramAsBoolean("local", getAliasesRequest.local()));
 
-        client.admin().indices().getAliases(getAliasesRequest, new RestBuilderListener<GetAliasesResponse>(channel) {
+        return channel -> client.admin().indices().getAliases(getAliasesRequest, new RestBuilderListener<GetAliasesResponse>(channel) {
             @Override
             public RestResponse buildResponse(GetAliasesResponse response, XContentBuilder builder) throws Exception {
                 // empty body, if indices were specified but no aliases were
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetFieldMappingAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetFieldMappingAction.java
index fba4192d5d..ac51978a39 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetFieldMappingAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetFieldMappingAction.java
@@ -30,13 +30,13 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
 import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.rest.action.RestBuilderListener;
 
+import java.io.IOException;
 import java.util.Map;
 
 import static org.elasticsearch.rest.RestRequest.Method.GET;
@@ -56,7 +56,7 @@ public class RestGetFieldMappingAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         final String[] indices = Strings.splitStringByCommaToArray(request.param("index"));
         final String[] types = request.paramAsStringArrayOrEmptyIfAll("type");
         final String[] fields = Strings.splitStringByCommaToArray(request.param("fields"));
@@ -64,26 +64,27 @@ public class RestGetFieldMappingAction extends BaseRestHandler {
         getMappingsRequest.indices(indices).types(types).fields(fields).includeDefaults(request.paramAsBoolean("include_defaults", false));
         getMappingsRequest.indicesOptions(IndicesOptions.fromRequest(request, getMappingsRequest.indicesOptions()));
         getMappingsRequest.local(request.paramAsBoolean("local", getMappingsRequest.local()));
-        client.admin().indices().getFieldMappings(getMappingsRequest, new RestBuilderListener<GetFieldMappingsResponse>(channel) {
-            @Override
-            public RestResponse buildResponse(GetFieldMappingsResponse response, XContentBuilder builder) throws Exception {
-                Map<String, Map<String, Map<String, FieldMappingMetaData>>> mappingsByIndex = response.mappings();
+        return channel ->
+                client.admin().indices().getFieldMappings(getMappingsRequest, new RestBuilderListener<GetFieldMappingsResponse>(channel) {
+                    @Override
+                    public RestResponse buildResponse(GetFieldMappingsResponse response, XContentBuilder builder) throws Exception {
+                        Map<String, Map<String, Map<String, FieldMappingMetaData>>> mappingsByIndex = response.mappings();
 
-                boolean isPossibleSingleFieldRequest = indices.length == 1 && types.length == 1 && fields.length == 1;
-                if (isPossibleSingleFieldRequest && isFieldMappingMissingField(mappingsByIndex)) {
-                    return new BytesRestResponse(OK, builder.startObject().endObject());
-                }
+                        boolean isPossibleSingleFieldRequest = indices.length == 1 && types.length == 1 && fields.length == 1;
+                        if (isPossibleSingleFieldRequest && isFieldMappingMissingField(mappingsByIndex)) {
+                            return new BytesRestResponse(OK, builder.startObject().endObject());
+                        }
 
-                RestStatus status = OK;
-                if (mappingsByIndex.isEmpty() && fields.length > 0) {
-                    status = NOT_FOUND;
-                }
-                builder.startObject();
-                response.toXContent(builder, request);
-                builder.endObject();
-                return new BytesRestResponse(status, builder);
-            }
-        });
+                        RestStatus status = OK;
+                        if (mappingsByIndex.isEmpty() && fields.length > 0) {
+                            status = NOT_FOUND;
+                        }
+                        builder.startObject();
+                        response.toXContent(builder, request);
+                        builder.endObject();
+                        return new BytesRestResponse(status, builder);
+                    }
+                });
     }
 
     /**
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetIndexTemplateAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetIndexTemplateAction.java
index ead0453259..0d1752d736 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetIndexTemplateAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetIndexTemplateAction.java
@@ -16,6 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
+
 package org.elasticsearch.rest.action.admin.indices;
 
 import org.elasticsearch.action.admin.indices.template.get.GetIndexTemplatesRequest;
@@ -25,12 +26,14 @@ import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.rest.action.RestToXContentListener;
 
+import java.io.IOException;
+import java.util.Set;
+
 import static org.elasticsearch.rest.RestRequest.Method.GET;
 import static org.elasticsearch.rest.RestStatus.NOT_FOUND;
 import static org.elasticsearch.rest.RestStatus.OK;
@@ -46,7 +49,7 @@ public class RestGetIndexTemplateAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         final String[] names = Strings.splitStringByCommaToArray(request.param("name"));
 
         GetIndexTemplatesRequest getIndexTemplatesRequest = new GetIndexTemplatesRequest(names);
@@ -55,13 +58,22 @@ public class RestGetIndexTemplateAction extends BaseRestHandler {
 
         final boolean implicitAll = getIndexTemplatesRequest.names().length == 0;
 
-        client.admin().indices().getTemplates(getIndexTemplatesRequest, new RestToXContentListener<GetIndexTemplatesResponse>(channel) {
-            @Override
-            protected RestStatus getStatus(GetIndexTemplatesResponse response) {
-                boolean templateExists = false == response.getIndexTemplates().isEmpty();
+        return channel ->
+                client.admin()
+                        .indices()
+                        .getTemplates(getIndexTemplatesRequest, new RestToXContentListener<GetIndexTemplatesResponse>(channel) {
+                            @Override
+                            protected RestStatus getStatus(GetIndexTemplatesResponse response) {
+                                boolean templateExists = false == response.getIndexTemplates().isEmpty();
+
+                                return (templateExists || implicitAll) ? OK : NOT_FOUND;
+                            }
+                        });
+    }
 
-                return (templateExists || implicitAll) ? OK : NOT_FOUND;
-            }
-        });
+    @Override
+    protected Set<String> responseParams() {
+        return Settings.FORMAT_PARAMS;
     }
+
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetIndicesAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetIndicesAction.java
index 2ad6f245cf..b7cf07945b 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetIndicesAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetIndicesAction.java
@@ -36,7 +36,6 @@ import org.elasticsearch.common.xcontent.ToXContent.Params;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
@@ -44,6 +43,7 @@ import org.elasticsearch.rest.action.RestBuilderListener;
 
 import java.io.IOException;
 import java.util.List;
+import java.util.Set;
 
 import static org.elasticsearch.rest.RestRequest.Method.GET;
 import static org.elasticsearch.rest.RestStatus.OK;
@@ -64,7 +64,7 @@ public class RestGetIndicesAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         String[] indices = Strings.splitStringByCommaToArray(request.param("index"));
         String[] featureParams = request.paramAsStringArray("type", null);
         // Work out if the indices is a list of features
@@ -81,7 +81,8 @@ public class RestGetIndicesAction extends BaseRestHandler {
         getIndexRequest.indicesOptions(IndicesOptions.fromRequest(request, getIndexRequest.indicesOptions()));
         getIndexRequest.local(request.paramAsBoolean("local", getIndexRequest.local()));
         getIndexRequest.humanReadable(request.paramAsBoolean("human", false));
-        client.admin().indices().getIndex(getIndexRequest, new RestBuilderListener<GetIndexResponse>(channel) {
+        final boolean defaults = request.paramAsBoolean("include_defaults", false);
+        return channel -> client.admin().indices().getIndex(getIndexRequest, new RestBuilderListener<GetIndexResponse>(channel) {
 
             @Override
             public RestResponse buildResponse(GetIndexResponse response, XContentBuilder builder) throws Exception {
@@ -100,7 +101,7 @@ public class RestGetIndicesAction extends BaseRestHandler {
                             writeMappings(response.mappings().get(index), builder, request);
                             break;
                         case SETTINGS:
-                            writeSettings(response.settings().get(index), builder, request);
+                            writeSettings(response.settings().get(index), builder, request, defaults);
                             break;
                         default:
                             throw new IllegalStateException("feature [" + feature + "] is not valid");
@@ -136,15 +137,15 @@ public class RestGetIndicesAction extends BaseRestHandler {
                 builder.endObject();
             }
 
-            private void writeSettings(Settings settings, XContentBuilder builder, Params params) throws IOException {
-                final boolean renderDefaults = request.paramAsBoolean("include_defaults", false);
+            private void writeSettings(Settings settings, XContentBuilder builder, Params params, boolean defaults) throws IOException {
                 builder.startObject(Fields.SETTINGS);
                 settings.toXContent(builder, params);
                 builder.endObject();
-                if (renderDefaults) {
+                if (defaults) {
                     builder.startObject("defaults");
-                    settingsFilter.filter(indexScopedSettings.diff(settings, RestGetIndicesAction.this.settings)).toXContent(builder,
-                            request);
+                    settingsFilter
+                        .filter(indexScopedSettings.diff(settings, RestGetIndicesAction.this.settings))
+                        .toXContent(builder, request);
                     builder.endObject();
                 }
             }
@@ -152,6 +153,11 @@ public class RestGetIndicesAction extends BaseRestHandler {
         });
     }
 
+    @Override
+    protected Set<String> responseParams() {
+        return Settings.FORMAT_PARAMS;
+    }
+
     static class Fields {
         static final String ALIASES = "aliases";
         static final String MAPPINGS = "mappings";
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetMappingAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetMappingAction.java
index b9c8472964..2bb3ae8170 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetMappingAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetMappingAction.java
@@ -20,7 +20,6 @@
 package org.elasticsearch.rest.action.admin.indices;
 
 import com.carrotsearch.hppc.cursors.ObjectObjectCursor;
-
 import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsRequest;
 import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsResponse;
 import org.elasticsearch.action.support.IndicesOptions;
@@ -35,12 +34,13 @@ import org.elasticsearch.index.IndexNotFoundException;
 import org.elasticsearch.indices.TypeMissingException;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
 import org.elasticsearch.rest.action.RestBuilderListener;
 
+import java.io.IOException;
+
 import static org.elasticsearch.rest.RestRequest.Method.GET;
 import static org.elasticsearch.rest.RestStatus.OK;
 
@@ -59,14 +59,14 @@ public class RestGetMappingAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         final String[] indices = Strings.splitStringByCommaToArray(request.param("index"));
         final String[] types = request.paramAsStringArrayOrEmptyIfAll("type");
         GetMappingsRequest getMappingsRequest = new GetMappingsRequest();
         getMappingsRequest.indices(indices).types(types);
         getMappingsRequest.indicesOptions(IndicesOptions.fromRequest(request, getMappingsRequest.indicesOptions()));
         getMappingsRequest.local(request.paramAsBoolean("local", getMappingsRequest.local()));
-        client.admin().indices().getMappings(getMappingsRequest, new RestBuilderListener<GetMappingsResponse>(channel) {
+        return channel -> client.admin().indices().getMappings(getMappingsRequest, new RestBuilderListener<GetMappingsResponse>(channel) {
             @Override
             public RestResponse buildResponse(GetMappingsResponse response, XContentBuilder builder) throws Exception {
                 builder.startObject();
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetSettingsAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetSettingsAction.java
index 936a96e035..10f0564de8 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetSettingsAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetSettingsAction.java
@@ -32,12 +32,13 @@ import org.elasticsearch.common.settings.SettingsFilter;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
 import org.elasticsearch.rest.action.RestBuilderListener;
 
+import java.io.IOException;
+
 import static org.elasticsearch.rest.RestRequest.Method.GET;
 import static org.elasticsearch.rest.RestStatus.OK;
 
@@ -58,7 +59,7 @@ public class RestGetSettingsAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         final String[] names = request.paramAsStringArrayOrEmptyIfAll("name");
         final boolean renderDefaults = request.paramAsBoolean("include_defaults", false);
         GetSettingsRequest getSettingsRequest = new GetSettingsRequest()
@@ -68,7 +69,7 @@ public class RestGetSettingsAction extends BaseRestHandler {
                 .names(names);
         getSettingsRequest.local(request.paramAsBoolean("local", getSettingsRequest.local()));
 
-        client.admin().indices().getSettings(getSettingsRequest, new RestBuilderListener<GetSettingsResponse>(channel) {
+        return channel -> client.admin().indices().getSettings(getSettingsRequest, new RestBuilderListener<GetSettingsResponse>(channel) {
 
             @Override
             public RestResponse buildResponse(GetSettingsResponse getSettingsResponse, XContentBuilder builder) throws Exception {
@@ -94,4 +95,5 @@ public class RestGetSettingsAction extends BaseRestHandler {
             }
         });
     }
+
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestHeadIndexTemplateAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestHeadIndexTemplateAction.java
index 3480fbb9af..bea821e485 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestHeadIndexTemplateAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestHeadIndexTemplateAction.java
@@ -26,19 +26,18 @@ import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
 import org.elasticsearch.rest.action.RestResponseListener;
 
+import java.io.IOException;
+import java.util.Set;
+
 import static org.elasticsearch.rest.RestRequest.Method.HEAD;
 import static org.elasticsearch.rest.RestStatus.NOT_FOUND;
 import static org.elasticsearch.rest.RestStatus.OK;
 
-/**
- *
- */
 public class RestHeadIndexTemplateAction extends BaseRestHandler {
 
     @Inject
@@ -49,20 +48,29 @@ public class RestHeadIndexTemplateAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         GetIndexTemplatesRequest getIndexTemplatesRequest = new GetIndexTemplatesRequest(request.param("name"));
         getIndexTemplatesRequest.local(request.paramAsBoolean("local", getIndexTemplatesRequest.local()));
         getIndexTemplatesRequest.masterNodeTimeout(request.paramAsTime("master_timeout", getIndexTemplatesRequest.masterNodeTimeout()));
-        client.admin().indices().getTemplates(getIndexTemplatesRequest, new RestResponseListener<GetIndexTemplatesResponse>(channel) {
-            @Override
-            public RestResponse buildResponse(GetIndexTemplatesResponse getIndexTemplatesResponse) {
-                boolean templateExists = getIndexTemplatesResponse.getIndexTemplates().size() > 0;
-                if (templateExists) {
-                    return new BytesRestResponse(OK, BytesRestResponse.TEXT_CONTENT_TYPE, BytesArray.EMPTY);
-                } else {
-                    return new BytesRestResponse(NOT_FOUND, BytesRestResponse.TEXT_CONTENT_TYPE, BytesArray.EMPTY);
-                }
-            }
-        });
+        return channel ->
+                client.admin()
+                        .indices()
+                        .getTemplates(getIndexTemplatesRequest, new RestResponseListener<GetIndexTemplatesResponse>(channel) {
+                            @Override
+                            public RestResponse buildResponse(GetIndexTemplatesResponse getIndexTemplatesResponse) {
+                                boolean templateExists = getIndexTemplatesResponse.getIndexTemplates().size() > 0;
+                                if (templateExists) {
+                                    return new BytesRestResponse(OK, BytesRestResponse.TEXT_CONTENT_TYPE, BytesArray.EMPTY);
+                                } else {
+                                    return new BytesRestResponse(NOT_FOUND, BytesRestResponse.TEXT_CONTENT_TYPE, BytesArray.EMPTY);
+                                }
+                            }
+                        });
     }
+
+    @Override
+    protected Set<String> responseParams() {
+        return Settings.FORMAT_PARAMS;
+    }
+
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestIndexDeleteAliasesAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestIndexDeleteAliasesAction.java
index b027aeb8d6..ea6d1fca64 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestIndexDeleteAliasesAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestIndexDeleteAliasesAction.java
@@ -19,18 +19,18 @@
 package org.elasticsearch.rest.action.admin.indices;
 
 import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequest;
-import org.elasticsearch.action.admin.indices.alias.IndicesAliasesResponse;
 import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequest.AliasActions;
 import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.AcknowledgedRestListener;
 
+import java.io.IOException;
+
 import static org.elasticsearch.rest.RestRequest.Method.DELETE;
 
 /**
@@ -45,7 +45,7 @@ public class RestIndexDeleteAliasesAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         final String[] indices = Strings.splitStringByCommaToArray(request.param("index"));
         final String[] aliases = Strings.splitStringByCommaToArray(request.param("name"));
         IndicesAliasesRequest indicesAliasesRequest = new IndicesAliasesRequest();
@@ -53,6 +53,6 @@ public class RestIndexDeleteAliasesAction extends BaseRestHandler {
         indicesAliasesRequest.addAliasAction(AliasActions.remove().indices(indices).aliases(aliases));
         indicesAliasesRequest.masterNodeTimeout(request.paramAsTime("master_timeout", indicesAliasesRequest.masterNodeTimeout()));
 
-        client.admin().indices().aliases(indicesAliasesRequest, new AcknowledgedRestListener<IndicesAliasesResponse>(channel));
+        return channel -> client.admin().indices().aliases(indicesAliasesRequest, new AcknowledgedRestListener<>(channel));
     }
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestIndexPutAliasAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestIndexPutAliasAction.java
index f7546bd57d..da8ab809c1 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestIndexPutAliasAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestIndexPutAliasAction.java
@@ -27,11 +27,11 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentFactory;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.AcknowledgedRestListener;
 
+import java.io.IOException;
 import java.util.Map;
 
 import static org.elasticsearch.rest.RestRequest.Method.POST;
@@ -58,7 +58,7 @@ public class RestIndexPutAliasAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) throws Exception {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         String[] indices = Strings.splitStringByCommaToArray(request.param("index"));
         String alias = request.param("name");
         Map<String, Object> filter = null;
@@ -117,6 +117,6 @@ public class RestIndexPutAliasAction extends BaseRestHandler {
             aliasAction.filter(filter);
         }
         indicesAliasesRequest.addAliasAction(aliasAction);
-        client.admin().indices().aliases(indicesAliasesRequest, new AcknowledgedRestListener<>(channel));
+        return channel -> client.admin().indices().aliases(indicesAliasesRequest, new AcknowledgedRestListener<>(channel));
     }
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestIndicesAliasesAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestIndicesAliasesAction.java
index fe8a6a1662..d02e2caf5a 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestIndicesAliasesAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestIndicesAliasesAction.java
@@ -32,11 +32,12 @@ import org.elasticsearch.common.xcontent.ObjectParser;
 import org.elasticsearch.common.xcontent.XContentFactory;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.AcknowledgedRestListener;
 
+import java.io.IOException;
+
 import static org.elasticsearch.rest.RestRequest.Method.POST;
 
 public class RestIndicesAliasesAction extends BaseRestHandler {
@@ -56,7 +57,7 @@ public class RestIndicesAliasesAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) throws Exception {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         IndicesAliasesRequest indicesAliasesRequest = new IndicesAliasesRequest();
         indicesAliasesRequest.masterNodeTimeout(request.paramAsTime("master_timeout", indicesAliasesRequest.masterNodeTimeout()));
         indicesAliasesRequest.timeout(request.paramAsTime("timeout", indicesAliasesRequest.timeout()));
@@ -66,6 +67,6 @@ public class RestIndicesAliasesAction extends BaseRestHandler {
         if (indicesAliasesRequest.getAliasActions().isEmpty()) {
             throw new IllegalArgumentException("No action specified");
         }
-        client.admin().indices().aliases(indicesAliasesRequest, new AcknowledgedRestListener<IndicesAliasesResponse>(channel));
+        return channel -> client.admin().indices().aliases(indicesAliasesRequest, new AcknowledgedRestListener<>(channel));
     }
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestIndicesExistsAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestIndicesExistsAction.java
index fa62a84435..520bca4e53 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestIndicesExistsAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestIndicesExistsAction.java
@@ -29,19 +29,17 @@ import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
 import org.elasticsearch.rest.action.RestResponseListener;
 
+import java.io.IOException;
+
 import static org.elasticsearch.rest.RestRequest.Method.HEAD;
 import static org.elasticsearch.rest.RestStatus.NOT_FOUND;
 import static org.elasticsearch.rest.RestStatus.OK;
 
-/**
- *
- */
 public class RestIndicesExistsAction extends BaseRestHandler {
 
     @Inject
@@ -51,11 +49,11 @@ public class RestIndicesExistsAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         IndicesExistsRequest indicesExistsRequest = new IndicesExistsRequest(Strings.splitStringByCommaToArray(request.param("index")));
         indicesExistsRequest.indicesOptions(IndicesOptions.fromRequest(request, indicesExistsRequest.indicesOptions()));
         indicesExistsRequest.local(request.paramAsBoolean("local", indicesExistsRequest.local()));
-        client.admin().indices().exists(indicesExistsRequest, new RestResponseListener<IndicesExistsResponse>(channel) {
+        return channel -> client.admin().indices().exists(indicesExistsRequest, new RestResponseListener<IndicesExistsResponse>(channel) {
             @Override
             public RestResponse buildResponse(IndicesExistsResponse response) {
                 if (response.isExists()) {
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestIndicesSegmentsAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestIndicesSegmentsAction.java
index db9de980c5..556bb5b1d1 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestIndicesSegmentsAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestIndicesSegmentsAction.java
@@ -29,12 +29,13 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
 import org.elasticsearch.rest.action.RestBuilderListener;
 
+import java.io.IOException;
+
 import static org.elasticsearch.rest.RestRequest.Method.GET;
 import static org.elasticsearch.rest.RestStatus.OK;
 import static org.elasticsearch.rest.action.RestActions.buildBroadcastShardsHeader;
@@ -49,20 +50,21 @@ public class RestIndicesSegmentsAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         IndicesSegmentsRequest indicesSegmentsRequest = new IndicesSegmentsRequest(
                 Strings.splitStringByCommaToArray(request.param("index")));
         indicesSegmentsRequest.verbose(request.paramAsBoolean("verbose", false));
         indicesSegmentsRequest.indicesOptions(IndicesOptions.fromRequest(request, indicesSegmentsRequest.indicesOptions()));
-        client.admin().indices().segments(indicesSegmentsRequest, new RestBuilderListener<IndicesSegmentResponse>(channel) {
-            @Override
-            public RestResponse buildResponse(IndicesSegmentResponse response, XContentBuilder builder) throws Exception {
-                builder.startObject();
-                buildBroadcastShardsHeader(builder, request, response);
-                response.toXContent(builder, request);
-                builder.endObject();
-                return new BytesRestResponse(OK, builder);
-            }
-        });
+        return channel ->
+            client.admin().indices().segments(indicesSegmentsRequest, new RestBuilderListener<IndicesSegmentResponse>(channel) {
+                @Override
+                public RestResponse buildResponse(IndicesSegmentResponse response, XContentBuilder builder) throws Exception {
+                    builder.startObject();
+                    buildBroadcastShardsHeader(builder, request, response);
+                    response.toXContent(builder, request);
+                    builder.endObject();
+                    return new BytesRestResponse(OK, builder);
+                }
+            });
     }
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestIndicesShardStoresAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestIndicesShardStoresAction.java
index 65c0dc8aa4..cf1ef9ce5e 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestIndicesShardStoresAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestIndicesShardStoresAction.java
@@ -30,12 +30,13 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
 import org.elasticsearch.rest.action.RestBuilderListener;
 
+import java.io.IOException;
+
 import static org.elasticsearch.rest.RestRequest.Method.GET;
 import static org.elasticsearch.rest.RestStatus.OK;
 
@@ -52,21 +53,26 @@ public class RestIndicesShardStoresAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         IndicesShardStoresRequest indicesShardStoresRequest = new IndicesShardStoresRequest(
                 Strings.splitStringByCommaToArray(request.param("index")));
         if (request.hasParam("status")) {
             indicesShardStoresRequest.shardStatuses(Strings.splitStringByCommaToArray(request.param("status")));
         }
         indicesShardStoresRequest.indicesOptions(IndicesOptions.fromRequest(request, indicesShardStoresRequest.indicesOptions()));
-        client.admin().indices().shardStores(indicesShardStoresRequest, new RestBuilderListener<IndicesShardStoresResponse>(channel) {
-            @Override
-            public RestResponse buildResponse(IndicesShardStoresResponse response, XContentBuilder builder) throws Exception {
-                builder.startObject();
-                response.toXContent(builder, request);
-                builder.endObject();
-                return new BytesRestResponse(OK, builder);
-            }
-        });
+        return channel ->
+            client.admin()
+                .indices()
+                .shardStores(indicesShardStoresRequest, new RestBuilderListener<IndicesShardStoresResponse>(channel) {
+                    @Override
+                    public RestResponse buildResponse(
+                        IndicesShardStoresResponse response,
+                        XContentBuilder builder) throws Exception {
+                        builder.startObject();
+                        response.toXContent(builder, request);
+                        builder.endObject();
+                        return new BytesRestResponse(OK, builder);
+                    }
+                });
     }
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestIndicesStatsAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestIndicesStatsAction.java
index c7dd62688f..e7336225c5 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestIndicesStatsAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestIndicesStatsAction.java
@@ -29,12 +29,13 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
 import org.elasticsearch.rest.action.RestBuilderListener;
 
+import java.io.IOException;
+import java.util.Collections;
 import java.util.Set;
 
 import static org.elasticsearch.rest.RestRequest.Method.GET;
@@ -54,7 +55,7 @@ public class RestIndicesStatsAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         IndicesStatsRequest indicesStatsRequest = new IndicesStatsRequest();
         indicesStatsRequest.indicesOptions(IndicesOptions.fromRequest(request, indicesStatsRequest.indicesOptions()));
         indicesStatsRequest.indices(Strings.splitStringByCommaToArray(request.param("index")));
@@ -106,7 +107,7 @@ public class RestIndicesStatsAction extends BaseRestHandler {
             indicesStatsRequest.includeSegmentFileSizes(true);
         }
 
-        client.admin().indices().stats(indicesStatsRequest, new RestBuilderListener<IndicesStatsResponse>(channel) {
+        return channel -> client.admin().indices().stats(indicesStatsRequest, new RestBuilderListener<IndicesStatsResponse>(channel) {
             @Override
             public RestResponse buildResponse(IndicesStatsResponse response, XContentBuilder builder) throws Exception {
                 builder.startObject();
@@ -122,4 +123,12 @@ public class RestIndicesStatsAction extends BaseRestHandler {
     public boolean canTripCircuitBreaker() {
         return false;
     }
+
+    private static final Set<String> RESPONSE_PARAMS = Collections.singleton("level");
+
+    @Override
+    protected Set<String> responseParams() {
+        return RESPONSE_PARAMS;
+    }
+
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestOpenIndexAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestOpenIndexAction.java
index dd40705769..e606f39dfe 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestOpenIndexAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestOpenIndexAction.java
@@ -27,11 +27,12 @@ import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.AcknowledgedRestListener;
 
+import java.io.IOException;
+
 /**
  *
  */
@@ -45,11 +46,11 @@ public class RestOpenIndexAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         OpenIndexRequest openIndexRequest = new OpenIndexRequest(Strings.splitStringByCommaToArray(request.param("index")));
         openIndexRequest.timeout(request.paramAsTime("timeout", openIndexRequest.timeout()));
         openIndexRequest.masterNodeTimeout(request.paramAsTime("master_timeout", openIndexRequest.masterNodeTimeout()));
         openIndexRequest.indicesOptions(IndicesOptions.fromRequest(request, openIndexRequest.indicesOptions()));
-        client.admin().indices().open(openIndexRequest, new AcknowledgedRestListener<OpenIndexResponse>(channel));
+        return channel -> client.admin().indices().open(openIndexRequest, new AcknowledgedRestListener<OpenIndexResponse>(channel));
     }
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestPutIndexTemplateAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestPutIndexTemplateAction.java
index 3deba4c32f..aee4eb3a9e 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestPutIndexTemplateAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestPutIndexTemplateAction.java
@@ -16,6 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
+
 package org.elasticsearch.rest.action.admin.indices;
 
 import org.elasticsearch.action.admin.indices.template.put.PutIndexTemplateRequest;
@@ -23,14 +24,12 @@ import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.AcknowledgedRestListener;
 
-/**
- *
- */
+import java.io.IOException;
+
 public class RestPutIndexTemplateAction extends BaseRestHandler {
 
     @Inject
@@ -41,7 +40,7 @@ public class RestPutIndexTemplateAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         PutIndexTemplateRequest putRequest = new PutIndexTemplateRequest(request.param("name"));
         putRequest.template(request.param("template", putRequest.template()));
         putRequest.order(request.paramAsInt("order", putRequest.order()));
@@ -49,6 +48,7 @@ public class RestPutIndexTemplateAction extends BaseRestHandler {
         putRequest.create(request.paramAsBoolean("create", false));
         putRequest.cause(request.param("cause", ""));
         putRequest.source(request.content());
-        client.admin().indices().putTemplate(putRequest, new AcknowledgedRestListener<>(channel));
+        return channel -> client.admin().indices().putTemplate(putRequest, new AcknowledgedRestListener<>(channel));
     }
+
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestPutMappingAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestPutMappingAction.java
index 3a582d0b0a..7d3b9ba479 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestPutMappingAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestPutMappingAction.java
@@ -27,11 +27,12 @@ import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.AcknowledgedRestListener;
 
+import java.io.IOException;
+
 import static org.elasticsearch.client.Requests.putMappingRequest;
 import static org.elasticsearch.rest.RestRequest.Method.POST;
 import static org.elasticsearch.rest.RestRequest.Method.PUT;
@@ -68,7 +69,7 @@ public class RestPutMappingAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         PutMappingRequest putMappingRequest = putMappingRequest(Strings.splitStringByCommaToArray(request.param("index")));
         putMappingRequest.type(request.param("type"));
         putMappingRequest.source(request.content().utf8ToString());
@@ -76,6 +77,6 @@ public class RestPutMappingAction extends BaseRestHandler {
         putMappingRequest.timeout(request.paramAsTime("timeout", putMappingRequest.timeout()));
         putMappingRequest.masterNodeTimeout(request.paramAsTime("master_timeout", putMappingRequest.masterNodeTimeout()));
         putMappingRequest.indicesOptions(IndicesOptions.fromRequest(request, putMappingRequest.indicesOptions()));
-        client.admin().indices().putMapping(putMappingRequest, new AcknowledgedRestListener<PutMappingResponse>(channel));
+        return channel -> client.admin().indices().putMapping(putMappingRequest, new AcknowledgedRestListener<PutMappingResponse>(channel));
     }
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestRecoveryAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestRecoveryAction.java
index 5dee73606f..c20c70c8c2 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestRecoveryAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestRecoveryAction.java
@@ -29,12 +29,13 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
 import org.elasticsearch.rest.action.RestBuilderListener;
 
+import java.io.IOException;
+
 import static org.elasticsearch.rest.RestRequest.Method.GET;
 import static org.elasticsearch.rest.RestStatus.OK;
 
@@ -51,14 +52,14 @@ public class RestRecoveryAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
 
         final RecoveryRequest recoveryRequest = new RecoveryRequest(Strings.splitStringByCommaToArray(request.param("index")));
         recoveryRequest.detailed(request.paramAsBoolean("detailed", false));
         recoveryRequest.activeOnly(request.paramAsBoolean("active_only", false));
         recoveryRequest.indicesOptions(IndicesOptions.fromRequest(request, recoveryRequest.indicesOptions()));
 
-        client.admin().indices().recoveries(recoveryRequest, new RestBuilderListener<RecoveryResponse>(channel) {
+        return channel -> client.admin().indices().recoveries(recoveryRequest, new RestBuilderListener<RecoveryResponse>(channel) {
             @Override
             public RestResponse buildResponse(RecoveryResponse response, XContentBuilder builder) throws Exception {
                 response.detailed(recoveryRequest.detailed());
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestRefreshAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestRefreshAction.java
index 54088a7ddb..96afc014e9 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestRefreshAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestRefreshAction.java
@@ -29,20 +29,18 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
 import org.elasticsearch.rest.action.RestBuilderListener;
 
+import java.io.IOException;
+
 import static org.elasticsearch.rest.RestRequest.Method.GET;
 import static org.elasticsearch.rest.RestRequest.Method.POST;
 import static org.elasticsearch.rest.RestStatus.OK;
 import static org.elasticsearch.rest.action.RestActions.buildBroadcastShardsHeader;
 
-/**
- *
- */
 public class RestRefreshAction extends BaseRestHandler {
 
     @Inject
@@ -56,10 +54,10 @@ public class RestRefreshAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         RefreshRequest refreshRequest = new RefreshRequest(Strings.splitStringByCommaToArray(request.param("index")));
         refreshRequest.indicesOptions(IndicesOptions.fromRequest(request, refreshRequest.indicesOptions()));
-        client.admin().indices().refresh(refreshRequest, new RestBuilderListener<RefreshResponse>(channel) {
+        return channel -> client.admin().indices().refresh(refreshRequest, new RestBuilderListener<RefreshResponse>(channel) {
             @Override
             public RestResponse buildResponse(RefreshResponse response, XContentBuilder builder) throws Exception {
                 builder.startObject();
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestRolloverIndexAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestRolloverIndexAction.java
index 1433bc4257..cecfb31916 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestRolloverIndexAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestRolloverIndexAction.java
@@ -20,16 +20,17 @@
 package org.elasticsearch.rest.action.admin.indices;
 
 import org.elasticsearch.action.admin.indices.rollover.RolloverRequest;
-import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.action.support.ActiveShardCount;
+import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.RestToXContentListener;
 
+import java.io.IOException;
+
 /**
  *
  */
@@ -44,7 +45,7 @@ public class RestRolloverIndexAction extends BaseRestHandler {
 
     @SuppressWarnings({"unchecked"})
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         RolloverRequest rolloverIndexRequest = new RolloverRequest(request.param("index"), request.param("new_index"));
         if (request.hasContent()) {
             rolloverIndexRequest.source(request.content());
@@ -53,6 +54,6 @@ public class RestRolloverIndexAction extends BaseRestHandler {
         rolloverIndexRequest.timeout(request.paramAsTime("timeout", rolloverIndexRequest.timeout()));
         rolloverIndexRequest.masterNodeTimeout(request.paramAsTime("master_timeout", rolloverIndexRequest.masterNodeTimeout()));
         rolloverIndexRequest.setWaitForActiveShards(ActiveShardCount.parseString(request.param("wait_for_active_shards")));
-        client.admin().indices().rolloverIndex(rolloverIndexRequest, new RestToXContentListener<>(channel));
+        return channel -> client.admin().indices().rolloverIndex(rolloverIndexRequest, new RestToXContentListener<>(channel));
     }
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestShrinkIndexAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestShrinkIndexAction.java
index f04c9760a6..fb65825ac6 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestShrinkIndexAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestShrinkIndexAction.java
@@ -20,23 +20,19 @@
 package org.elasticsearch.rest.action.admin.indices;
 
 import org.elasticsearch.action.admin.indices.shrink.ShrinkRequest;
-import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.action.admin.indices.shrink.ShrinkResponse;
 import org.elasticsearch.action.support.ActiveShardCount;
+import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.AcknowledgedRestListener;
 
 import java.io.IOException;
 
-/**
- *
- */
 public class RestShrinkIndexAction extends BaseRestHandler {
 
     @Inject
@@ -48,7 +44,7 @@ public class RestShrinkIndexAction extends BaseRestHandler {
 
     @SuppressWarnings({"unchecked"})
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         if (request.param("target") == null) {
             throw new IllegalArgumentException("no target index");
         }
@@ -62,7 +58,7 @@ public class RestShrinkIndexAction extends BaseRestHandler {
         shrinkIndexRequest.timeout(request.paramAsTime("timeout", shrinkIndexRequest.timeout()));
         shrinkIndexRequest.masterNodeTimeout(request.paramAsTime("master_timeout", shrinkIndexRequest.masterNodeTimeout()));
         shrinkIndexRequest.setWaitForActiveShards(ActiveShardCount.parseString(request.param("wait_for_active_shards")));
-        client.admin().indices().shrinkIndex(shrinkIndexRequest, new AcknowledgedRestListener<ShrinkResponse>(channel) {
+        return channel -> client.admin().indices().shrinkIndex(shrinkIndexRequest, new AcknowledgedRestListener<ShrinkResponse>(channel) {
             @Override
             public void addCustomFields(XContentBuilder builder, ShrinkResponse response) throws IOException {
                 response.addCustomFields(builder);
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestSyncedFlushAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestSyncedFlushAction.java
index 784a588db8..194bf0f2f8 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestSyncedFlushAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestSyncedFlushAction.java
@@ -29,12 +29,13 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
 import org.elasticsearch.rest.action.RestBuilderListener;
 
+import java.io.IOException;
+
 import static org.elasticsearch.rest.RestRequest.Method.GET;
 import static org.elasticsearch.rest.RestRequest.Method.POST;
 
@@ -54,11 +55,11 @@ public class RestSyncedFlushAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         IndicesOptions indicesOptions = IndicesOptions.fromRequest(request, IndicesOptions.lenientExpandOpen());
         SyncedFlushRequest syncedFlushRequest = new SyncedFlushRequest(Strings.splitStringByCommaToArray(request.param("index")));
         syncedFlushRequest.indicesOptions(indicesOptions);
-        client.admin().indices().syncedFlush(syncedFlushRequest, new RestBuilderListener<SyncedFlushResponse>(channel) {
+        return channel -> client.admin().indices().syncedFlush(syncedFlushRequest, new RestBuilderListener<SyncedFlushResponse>(channel) {
             @Override
             public RestResponse buildResponse(SyncedFlushResponse results, XContentBuilder builder) throws Exception {
                 builder.startObject();
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestTypesExistsAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestTypesExistsAction.java
index 3877715395..799ba8b1c6 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestTypesExistsAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestTypesExistsAction.java
@@ -28,12 +28,13 @@ import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
 import org.elasticsearch.rest.action.RestResponseListener;
 
+import java.io.IOException;
+
 import static org.elasticsearch.rest.RestRequest.Method.HEAD;
 import static org.elasticsearch.rest.RestStatus.NOT_FOUND;
 import static org.elasticsearch.rest.RestStatus.OK;
@@ -52,13 +53,13 @@ public class RestTypesExistsAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         TypesExistsRequest typesExistsRequest = new TypesExistsRequest(
                 Strings.splitStringByCommaToArray(request.param("index")), Strings.splitStringByCommaToArray(request.param("type"))
         );
         typesExistsRequest.local(request.paramAsBoolean("local", typesExistsRequest.local()));
         typesExistsRequest.indicesOptions(IndicesOptions.fromRequest(request, typesExistsRequest.indicesOptions()));
-        client.admin().indices().typesExists(typesExistsRequest, new RestResponseListener<TypesExistsResponse>(channel) {
+        return channel -> client.admin().indices().typesExists(typesExistsRequest, new RestResponseListener<TypesExistsResponse>(channel) {
             @Override
             public RestResponse buildResponse(TypesExistsResponse response) throws Exception {
                 if (response.isExists()) {
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestUpdateSettingsAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestUpdateSettingsAction.java
index 0c1b535901..7ebbe1205e 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestUpdateSettingsAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestUpdateSettingsAction.java
@@ -26,11 +26,11 @@ import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.AcknowledgedRestListener;
 
+import java.io.IOException;
 import java.util.Map;
 import java.util.Set;
 
@@ -60,7 +60,7 @@ public class RestUpdateSettingsAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         UpdateSettingsRequest updateSettingsRequest = updateSettingsRequest(Strings.splitStringByCommaToArray(request.param("index")));
         updateSettingsRequest.timeout(request.paramAsTime("timeout", updateSettingsRequest.timeout()));
         updateSettingsRequest.setPreserveExisting(request.paramAsBoolean("preserve_existing", updateSettingsRequest.isPreserveExisting()));
@@ -89,6 +89,12 @@ public class RestUpdateSettingsAction extends BaseRestHandler {
         }
         updateSettingsRequest.settings(updateSettings);
 
-        client.admin().indices().updateSettings(updateSettingsRequest, new AcknowledgedRestListener<>(channel));
+        return channel -> client.admin().indices().updateSettings(updateSettingsRequest, new AcknowledgedRestListener<>(channel));
     }
+
+    @Override
+    protected Set<String> responseParams() {
+        return Settings.FORMAT_PARAMS;
+    }
+
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestUpgradeAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestUpgradeAction.java
index e0659e1cf5..afde577de1 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestUpgradeAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestUpgradeAction.java
@@ -31,12 +31,12 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
 import org.elasticsearch.rest.action.RestBuilderListener;
 
+import java.io.IOException;
 import java.util.Map;
 
 import static org.elasticsearch.rest.RestRequest.Method.GET;
@@ -44,7 +44,6 @@ import static org.elasticsearch.rest.RestRequest.Method.POST;
 import static org.elasticsearch.rest.RestStatus.OK;
 import static org.elasticsearch.rest.action.RestActions.buildBroadcastShardsHeader;
 
-
 public class RestUpgradeAction extends BaseRestHandler {
 
     @Inject
@@ -58,31 +57,33 @@ public class RestUpgradeAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(RestRequest request, RestChannel channel, NodeClient client) throws Exception {
+    public RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
         if (request.method().equals(RestRequest.Method.GET)) {
-            handleGet(request, channel, client);
+            return handleGet(request, client);
         } else if (request.method().equals(RestRequest.Method.POST)) {
-            handlePost(request, channel, client);
+            return handlePost(request, client);
+        } else {
+            throw new IllegalArgumentException("illegal method [" + request.method() + "] for request [" + request.path() + "]");
         }
     }
 
-    void handleGet(final RestRequest request, RestChannel channel, NodeClient client) {
-        client.admin().indices().prepareUpgradeStatus(Strings.splitStringByCommaToArray(request.param("index")))
-                .execute(new RestBuilderListener<UpgradeStatusResponse>(channel) {
-                    @Override
-                    public RestResponse buildResponse(UpgradeStatusResponse response, XContentBuilder builder) throws Exception {
-                        builder.startObject();
-                        response.toXContent(builder, request);
-                        builder.endObject();
-                        return new BytesRestResponse(OK, builder);
-                    }
-                });
+    private RestChannelConsumer handleGet(final RestRequest request, NodeClient client) {
+        return channel -> client.admin().indices().prepareUpgradeStatus(Strings.splitStringByCommaToArray(request.param("index")))
+            .execute(new RestBuilderListener<UpgradeStatusResponse>(channel) {
+                @Override
+                public RestResponse buildResponse(UpgradeStatusResponse response, XContentBuilder builder) throws Exception {
+                    builder.startObject();
+                    response.toXContent(builder, request);
+                    builder.endObject();
+                    return new BytesRestResponse(OK, builder);
+                }
+            });
     }
 
-    void handlePost(final RestRequest request, RestChannel channel, NodeClient client) {
+    private RestChannelConsumer handlePost(final RestRequest request, NodeClient client) {
         UpgradeRequest upgradeReq = new UpgradeRequest(Strings.splitStringByCommaToArray(request.param("index")));
         upgradeReq.upgradeOnlyAncientSegments(request.paramAsBoolean("only_ancient_segments", false));
-        client.admin().indices().upgrade(upgradeReq, new RestBuilderListener<UpgradeResponse>(channel) {
+        return channel -> client.admin().indices().upgrade(upgradeReq, new RestBuilderListener<UpgradeResponse>(channel) {
             @Override
             public RestResponse buildResponse(UpgradeResponse response, XContentBuilder builder) throws Exception {
                 builder.startObject();
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestValidateQueryAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestValidateQueryAction.java
index 7bf2a34ef6..3a1ea5fa7d 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestValidateQueryAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestValidateQueryAction.java
@@ -47,9 +47,6 @@ import static org.elasticsearch.rest.RestRequest.Method.POST;
 import static org.elasticsearch.rest.RestStatus.OK;
 import static org.elasticsearch.rest.action.RestActions.buildBroadcastShardsHeader;
 
-/**
- *
- */
 public class RestValidateQueryAction extends BaseRestHandler {
 
     private final IndicesQueriesRegistry indicesQueriesRegistry;
@@ -67,58 +64,69 @@ public class RestValidateQueryAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) throws Exception {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         ValidateQueryRequest validateQueryRequest = new ValidateQueryRequest(Strings.splitStringByCommaToArray(request.param("index")));
         validateQueryRequest.indicesOptions(IndicesOptions.fromRequest(request, validateQueryRequest.indicesOptions()));
         validateQueryRequest.explain(request.paramAsBoolean("explain", false));
+        validateQueryRequest.types(Strings.splitStringByCommaToArray(request.param("type")));
+        validateQueryRequest.rewrite(request.paramAsBoolean("rewrite", false));
+
+        Exception bodyParsingException = null;
         if (RestActions.hasBodyContent(request)) {
             try {
-                validateQueryRequest
-                        .query(RestActions.getQueryContent(RestActions.getRestContent(request), indicesQueriesRegistry, parseFieldMatcher));
-            } catch(ParsingException e) {
-                channel.sendResponse(buildErrorResponse(channel.newBuilder(), e.getDetailedMessage(), validateQueryRequest.explain()));
-                return;
-            } catch(Exception e) {
-                channel.sendResponse(buildErrorResponse(channel.newBuilder(), e.getMessage(), validateQueryRequest.explain()));
-                return;
+                validateQueryRequest.query(
+                    RestActions.getQueryContent(RestActions.getRestContent(request), indicesQueriesRegistry, parseFieldMatcher));
+            } catch (Exception e) {
+                bodyParsingException = e;
             }
-        } else {
+        } else if (request.hasParam("q")) {
             QueryBuilder queryBuilder = RestActions.urlParamsToQueryBuilder(request);
-            if (queryBuilder != null) {
-                validateQueryRequest.query(queryBuilder);
-            }
+            validateQueryRequest.query(queryBuilder);
         }
-        validateQueryRequest.types(Strings.splitStringByCommaToArray(request.param("type")));
-        validateQueryRequest.rewrite(request.paramAsBoolean("rewrite", false));
 
-        client.admin().indices().validateQuery(validateQueryRequest, new RestBuilderListener<ValidateQueryResponse>(channel) {
-            @Override
-            public RestResponse buildResponse(ValidateQueryResponse response, XContentBuilder builder) throws Exception {
-                builder.startObject();
-                builder.field(VALID_FIELD, response.isValid());
-                buildBroadcastShardsHeader(builder, request, response);
-                if (response.getQueryExplanation() != null && !response.getQueryExplanation().isEmpty()) {
-                    builder.startArray(EXPLANATIONS_FIELD);
-                    for (QueryExplanation explanation : response.getQueryExplanation()) {
+        final Exception finalBodyParsingException = bodyParsingException;
+        return channel -> {
+            if (finalBodyParsingException != null) {
+                if (finalBodyParsingException instanceof ParsingException) {
+                    handleException(validateQueryRequest, ((ParsingException) finalBodyParsingException).getDetailedMessage(), channel);
+                } else {
+                    handleException(validateQueryRequest, finalBodyParsingException.getMessage(), channel);
+                }
+            } else {
+                client.admin().indices().validateQuery(validateQueryRequest, new RestBuilderListener<ValidateQueryResponse>(channel) {
+                    @Override
+                    public RestResponse buildResponse(ValidateQueryResponse response, XContentBuilder builder) throws Exception {
                         builder.startObject();
-                        if (explanation.getIndex() != null) {
-                            builder.field(INDEX_FIELD, explanation.getIndex());
-                        }
-                        builder.field(VALID_FIELD, explanation.isValid());
-                        if (explanation.getError() != null) {
-                            builder.field(ERROR_FIELD, explanation.getError());
-                        }
-                        if (explanation.getExplanation() != null) {
-                            builder.field(EXPLANATION_FIELD, explanation.getExplanation());
+                        builder.field(VALID_FIELD, response.isValid());
+                        buildBroadcastShardsHeader(builder, request, response);
+                        if (response.getQueryExplanation() != null && !response.getQueryExplanation().isEmpty()) {
+                            builder.startArray(EXPLANATIONS_FIELD);
+                            for (QueryExplanation explanation : response.getQueryExplanation()) {
+                                builder.startObject();
+                                if (explanation.getIndex() != null) {
+                                    builder.field(INDEX_FIELD, explanation.getIndex());
+                                }
+                                builder.field(VALID_FIELD, explanation.isValid());
+                                if (explanation.getError() != null) {
+                                    builder.field(ERROR_FIELD, explanation.getError());
+                                }
+                                if (explanation.getExplanation() != null) {
+                                    builder.field(EXPLANATION_FIELD, explanation.getExplanation());
+                                }
+                                builder.endObject();
+                            }
+                            builder.endArray();
                         }
                         builder.endObject();
+                        return new BytesRestResponse(OK, builder);
                     }
-                    builder.endArray();
-                }
-                builder.endObject();
-                return new BytesRestResponse(OK, builder);
+                });
             }
-        });
+        };
+    }
+
+    private void handleException(final ValidateQueryRequest request, final String message, final RestChannel channel) throws IOException {
+        channel.sendResponse(buildErrorResponse(channel.newBuilder(), message, request.explain()));
     }
 
     private static BytesRestResponse buildErrorResponse(XContentBuilder builder, String error, boolean explain) throws IOException {
diff --git a/core/src/main/java/org/elasticsearch/rest/action/cat/AbstractCatAction.java b/core/src/main/java/org/elasticsearch/rest/action/cat/AbstractCatAction.java
index 8315e34d08..86739046eb 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/cat/AbstractCatAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/cat/AbstractCatAction.java
@@ -25,49 +25,62 @@ import org.elasticsearch.common.io.stream.BytesStreamOutput;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestStatus;
 
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+
 import static org.elasticsearch.rest.action.cat.RestTable.buildHelpWidths;
 import static org.elasticsearch.rest.action.cat.RestTable.pad;
 
-/**
- *
- */
 public abstract class AbstractCatAction extends BaseRestHandler {
 
     public AbstractCatAction(Settings settings) {
         super(settings);
     }
 
-    protected abstract void doRequest(final RestRequest request, final RestChannel channel, final NodeClient client);
+    protected abstract RestChannelConsumer doCatRequest(final RestRequest request, final NodeClient client);
 
     protected abstract void documentation(StringBuilder sb);
 
     protected abstract Table getTableWithHeader(final RestRequest request);
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) throws Exception {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         boolean helpWanted = request.paramAsBoolean("help", false);
         if (helpWanted) {
-            Table table = getTableWithHeader(request);
-            int[] width = buildHelpWidths(table, request);
-            BytesStreamOutput bytesOutput = channel.bytesOutput();
-            UTF8StreamWriter out = new UTF8StreamWriter().setOutput(bytesOutput);
-            for (Table.Cell cell : table.getHeaders()) {
-                // need to do left-align always, so create new cells
-                pad(new Table.Cell(cell.value), width[0], request, out);
-                out.append(" | ");
-                pad(new Table.Cell(cell.attr.containsKey("alias") ? cell.attr.get("alias") : ""), width[1], request, out);
-                out.append(" | ");
-                pad(new Table.Cell(cell.attr.containsKey("desc") ? cell.attr.get("desc") : "not available"), width[2], request, out);
-                out.append("\n");
-            }
-            out.close();
-            channel.sendResponse(new BytesRestResponse(RestStatus.OK, BytesRestResponse.TEXT_CONTENT_TYPE, bytesOutput.bytes()));
+            return channel -> {
+                Table table = getTableWithHeader(request);
+                int[] width = buildHelpWidths(table, request);
+                BytesStreamOutput bytesOutput = channel.bytesOutput();
+                UTF8StreamWriter out = new UTF8StreamWriter().setOutput(bytesOutput);
+                for (Table.Cell cell : table.getHeaders()) {
+                    // need to do left-align always, so create new cells
+                    pad(new Table.Cell(cell.value), width[0], request, out);
+                    out.append(" | ");
+                    pad(new Table.Cell(cell.attr.containsKey("alias") ? cell.attr.get("alias") : ""), width[1], request, out);
+                    out.append(" | ");
+                    pad(new Table.Cell(cell.attr.containsKey("desc") ? cell.attr.get("desc") : "not available"), width[2], request, out);
+                    out.append("\n");
+                }
+                out.close();
+                channel.sendResponse(new BytesRestResponse(RestStatus.OK, BytesRestResponse.TEXT_CONTENT_TYPE, bytesOutput.bytes()));
+            };
         } else {
-            doRequest(request, channel, client);
+            return doCatRequest(request, client);
         }
     }
+
+    static Set<String> RESPONSE_PARAMS =
+            Collections.unmodifiableSet(new HashSet<>(Arrays.asList("format", "h", "v", "ts", "pri", "bytes", "size", "time")));
+
+    @Override
+    protected Set<String> responseParams() {
+        return RESPONSE_PARAMS;
+    }
+
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/cat/RestAliasAction.java b/core/src/main/java/org/elasticsearch/rest/action/cat/RestAliasAction.java
index 82d59784fc..981b573f0d 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/cat/RestAliasAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/cat/RestAliasAction.java
@@ -27,7 +27,6 @@ import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.Table;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
@@ -37,9 +36,6 @@ import java.util.List;
 
 import static org.elasticsearch.rest.RestRequest.Method.GET;
 
-/**
- *
- */
 public class RestAliasAction extends AbstractCatAction {
 
     @Inject
@@ -51,13 +47,13 @@ public class RestAliasAction extends AbstractCatAction {
 
 
     @Override
-    protected void doRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    protected RestChannelConsumer doCatRequest(final RestRequest request, final NodeClient client) {
         final GetAliasesRequest getAliasesRequest = request.hasParam("alias") ?
                 new GetAliasesRequest(request.param("alias")) :
                 new GetAliasesRequest();
         getAliasesRequest.local(request.paramAsBoolean("local", getAliasesRequest.local()));
 
-        client.admin().indices().getAliases(getAliasesRequest, new RestResponseListener<GetAliasesResponse>(channel) {
+        return channel -> client.admin().indices().getAliases(getAliasesRequest, new RestResponseListener<GetAliasesResponse>(channel) {
             @Override
             public RestResponse buildResponse(GetAliasesResponse response) throws Exception {
                 Table tab = buildTable(request, response);
diff --git a/core/src/main/java/org/elasticsearch/rest/action/cat/RestAllocationAction.java b/core/src/main/java/org/elasticsearch/rest/action/cat/RestAllocationAction.java
index c813878088..86ff1c4480 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/cat/RestAllocationAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/cat/RestAllocationAction.java
@@ -34,7 +34,6 @@ import org.elasticsearch.common.Table;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.ByteSizeValue;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
@@ -59,14 +58,14 @@ public class RestAllocationAction extends AbstractCatAction {
     }
 
     @Override
-    public void doRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer doCatRequest(final RestRequest request, final NodeClient client) {
         final String[] nodes = Strings.splitStringByCommaToArray(request.param("nodes", "data:true"));
         final ClusterStateRequest clusterStateRequest = new ClusterStateRequest();
         clusterStateRequest.clear().routingTable(true);
         clusterStateRequest.local(request.paramAsBoolean("local", clusterStateRequest.local()));
         clusterStateRequest.masterNodeTimeout(request.paramAsTime("master_timeout", clusterStateRequest.masterNodeTimeout()));
 
-        client.admin().cluster().state(clusterStateRequest, new RestActionListener<ClusterStateResponse>(channel) {
+        return channel -> client.admin().cluster().state(clusterStateRequest, new RestActionListener<ClusterStateResponse>(channel) {
             @Override
             public void processResponse(final ClusterStateResponse state) {
                 NodesStatsRequest statsRequest = new NodesStatsRequest(nodes);
diff --git a/core/src/main/java/org/elasticsearch/rest/action/cat/RestCatAction.java b/core/src/main/java/org/elasticsearch/rest/action/cat/RestCatAction.java
index b9cc5011a8..c95b35a989 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/cat/RestCatAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/cat/RestCatAction.java
@@ -24,11 +24,11 @@ import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestStatus;
 
+import java.io.IOException;
 import java.util.Set;
 
 import static org.elasticsearch.rest.RestRequest.Method.GET;
@@ -52,7 +52,8 @@ public class RestCatAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
-        channel.sendResponse(new BytesRestResponse(RestStatus.OK, HELP));
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
+        return channel -> channel.sendResponse(new BytesRestResponse(RestStatus.OK, HELP));
     }
+
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/cat/RestCountAction.java b/core/src/main/java/org/elasticsearch/rest/action/cat/RestCountAction.java
index 4faddc3168..6af900ce51 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/cat/RestCountAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/cat/RestCountAction.java
@@ -29,7 +29,6 @@ import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.index.query.QueryBuilder;
 import org.elasticsearch.indices.query.IndicesQueriesRegistry;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
@@ -58,7 +57,7 @@ public class RestCountAction extends AbstractCatAction {
     }
 
     @Override
-    public void doRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer doCatRequest(final RestRequest request, final NodeClient client) {
         String[] indices = Strings.splitStringByCommaToArray(request.param("index"));
         SearchRequest countRequest = new SearchRequest(indices);
         String source = request.param("source");
@@ -72,7 +71,7 @@ public class RestCountAction extends AbstractCatAction {
                 searchSourceBuilder.query(queryBuilder);
             }
         }
-        client.search(countRequest, new RestResponseListener<SearchResponse>(channel) {
+        return channel -> client.search(countRequest, new RestResponseListener<SearchResponse>(channel) {
             @Override
             public RestResponse buildResponse(SearchResponse countResponse) throws Exception {
                 return RestTable.buildResponse(buildTable(request, countResponse), channel);
diff --git a/core/src/main/java/org/elasticsearch/rest/action/cat/RestFielddataAction.java b/core/src/main/java/org/elasticsearch/rest/action/cat/RestFielddataAction.java
index fcdad0c3f7..4ad4d8aca2 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/cat/RestFielddataAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/cat/RestFielddataAction.java
@@ -28,7 +28,6 @@ import org.elasticsearch.common.Table;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.ByteSizeValue;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
@@ -49,14 +48,14 @@ public class RestFielddataAction extends AbstractCatAction {
     }
 
     @Override
-    protected void doRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    protected RestChannelConsumer doCatRequest(final RestRequest request, final NodeClient client) {
         final NodesStatsRequest nodesStatsRequest = new NodesStatsRequest("data:true");
         nodesStatsRequest.clear();
         nodesStatsRequest.indices(true);
         String[] fields = request.paramAsStringArray("fields", null);
         nodesStatsRequest.indices().fieldDataFields(fields == null ? new String[] {"*"} : fields);
 
-        client.admin().cluster().nodesStats(nodesStatsRequest, new RestResponseListener<NodesStatsResponse>(channel) {
+        return channel -> client.admin().cluster().nodesStats(nodesStatsRequest, new RestResponseListener<NodesStatsResponse>(channel) {
             @Override
             public RestResponse buildResponse(NodesStatsResponse nodeStatses) throws Exception {
                 return RestTable.buildResponse(buildTable(request, nodeStatses), channel);
diff --git a/core/src/main/java/org/elasticsearch/rest/action/cat/RestHealthAction.java b/core/src/main/java/org/elasticsearch/rest/action/cat/RestHealthAction.java
index cd226e28b5..fc6462bcaa 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/cat/RestHealthAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/cat/RestHealthAction.java
@@ -25,7 +25,6 @@ import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.Table;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
@@ -49,10 +48,10 @@ public class RestHealthAction extends AbstractCatAction {
     }
 
     @Override
-    public void doRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer doCatRequest(final RestRequest request, final NodeClient client) {
         ClusterHealthRequest clusterHealthRequest = new ClusterHealthRequest();
 
-        client.admin().cluster().health(clusterHealthRequest, new RestResponseListener<ClusterHealthResponse>(channel) {
+        return channel -> client.admin().cluster().health(clusterHealthRequest, new RestResponseListener<ClusterHealthResponse>(channel) {
             @Override
             public RestResponse buildResponse(final ClusterHealthResponse health) throws Exception {
                 return RestTable.buildResponse(buildTable(health, request), channel);
diff --git a/core/src/main/java/org/elasticsearch/rest/action/cat/RestIndicesAction.java b/core/src/main/java/org/elasticsearch/rest/action/cat/RestIndicesAction.java
index 782c0ea444..e56347f16b 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/cat/RestIndicesAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/cat/RestIndicesAction.java
@@ -40,7 +40,6 @@ import org.elasticsearch.common.Table;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.index.Index;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
@@ -49,7 +48,11 @@ import org.elasticsearch.rest.action.RestResponseListener;
 import org.joda.time.DateTime;
 import org.joda.time.DateTimeZone;
 
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
 import java.util.Locale;
+import java.util.Set;
 
 import static org.elasticsearch.rest.RestRequest.Method.GET;
 
@@ -72,7 +75,7 @@ public class RestIndicesAction extends AbstractCatAction {
     }
 
     @Override
-    public void doRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer doCatRequest(final RestRequest request, final NodeClient client) {
         final String[] indices = Strings.splitStringByCommaToArray(request.param("index"));
         final ClusterStateRequest clusterStateRequest = new ClusterStateRequest();
         clusterStateRequest.clear().indices(indices).metaData(true);
@@ -81,7 +84,7 @@ public class RestIndicesAction extends AbstractCatAction {
         final IndicesOptions strictExpandIndicesOptions = IndicesOptions.strictExpand();
         clusterStateRequest.indicesOptions(strictExpandIndicesOptions);
 
-        client.admin().cluster().state(clusterStateRequest, new RestActionListener<ClusterStateResponse>(channel) {
+        return channel -> client.admin().cluster().state(clusterStateRequest, new RestActionListener<ClusterStateResponse>(channel) {
             @Override
             public void processResponse(final ClusterStateResponse clusterStateResponse) {
                 final ClusterState state = clusterStateResponse.getState();
@@ -123,6 +126,19 @@ public class RestIndicesAction extends AbstractCatAction {
         });
     }
 
+    private static final Set<String> RESPONSE_PARAMS;
+
+    static {
+        final Set<String> responseParams = new HashSet<>(Arrays.asList("local", "health"));
+        responseParams.addAll(AbstractCatAction.RESPONSE_PARAMS);
+        RESPONSE_PARAMS = Collections.unmodifiableSet(responseParams);
+    }
+
+    @Override
+    protected Set<String> responseParams() {
+        return RESPONSE_PARAMS;
+    }
+
     @Override
     protected Table getTableWithHeader(final RestRequest request) {
         Table table = new Table();
diff --git a/core/src/main/java/org/elasticsearch/rest/action/cat/RestMasterAction.java b/core/src/main/java/org/elasticsearch/rest/action/cat/RestMasterAction.java
index 5902ba60e5..be2aeaafd4 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/cat/RestMasterAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/cat/RestMasterAction.java
@@ -27,7 +27,6 @@ import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.common.Table;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
@@ -49,13 +48,13 @@ public class RestMasterAction extends AbstractCatAction {
     }
 
     @Override
-    public void doRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer doCatRequest(final RestRequest request, final NodeClient client) {
         final ClusterStateRequest clusterStateRequest = new ClusterStateRequest();
         clusterStateRequest.clear().nodes(true);
         clusterStateRequest.local(request.paramAsBoolean("local", clusterStateRequest.local()));
         clusterStateRequest.masterNodeTimeout(request.paramAsTime("master_timeout", clusterStateRequest.masterNodeTimeout()));
 
-        client.admin().cluster().state(clusterStateRequest, new RestResponseListener<ClusterStateResponse>(channel) {
+        return channel -> client.admin().cluster().state(clusterStateRequest, new RestResponseListener<ClusterStateResponse>(channel) {
             @Override
             public RestResponse buildResponse(final ClusterStateResponse clusterStateResponse) throws Exception {
                 return RestTable.buildResponse(buildTable(request, clusterStateResponse), channel);
diff --git a/core/src/main/java/org/elasticsearch/rest/action/cat/RestNodeAttrsAction.java b/core/src/main/java/org/elasticsearch/rest/action/cat/RestNodeAttrsAction.java
index 5ab98316c7..37c5cf26e4 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/cat/RestNodeAttrsAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/cat/RestNodeAttrsAction.java
@@ -32,7 +32,6 @@ import org.elasticsearch.common.Table;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.transport.InetSocketTransportAddress;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
@@ -57,13 +56,13 @@ public class RestNodeAttrsAction extends AbstractCatAction {
     }
 
     @Override
-    public void doRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer doCatRequest(final RestRequest request, final NodeClient client) {
         final ClusterStateRequest clusterStateRequest = new ClusterStateRequest();
         clusterStateRequest.clear().nodes(true);
         clusterStateRequest.local(request.paramAsBoolean("local", clusterStateRequest.local()));
         clusterStateRequest.masterNodeTimeout(request.paramAsTime("master_timeout", clusterStateRequest.masterNodeTimeout()));
 
-        client.admin().cluster().state(clusterStateRequest, new RestActionListener<ClusterStateResponse>(channel) {
+        return channel -> client.admin().cluster().state(clusterStateRequest, new RestActionListener<ClusterStateResponse>(channel) {
             @Override
             public void processResponse(final ClusterStateResponse clusterStateResponse) {
                 NodesInfoRequest nodesInfoRequest = new NodesInfoRequest();
diff --git a/core/src/main/java/org/elasticsearch/rest/action/cat/RestNodesAction.java b/core/src/main/java/org/elasticsearch/rest/action/cat/RestNodesAction.java
index 2c1900feef..d428b9db9c 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/cat/RestNodesAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/cat/RestNodesAction.java
@@ -54,7 +54,6 @@ import org.elasticsearch.monitor.jvm.JvmInfo;
 import org.elasticsearch.monitor.jvm.JvmStats;
 import org.elasticsearch.monitor.os.OsStats;
 import org.elasticsearch.monitor.process.ProcessStats;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
@@ -82,13 +81,13 @@ public class RestNodesAction extends AbstractCatAction {
     }
 
     @Override
-    public void doRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer doCatRequest(final RestRequest request, final NodeClient client) {
         final ClusterStateRequest clusterStateRequest = new ClusterStateRequest();
         clusterStateRequest.clear().nodes(true);
         clusterStateRequest.local(request.paramAsBoolean("local", clusterStateRequest.local()));
         clusterStateRequest.masterNodeTimeout(request.paramAsTime("master_timeout", clusterStateRequest.masterNodeTimeout()));
 
-        client.admin().cluster().state(clusterStateRequest, new RestActionListener<ClusterStateResponse>(channel) {
+        return channel -> client.admin().cluster().state(clusterStateRequest, new RestActionListener<ClusterStateResponse>(channel) {
             @Override
             public void processResponse(final ClusterStateResponse clusterStateResponse) {
                 NodesInfoRequest nodesInfoRequest = new NodesInfoRequest();
diff --git a/core/src/main/java/org/elasticsearch/rest/action/cat/RestPendingClusterTasksAction.java b/core/src/main/java/org/elasticsearch/rest/action/cat/RestPendingClusterTasksAction.java
index 773c6d292b..aa92164506 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/cat/RestPendingClusterTasksAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/cat/RestPendingClusterTasksAction.java
@@ -26,7 +26,6 @@ import org.elasticsearch.cluster.service.PendingClusterTask;
 import org.elasticsearch.common.Table;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
@@ -47,17 +46,20 @@ public class RestPendingClusterTasksAction extends AbstractCatAction {
     }
 
     @Override
-    public void doRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer doCatRequest(final RestRequest request, final NodeClient client) {
         PendingClusterTasksRequest pendingClusterTasksRequest = new PendingClusterTasksRequest();
         pendingClusterTasksRequest.masterNodeTimeout(request.paramAsTime("master_timeout", pendingClusterTasksRequest.masterNodeTimeout()));
         pendingClusterTasksRequest.local(request.paramAsBoolean("local", pendingClusterTasksRequest.local()));
-        client.admin().cluster().pendingClusterTasks(pendingClusterTasksRequest, new RestResponseListener<PendingClusterTasksResponse>(channel) {
-            @Override
-            public RestResponse buildResponse(PendingClusterTasksResponse pendingClusterTasks) throws Exception {
-                Table tab = buildTable(request, pendingClusterTasks);
-                return RestTable.buildResponse(tab, channel);
-            }
-        });
+        return channel ->
+                client.admin()
+                        .cluster()
+                        .pendingClusterTasks(pendingClusterTasksRequest, new RestResponseListener<PendingClusterTasksResponse>(channel) {
+                            @Override
+                            public RestResponse buildResponse(PendingClusterTasksResponse pendingClusterTasks) throws Exception {
+                                Table tab = buildTable(request, pendingClusterTasks);
+                                return RestTable.buildResponse(tab, channel);
+                            }
+                        });
     }
 
     @Override
diff --git a/core/src/main/java/org/elasticsearch/rest/action/cat/RestPluginsAction.java b/core/src/main/java/org/elasticsearch/rest/action/cat/RestPluginsAction.java
index ef8385653f..b729c4879f 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/cat/RestPluginsAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/cat/RestPluginsAction.java
@@ -31,7 +31,6 @@ import org.elasticsearch.common.Table;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.plugins.PluginInfo;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
@@ -54,13 +53,13 @@ public class RestPluginsAction extends AbstractCatAction {
     }
 
     @Override
-    public void doRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer doCatRequest(final RestRequest request, final NodeClient client) {
         final ClusterStateRequest clusterStateRequest = new ClusterStateRequest();
         clusterStateRequest.clear().nodes(true);
         clusterStateRequest.local(request.paramAsBoolean("local", clusterStateRequest.local()));
         clusterStateRequest.masterNodeTimeout(request.paramAsTime("master_timeout", clusterStateRequest.masterNodeTimeout()));
 
-        client.admin().cluster().state(clusterStateRequest, new RestActionListener<ClusterStateResponse>(channel) {
+        return channel -> client.admin().cluster().state(clusterStateRequest, new RestActionListener<ClusterStateResponse>(channel) {
             @Override
             public void processResponse(final ClusterStateResponse clusterStateResponse) throws Exception {
                 NodesInfoRequest nodesInfoRequest = new NodesInfoRequest();
diff --git a/core/src/main/java/org/elasticsearch/rest/action/cat/RestRecoveryAction.java b/core/src/main/java/org/elasticsearch/rest/action/cat/RestRecoveryAction.java
index b0ab8db8b2..612c2aaf42 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/cat/RestRecoveryAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/cat/RestRecoveryAction.java
@@ -32,7 +32,6 @@ import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.indices.recovery.RecoveryState;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
@@ -65,13 +64,13 @@ public class RestRecoveryAction extends AbstractCatAction {
     }
 
     @Override
-    public void doRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer doCatRequest(final RestRequest request, final NodeClient client) {
         final RecoveryRequest recoveryRequest = new RecoveryRequest(Strings.splitStringByCommaToArray(request.param("index")));
         recoveryRequest.detailed(request.paramAsBoolean("detailed", false));
         recoveryRequest.activeOnly(request.paramAsBoolean("active_only", false));
         recoveryRequest.indicesOptions(IndicesOptions.fromRequest(request, recoveryRequest.indicesOptions()));
 
-        client.admin().indices().recoveries(recoveryRequest, new RestResponseListener<RecoveryResponse>(channel) {
+        return channel -> client.admin().indices().recoveries(recoveryRequest, new RestResponseListener<RecoveryResponse>(channel) {
             @Override
             public RestResponse buildResponse(final RecoveryResponse response) throws Exception {
                 return RestTable.buildResponse(buildRecoveryTable(request, response), channel);
diff --git a/core/src/main/java/org/elasticsearch/rest/action/cat/RestRepositoriesAction.java b/core/src/main/java/org/elasticsearch/rest/action/cat/RestRepositoriesAction.java
index 05130504e5..71aab69ad4 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/cat/RestRepositoriesAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/cat/RestRepositoriesAction.java
@@ -26,7 +26,6 @@ import org.elasticsearch.cluster.metadata.RepositoryMetaData;
 import org.elasticsearch.common.Table;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
@@ -45,17 +44,20 @@ public class RestRepositoriesAction extends AbstractCatAction {
     }
 
     @Override
-    protected void doRequest(RestRequest request, RestChannel channel, NodeClient client) {
+    protected RestChannelConsumer doCatRequest(RestRequest request, NodeClient client) {
         GetRepositoriesRequest getRepositoriesRequest = new GetRepositoriesRequest();
         getRepositoriesRequest.local(request.paramAsBoolean("local", getRepositoriesRequest.local()));
         getRepositoriesRequest.masterNodeTimeout(request.paramAsTime("master_timeout", getRepositoriesRequest.masterNodeTimeout()));
 
-        client.admin().cluster().getRepositories(getRepositoriesRequest, new RestResponseListener<GetRepositoriesResponse>(channel) {
-            @Override
-            public RestResponse buildResponse(GetRepositoriesResponse getRepositoriesResponse) throws Exception {
-                return RestTable.buildResponse(buildTable(request, getRepositoriesResponse), channel);
-            }
-        });
+        return channel ->
+                client.admin()
+                        .cluster()
+                        .getRepositories(getRepositoriesRequest, new RestResponseListener<GetRepositoriesResponse>(channel) {
+                            @Override
+                            public RestResponse buildResponse(GetRepositoriesResponse getRepositoriesResponse) throws Exception {
+                                return RestTable.buildResponse(buildTable(request, getRepositoriesResponse), channel);
+                            }
+                        });
     }
 
     @Override
diff --git a/core/src/main/java/org/elasticsearch/rest/action/cat/RestSegmentsAction.java b/core/src/main/java/org/elasticsearch/rest/action/cat/RestSegmentsAction.java
index d2b30c49ca..6b665d957d 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/cat/RestSegmentsAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/cat/RestSegmentsAction.java
@@ -33,7 +33,6 @@ import org.elasticsearch.common.Table;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.index.engine.Segment;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
@@ -55,7 +54,7 @@ public class RestSegmentsAction extends AbstractCatAction {
     }
 
     @Override
-    protected void doRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    protected RestChannelConsumer doCatRequest(final RestRequest request, final NodeClient client) {
         final String[] indices = Strings.splitStringByCommaToArray(request.param("index"));
 
         final ClusterStateRequest clusterStateRequest = new ClusterStateRequest();
@@ -63,7 +62,7 @@ public class RestSegmentsAction extends AbstractCatAction {
         clusterStateRequest.masterNodeTimeout(request.paramAsTime("master_timeout", clusterStateRequest.masterNodeTimeout()));
         clusterStateRequest.clear().nodes(true).routingTable(true).indices(indices);
 
-        client.admin().cluster().state(clusterStateRequest, new RestActionListener<ClusterStateResponse>(channel) {
+        return channel -> client.admin().cluster().state(clusterStateRequest, new RestActionListener<ClusterStateResponse>(channel) {
             @Override
             public void processResponse(final ClusterStateResponse clusterStateResponse) {
                 final IndicesSegmentsRequest indicesSegmentsRequest = new IndicesSegmentsRequest();
diff --git a/core/src/main/java/org/elasticsearch/rest/action/cat/RestShardsAction.java b/core/src/main/java/org/elasticsearch/rest/action/cat/RestShardsAction.java
index 2dd4b6a10d..295f08ffe2 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/cat/RestShardsAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/cat/RestShardsAction.java
@@ -36,7 +36,6 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.index.engine.CommitStats;
 import org.elasticsearch.index.engine.Engine;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
@@ -63,13 +62,13 @@ public class RestShardsAction extends AbstractCatAction {
     }
 
     @Override
-    public void doRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer doCatRequest(final RestRequest request, final NodeClient client) {
         final String[] indices = Strings.splitStringByCommaToArray(request.param("index"));
         final ClusterStateRequest clusterStateRequest = new ClusterStateRequest();
         clusterStateRequest.local(request.paramAsBoolean("local", clusterStateRequest.local()));
         clusterStateRequest.masterNodeTimeout(request.paramAsTime("master_timeout", clusterStateRequest.masterNodeTimeout()));
         clusterStateRequest.clear().nodes(true).metaData(true).routingTable(true).indices(indices);
-        client.admin().cluster().state(clusterStateRequest, new RestActionListener<ClusterStateResponse>(channel) {
+        return channel -> client.admin().cluster().state(clusterStateRequest, new RestActionListener<ClusterStateResponse>(channel) {
             @Override
             public void processResponse(final ClusterStateResponse clusterStateResponse) {
                 IndicesStatsRequest indicesStatsRequest = new IndicesStatsRequest();
diff --git a/core/src/main/java/org/elasticsearch/rest/action/cat/RestSnapshotAction.java b/core/src/main/java/org/elasticsearch/rest/action/cat/RestSnapshotAction.java
index 021b00be24..ead3696281 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/cat/RestSnapshotAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/cat/RestSnapshotAction.java
@@ -27,7 +27,6 @@ import org.elasticsearch.common.Table;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
@@ -48,11 +47,12 @@ public class RestSnapshotAction extends AbstractCatAction {
     @Inject
     public RestSnapshotAction(Settings settings, RestController controller) {
         super(settings);
+        controller.registerHandler(GET, "/_cat/snapshots", this);
         controller.registerHandler(GET, "/_cat/snapshots/{repository}", this);
     }
 
     @Override
-    protected void doRequest(final RestRequest request, RestChannel channel, NodeClient client) {
+    protected RestChannelConsumer doCatRequest(final RestRequest request, NodeClient client) {
         GetSnapshotsRequest getSnapshotsRequest = new GetSnapshotsRequest()
                 .repository(request.param("repository"))
                 .snapshots(new String[]{GetSnapshotsRequest.ALL_SNAPSHOTS});
@@ -61,12 +61,13 @@ public class RestSnapshotAction extends AbstractCatAction {
 
         getSnapshotsRequest.masterNodeTimeout(request.paramAsTime("master_timeout", getSnapshotsRequest.masterNodeTimeout()));
 
-        client.admin().cluster().getSnapshots(getSnapshotsRequest, new RestResponseListener<GetSnapshotsResponse>(channel) {
-            @Override
-            public RestResponse buildResponse(GetSnapshotsResponse getSnapshotsResponse) throws Exception {
-                return RestTable.buildResponse(buildTable(request, getSnapshotsResponse), channel);
-            }
-        });
+        return channel ->
+            client.admin().cluster().getSnapshots(getSnapshotsRequest, new RestResponseListener<GetSnapshotsResponse>(channel) {
+                @Override
+                public RestResponse buildResponse(GetSnapshotsResponse getSnapshotsResponse) throws Exception {
+                    return RestTable.buildResponse(buildTable(request, getSnapshotsResponse), channel);
+                }
+            });
     }
 
     @Override
diff --git a/core/src/main/java/org/elasticsearch/rest/action/cat/RestTable.java b/core/src/main/java/org/elasticsearch/rest/action/cat/RestTable.java
index 5fad57da66..61932ad144 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/cat/RestTable.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/cat/RestTable.java
@@ -42,8 +42,6 @@ import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Set;
 
-/**
- */
 public class RestTable {
 
     public static RestResponse buildResponse(Table table, RestChannel channel) throws Exception {
diff --git a/core/src/main/java/org/elasticsearch/rest/action/cat/RestTasksAction.java b/core/src/main/java/org/elasticsearch/rest/action/cat/RestTasksAction.java
index 99eee4e735..a2bd220239 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/cat/RestTasksAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/cat/RestTasksAction.java
@@ -31,7 +31,6 @@ import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.transport.InetSocketTransportAddress;
 import org.elasticsearch.common.unit.TimeValue;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
@@ -41,7 +40,10 @@ import org.joda.time.format.DateTimeFormat;
 import org.joda.time.format.DateTimeFormatter;
 
 import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Set;
 
 import static org.elasticsearch.rest.RestRequest.Method.GET;
 import static org.elasticsearch.rest.action.admin.cluster.RestListTasksAction.generateListTasksRequest;
@@ -62,8 +64,9 @@ public class RestTasksAction extends AbstractCatAction {
     }
 
     @Override
-    public void doRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
-        client.admin().cluster().listTasks(generateListTasksRequest(request), new RestResponseListener<ListTasksResponse>(channel) {
+    public RestChannelConsumer doCatRequest(final RestRequest request, final NodeClient client) {
+        return channel ->
+                client.admin().cluster().listTasks(generateListTasksRequest(request), new RestResponseListener<ListTasksResponse>(channel) {
             @Override
             public RestResponse buildResponse(ListTasksResponse listTasksResponse) throws Exception {
                 return RestTable.buildResponse(buildTable(request, listTasksResponse), channel);
@@ -71,6 +74,20 @@ public class RestTasksAction extends AbstractCatAction {
         });
     }
 
+    private static final Set<String> RESPONSE_PARAMS;
+
+    static {
+        final Set<String> responseParams = new HashSet<>();
+        responseParams.add("detailed");
+        responseParams.addAll(AbstractCatAction.RESPONSE_PARAMS);
+        RESPONSE_PARAMS = Collections.unmodifiableSet(responseParams);
+    }
+
+    @Override
+    protected Set<String> responseParams() {
+        return RESPONSE_PARAMS;
+    }
+
     @Override
     protected Table getTableWithHeader(final RestRequest request) {
         boolean detailed = request.paramAsBoolean("detailed", false);
diff --git a/core/src/main/java/org/elasticsearch/rest/action/cat/RestTemplatesAction.java b/core/src/main/java/org/elasticsearch/rest/action/cat/RestTemplatesAction.java
index 5ee92cbb76..b62009512a 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/cat/RestTemplatesAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/cat/RestTemplatesAction.java
@@ -29,7 +29,6 @@ import org.elasticsearch.common.Table;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.regex.Regex;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
@@ -51,14 +50,14 @@ public class RestTemplatesAction extends AbstractCatAction {
     }
 
     @Override
-    protected void doRequest(final RestRequest request, RestChannel channel, NodeClient client) {
+    protected RestChannelConsumer doCatRequest(final RestRequest request, NodeClient client) {
         final String matchPattern = request.hasParam("name") ? request.param("name") : null;
         final ClusterStateRequest clusterStateRequest = new ClusterStateRequest();
         clusterStateRequest.clear().metaData(true);
         clusterStateRequest.local(request.paramAsBoolean("local", clusterStateRequest.local()));
         clusterStateRequest.masterNodeTimeout(request.paramAsTime("master_timeout", clusterStateRequest.masterNodeTimeout()));
 
-        client.admin().cluster().state(clusterStateRequest, new RestResponseListener<ClusterStateResponse>(channel) {
+        return channel -> client.admin().cluster().state(clusterStateRequest, new RestResponseListener<ClusterStateResponse>(channel) {
             @Override
             public RestResponse buildResponse(ClusterStateResponse clusterStateResponse) throws Exception {
                 return RestTable.buildResponse(buildTable(request, clusterStateResponse, matchPattern), channel);
diff --git a/core/src/main/java/org/elasticsearch/rest/action/cat/RestThreadPoolAction.java b/core/src/main/java/org/elasticsearch/rest/action/cat/RestThreadPoolAction.java
index 6f3c5c11ce..ab77ded35e 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/cat/RestThreadPoolAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/cat/RestThreadPoolAction.java
@@ -35,7 +35,6 @@ import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.regex.Regex;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.transport.InetSocketTransportAddress;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
@@ -69,13 +68,13 @@ public class RestThreadPoolAction extends AbstractCatAction {
     }
 
     @Override
-    public void doRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer doCatRequest(final RestRequest request, final NodeClient client) {
         final ClusterStateRequest clusterStateRequest = new ClusterStateRequest();
         clusterStateRequest.clear().nodes(true);
         clusterStateRequest.local(request.paramAsBoolean("local", clusterStateRequest.local()));
         clusterStateRequest.masterNodeTimeout(request.paramAsTime("master_timeout", clusterStateRequest.masterNodeTimeout()));
 
-        client.admin().cluster().state(clusterStateRequest, new RestActionListener<ClusterStateResponse>(channel) {
+        return channel -> client.admin().cluster().state(clusterStateRequest, new RestActionListener<ClusterStateResponse>(channel) {
             @Override
             public void processResponse(final ClusterStateResponse clusterStateResponse) {
                 NodesInfoRequest nodesInfoRequest = new NodesInfoRequest();
@@ -97,6 +96,20 @@ public class RestThreadPoolAction extends AbstractCatAction {
         });
     }
 
+    private static final Set<String> RESPONSE_PARAMS;
+
+    static {
+        final Set<String> responseParams = new HashSet<>();
+        responseParams.addAll(AbstractCatAction.RESPONSE_PARAMS);
+        responseParams.add("thread_pool_patterns");
+        RESPONSE_PARAMS = Collections.unmodifiableSet(responseParams);
+    }
+
+    @Override
+    protected Set<String> responseParams() {
+        return RESPONSE_PARAMS;
+    }
+
     @Override
     protected Table getTableWithHeader(final RestRequest request) {
         final Table table = new Table();
diff --git a/core/src/main/java/org/elasticsearch/rest/action/document/RestBulkAction.java b/core/src/main/java/org/elasticsearch/rest/action/document/RestBulkAction.java
index 4d5d3543cb..1801ddf2a4 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/document/RestBulkAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/document/RestBulkAction.java
@@ -34,13 +34,14 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
 import org.elasticsearch.rest.action.RestBuilderListener;
 import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
 
+import java.io.IOException;
+
 import static org.elasticsearch.rest.RestRequest.Method.POST;
 import static org.elasticsearch.rest.RestRequest.Method.PUT;
 import static org.elasticsearch.rest.RestStatus.OK;
@@ -75,7 +76,7 @@ public class RestBulkAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) throws Exception {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         BulkRequest bulkRequest = Requests.bulkRequest();
         String defaultIndex = request.param("index");
         String defaultType = request.param("type");
@@ -96,7 +97,7 @@ public class RestBulkAction extends BaseRestHandler {
         bulkRequest.add(request.content(), defaultIndex, defaultType, defaultRouting, defaultFields,
             defaultFetchSourceContext, defaultPipeline, null, allowExplicitIndex);
 
-        client.bulk(bulkRequest, new RestBuilderListener<BulkResponse>(channel) {
+        return channel -> client.bulk(bulkRequest, new RestBuilderListener<BulkResponse>(channel) {
             @Override
             public RestResponse buildResponse(BulkResponse response, XContentBuilder builder) throws Exception {
                 builder.startObject();
diff --git a/core/src/main/java/org/elasticsearch/rest/action/document/RestCountAction.java b/core/src/main/java/org/elasticsearch/rest/action/document/RestCountAction.java
index e5ca6f2cad..86cf0d8d90 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/document/RestCountAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/document/RestCountAction.java
@@ -32,7 +32,6 @@ import org.elasticsearch.index.query.QueryBuilder;
 import org.elasticsearch.indices.query.IndicesQueriesRegistry;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
@@ -40,14 +39,13 @@ import org.elasticsearch.rest.action.RestActions;
 import org.elasticsearch.rest.action.RestBuilderListener;
 import org.elasticsearch.search.builder.SearchSourceBuilder;
 
+import java.io.IOException;
+
 import static org.elasticsearch.rest.RestRequest.Method.GET;
 import static org.elasticsearch.rest.RestRequest.Method.POST;
 import static org.elasticsearch.rest.action.RestActions.buildBroadcastShardsHeader;
 import static org.elasticsearch.search.internal.SearchContext.DEFAULT_TERMINATE_AFTER;
 
-/**
- *
- */
 public class RestCountAction extends BaseRestHandler {
 
     private final IndicesQueriesRegistry indicesQueriesRegistry;
@@ -65,7 +63,7 @@ public class RestCountAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         SearchRequest countRequest = new SearchRequest(Strings.splitStringByCommaToArray(request.param("index")));
         countRequest.indicesOptions(IndicesOptions.fromRequest(request, countRequest.indicesOptions()));
         SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder().size(0);
@@ -93,7 +91,7 @@ public class RestCountAction extends BaseRestHandler {
         } else if (terminateAfter > 0) {
             searchSourceBuilder.terminateAfter(terminateAfter);
         }
-        client.search(countRequest, new RestBuilderListener<SearchResponse>(channel) {
+        return channel -> client.search(countRequest, new RestBuilderListener<SearchResponse>(channel) {
             @Override
             public RestResponse buildResponse(SearchResponse response, XContentBuilder builder) throws Exception {
                 builder.startObject();
@@ -109,4 +107,5 @@ public class RestCountAction extends BaseRestHandler {
             }
         });
     }
+
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/document/RestDeleteAction.java b/core/src/main/java/org/elasticsearch/rest/action/document/RestDeleteAction.java
index 392cff7ffb..6478b79966 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/document/RestDeleteAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/document/RestDeleteAction.java
@@ -26,12 +26,13 @@ import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.index.VersionType;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.RestActions;
 import org.elasticsearch.rest.action.RestStatusToXContentListener;
 
+import java.io.IOException;
+
 import static org.elasticsearch.rest.RestRequest.Method.DELETE;
 
 /**
@@ -46,7 +47,7 @@ public class RestDeleteAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         DeleteRequest deleteRequest = new DeleteRequest(request.param("index"), request.param("type"), request.param("id"));
         deleteRequest.routing(request.param("routing"));
         deleteRequest.parent(request.param("parent")); // order is important, set it after routing, so it will set the routing
@@ -60,6 +61,6 @@ public class RestDeleteAction extends BaseRestHandler {
             deleteRequest.waitForActiveShards(ActiveShardCount.parseString(waitForActiveShards));
         }
 
-        client.delete(deleteRequest, new RestStatusToXContentListener<>(channel));
+        return channel -> client.delete(deleteRequest, new RestStatusToXContentListener<>(channel));
     }
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/document/RestGetAction.java b/core/src/main/java/org/elasticsearch/rest/action/document/RestGetAction.java
index 550abb3e3b..d6590b89f8 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/document/RestGetAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/document/RestGetAction.java
@@ -29,7 +29,6 @@ import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.index.VersionType;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
@@ -37,6 +36,8 @@ import org.elasticsearch.rest.action.RestActions;
 import org.elasticsearch.rest.action.RestBuilderListener;
 import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
 
+import java.io.IOException;
+
 import static org.elasticsearch.rest.RestRequest.Method.GET;
 import static org.elasticsearch.rest.RestStatus.NOT_FOUND;
 import static org.elasticsearch.rest.RestStatus.OK;
@@ -50,7 +51,7 @@ public class RestGetAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         final GetRequest getRequest = new GetRequest(request.param("index"), request.param("type"), request.param("id"));
         getRequest.operationThreaded(true);
         getRequest.refresh(request.paramAsBoolean("refresh", getRequest.refresh()));
@@ -75,7 +76,7 @@ public class RestGetAction extends BaseRestHandler {
 
         getRequest.fetchSourceContext(FetchSourceContext.parseFromRestRequest(request));
 
-        client.get(getRequest, new RestBuilderListener<GetResponse>(channel) {
+        return channel -> client.get(getRequest, new RestBuilderListener<GetResponse>(channel) {
             @Override
             public RestResponse buildResponse(GetResponse response, XContentBuilder builder) throws Exception {
                 builder.startObject();
diff --git a/core/src/main/java/org/elasticsearch/rest/action/document/RestGetSourceAction.java b/core/src/main/java/org/elasticsearch/rest/action/document/RestGetSourceAction.java
index 1ecfe317f4..ed36cc3c89 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/document/RestGetSourceAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/document/RestGetSourceAction.java
@@ -28,7 +28,6 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
@@ -50,7 +49,7 @@ public class RestGetSourceAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         final GetRequest getRequest = new GetRequest(request.param("index"), request.param("type"), request.param("id"));
         getRequest.operationThreaded(true);
         getRequest.refresh(request.paramAsBoolean("refresh", getRequest.refresh()));
@@ -61,27 +60,25 @@ public class RestGetSourceAction extends BaseRestHandler {
 
         getRequest.fetchSourceContext(FetchSourceContext.parseFromRestRequest(request));
 
-        if (getRequest.fetchSourceContext() != null && !getRequest.fetchSourceContext().fetchSource()) {
-            try {
+        return channel -> {
+            if (getRequest.fetchSourceContext() != null && !getRequest.fetchSourceContext().fetchSource()) {
                 ActionRequestValidationException validationError = new ActionRequestValidationException();
                 validationError.addValidationError("fetching source can not be disabled");
                 channel.sendResponse(new BytesRestResponse(channel, validationError));
-            } catch (IOException e) {
-                logger.error("Failed to send failure response", e);
+            } else {
+                client.get(getRequest, new RestResponseListener<GetResponse>(channel) {
+                    @Override
+                    public RestResponse buildResponse(GetResponse response) throws Exception {
+                        XContentBuilder builder = channel.newBuilder(response.getSourceInternal(), false);
+                        if (response.isSourceEmpty()) { // check if doc source (or doc itself) is missing
+                            return new BytesRestResponse(NOT_FOUND, builder);
+                        } else {
+                            builder.rawValue(response.getSourceInternal());
+                            return new BytesRestResponse(OK, builder);
+                        }
+                    }
+                });
             }
-        }
-
-        client.get(getRequest, new RestResponseListener<GetResponse>(channel) {
-            @Override
-            public RestResponse buildResponse(GetResponse response) throws Exception {
-                XContentBuilder builder = channel.newBuilder(response.getSourceInternal(), false);
-                if (response.isSourceEmpty()) { // check if doc source (or doc itself) is missing
-                    return new BytesRestResponse(NOT_FOUND, builder);
-                } else {
-                    builder.rawValue(response.getSourceInternal());
-                    return new BytesRestResponse(OK, builder);
-                }
-            }
-        });
+        };
     }
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/document/RestHeadAction.java b/core/src/main/java/org/elasticsearch/rest/action/document/RestHeadAction.java
index ad2f826e58..0adb4fcfd2 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/document/RestHeadAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/document/RestHeadAction.java
@@ -28,12 +28,13 @@ import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
 import org.elasticsearch.rest.action.RestResponseListener;
 
+import java.io.IOException;
+
 import static org.elasticsearch.rest.RestRequest.Method.HEAD;
 import static org.elasticsearch.rest.RestStatus.NOT_FOUND;
 import static org.elasticsearch.rest.RestStatus.OK;
@@ -82,7 +83,7 @@ public abstract class RestHeadAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         final GetRequest getRequest = new GetRequest(request.param("index"), request.param("type"), request.param("id"));
         getRequest.operationThreaded(true);
         getRequest.refresh(request.paramAsBoolean("refresh", getRequest.refresh()));
@@ -94,7 +95,7 @@ public abstract class RestHeadAction extends BaseRestHandler {
         getRequest.storedFields(Strings.EMPTY_ARRAY);
         // TODO we can also just return the document size as Content-Length
 
-        client.get(getRequest, new RestResponseListener<GetResponse>(channel) {
+        return channel -> client.get(getRequest, new RestResponseListener<GetResponse>(channel) {
             @Override
             public RestResponse buildResponse(GetResponse response) {
                 if (!response.isExists()) {
@@ -107,4 +108,5 @@ public abstract class RestHeadAction extends BaseRestHandler {
             }
         });
     }
+
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/document/RestIndexAction.java b/core/src/main/java/org/elasticsearch/rest/action/document/RestIndexAction.java
index 6c9723b5b9..a3f211ced3 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/document/RestIndexAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/document/RestIndexAction.java
@@ -24,11 +24,8 @@ import org.elasticsearch.action.support.ActiveShardCount;
 import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.index.VersionType;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.RestActions;
@@ -38,11 +35,7 @@ import java.io.IOException;
 
 import static org.elasticsearch.rest.RestRequest.Method.POST;
 import static org.elasticsearch.rest.RestRequest.Method.PUT;
-import static org.elasticsearch.rest.RestStatus.BAD_REQUEST;
 
-/**
- *
- */
 public class RestIndexAction extends BaseRestHandler {
 
     @Inject
@@ -62,14 +55,14 @@ public class RestIndexAction extends BaseRestHandler {
         }
 
         @Override
-        public void handleRequest(RestRequest request, RestChannel channel, final NodeClient client) {
+        public RestChannelConsumer prepareRequest(RestRequest request, final NodeClient client) throws IOException {
             request.params().put("op_type", "create");
-            RestIndexAction.this.handleRequest(request, channel, client);
+            return RestIndexAction.this.prepareRequest(request, client);
         }
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         IndexRequest indexRequest = new IndexRequest(request.param("index"), request.param("type"), request.param("id"));
         indexRequest.routing(request.param("routing"));
         indexRequest.parent(request.param("parent")); // order is important, set it after routing, so it will set the routing
@@ -84,24 +77,16 @@ public class RestIndexAction extends BaseRestHandler {
         indexRequest.version(RestActions.parseVersion(request));
         indexRequest.versionType(VersionType.fromString(request.param("version_type"), indexRequest.versionType()));
         String sOpType = request.param("op_type");
-        if (sOpType != null) {
-            try {
-                indexRequest.opType(IndexRequest.OpType.fromString(sOpType));
-            } catch (IllegalArgumentException eia){
-                try {
-                    XContentBuilder builder = channel.newErrorBuilder();
-                    channel.sendResponse(
-                            new BytesRestResponse(BAD_REQUEST, builder.startObject().field("error", eia.getMessage()).endObject()));
-                } catch (IOException e1) {
-                    logger.warn("Failed to send response", e1);
-                    return;
-                }
-            }
-        }
         String waitForActiveShards = request.param("wait_for_active_shards");
         if (waitForActiveShards != null) {
             indexRequest.waitForActiveShards(ActiveShardCount.parseString(waitForActiveShards));
         }
-        client.index(indexRequest, new RestStatusToXContentListener<>(channel, r -> r.getLocation(indexRequest.routing())));
+        if (sOpType != null) {
+            indexRequest.opType(IndexRequest.OpType.fromString(sOpType));
+        }
+
+        return channel ->
+            client.index(indexRequest, new RestStatusToXContentListener<>(channel, r -> r.getLocation(indexRequest.routing())));
     }
+
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/document/RestMultiGetAction.java b/core/src/main/java/org/elasticsearch/rest/action/document/RestMultiGetAction.java
index 07d221fed8..df9af7f42a 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/document/RestMultiGetAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/document/RestMultiGetAction.java
@@ -20,19 +20,19 @@
 package org.elasticsearch.rest.action.document;
 
 import org.elasticsearch.action.get.MultiGetRequest;
-import org.elasticsearch.action.get.MultiGetResponse;
 import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.RestActions;
 import org.elasticsearch.rest.action.RestToXContentListener;
 import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
 
+import java.io.IOException;
+
 import static org.elasticsearch.rest.RestRequest.Method.GET;
 import static org.elasticsearch.rest.RestRequest.Method.POST;
 
@@ -54,7 +54,7 @@ public class RestMultiGetAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) throws Exception {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         MultiGetRequest multiGetRequest = new MultiGetRequest();
         multiGetRequest.refresh(request.paramAsBoolean("refresh", multiGetRequest.refresh()));
         multiGetRequest.preference(request.param("preference"));
@@ -71,8 +71,8 @@ public class RestMultiGetAction extends BaseRestHandler {
 
         FetchSourceContext defaultFetchSource = FetchSourceContext.parseFromRestRequest(request);
         multiGetRequest.add(request.param("index"), request.param("type"), sFields, defaultFetchSource,
-                request.param("routing"), RestActions.getRestContent(request), allowExplicitIndex);
+            request.param("routing"), RestActions.getRestContent(request), allowExplicitIndex);
 
-        client.multiGet(multiGetRequest, new RestToXContentListener<MultiGetResponse>(channel));
+        return channel -> client.multiGet(multiGetRequest, new RestToXContentListener<>(channel));
     }
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/document/RestMultiTermVectorsAction.java b/core/src/main/java/org/elasticsearch/rest/action/document/RestMultiTermVectorsAction.java
index dab23e8df3..fd087e7e76 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/document/RestMultiTermVectorsAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/document/RestMultiTermVectorsAction.java
@@ -27,12 +27,13 @@ import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.RestActions;
 import org.elasticsearch.rest.action.RestToXContentListener;
 
+import java.io.IOException;
+
 import static org.elasticsearch.rest.RestRequest.Method.GET;
 import static org.elasticsearch.rest.RestRequest.Method.POST;
 
@@ -50,7 +51,7 @@ public class RestMultiTermVectorsAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) throws Exception {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         MultiTermVectorsRequest multiTermVectorsRequest = new MultiTermVectorsRequest();
         TermVectorsRequest template = new TermVectorsRequest();
         template.index(request.param("index"));
@@ -59,6 +60,7 @@ public class RestMultiTermVectorsAction extends BaseRestHandler {
         multiTermVectorsRequest.ids(Strings.commaDelimitedListToStringArray(request.param("ids")));
         multiTermVectorsRequest.add(template, RestActions.getRestContent(request));
 
-        client.multiTermVectors(multiTermVectorsRequest, new RestToXContentListener<MultiTermVectorsResponse>(channel));
+        return channel -> client.multiTermVectors(multiTermVectorsRequest, new RestToXContentListener<MultiTermVectorsResponse>(channel));
     }
+
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/document/RestTermVectorsAction.java b/core/src/main/java/org/elasticsearch/rest/action/document/RestTermVectorsAction.java
index b64219215a..3724e34d33 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/document/RestTermVectorsAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/document/RestTermVectorsAction.java
@@ -20,7 +20,6 @@
 package org.elasticsearch.rest.action.document;
 
 import org.elasticsearch.action.termvectors.TermVectorsRequest;
-import org.elasticsearch.action.termvectors.TermVectorsResponse;
 import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.inject.Inject;
@@ -29,12 +28,12 @@ import org.elasticsearch.common.xcontent.XContentFactory;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.index.VersionType;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.RestActions;
 import org.elasticsearch.rest.action.RestToXContentListener;
 
+import java.io.IOException;
 import java.util.HashSet;
 import java.util.Set;
 
@@ -63,17 +62,17 @@ public class RestTermVectorsAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) throws Exception {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         TermVectorsRequest termVectorsRequest = new TermVectorsRequest(request.param("index"), request.param("type"), request.param("id"));
         if (RestActions.hasBodyContent(request)) {
             try (XContentParser parser = XContentFactory.xContent(RestActions.guessBodyContentType(request))
-                    .createParser(RestActions.getRestContent(request))){
+                .createParser(RestActions.getRestContent(request))) {
                 TermVectorsRequest.parseRequest(termVectorsRequest, parser);
             }
         }
         readURIParameters(termVectorsRequest, request);
 
-        client.termVectors(termVectorsRequest, new RestToXContentListener<TermVectorsResponse>(channel));
+        return channel -> client.termVectors(termVectorsRequest, new RestToXContentListener<>(channel));
     }
 
     public static void readURIParameters(TermVectorsRequest termVectorsRequest, RestRequest request) {
diff --git a/core/src/main/java/org/elasticsearch/rest/action/document/RestUpdateAction.java b/core/src/main/java/org/elasticsearch/rest/action/document/RestUpdateAction.java
index 91f71e7249..5cd7336a0f 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/document/RestUpdateAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/document/RestUpdateAction.java
@@ -30,13 +30,14 @@ import org.elasticsearch.common.logging.Loggers;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.index.VersionType;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.RestActions;
 import org.elasticsearch.rest.action.RestStatusToXContentListener;
 import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
 
+import java.io.IOException;
+
 import static org.elasticsearch.rest.RestRequest.Method.POST;
 
 /**
@@ -52,7 +53,7 @@ public class RestUpdateAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) throws Exception {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         UpdateRequest updateRequest = new UpdateRequest(request.param("index"), request.param("type"), request.param("id"));
         updateRequest.routing(request.param("routing"));
         updateRequest.parent(request.param("parent"));
@@ -108,6 +109,7 @@ public class RestUpdateAction extends BaseRestHandler {
             }
         }
 
-        client.update(updateRequest, new RestStatusToXContentListener<>(channel, r -> r.getLocation(updateRequest.routing())));
+        return channel ->
+            client.update(updateRequest, new RestStatusToXContentListener<>(channel, r -> r.getLocation(updateRequest.routing())));
     }
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/ingest/RestDeletePipelineAction.java b/core/src/main/java/org/elasticsearch/rest/action/ingest/RestDeletePipelineAction.java
index 593b55b8b7..f9430ea0de 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/ingest/RestDeletePipelineAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/ingest/RestDeletePipelineAction.java
@@ -24,11 +24,12 @@ import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.AcknowledgedRestListener;
 
+import java.io.IOException;
+
 public class RestDeletePipelineAction extends BaseRestHandler {
 
     @Inject
@@ -38,10 +39,10 @@ public class RestDeletePipelineAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(RestRequest restRequest, RestChannel channel, NodeClient client) throws Exception {
+    public RestChannelConsumer prepareRequest(RestRequest restRequest, NodeClient client) throws IOException {
         DeletePipelineRequest request = new DeletePipelineRequest(restRequest.param("id"));
         request.masterNodeTimeout(restRequest.paramAsTime("master_timeout", request.masterNodeTimeout()));
         request.timeout(restRequest.paramAsTime("timeout", request.timeout()));
-        client.admin().cluster().deletePipeline(request, new AcknowledgedRestListener<>(channel));
+        return channel -> client.admin().cluster().deletePipeline(request, new AcknowledgedRestListener<>(channel));
     }
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/ingest/RestGetPipelineAction.java b/core/src/main/java/org/elasticsearch/rest/action/ingest/RestGetPipelineAction.java
index 308fb146c3..7124bb35b3 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/ingest/RestGetPipelineAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/ingest/RestGetPipelineAction.java
@@ -25,11 +25,12 @@ import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.RestStatusToXContentListener;
 
+import java.io.IOException;
+
 public class RestGetPipelineAction extends BaseRestHandler {
 
     @Inject
@@ -40,9 +41,9 @@ public class RestGetPipelineAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(RestRequest restRequest, RestChannel channel, NodeClient client) throws Exception {
+    public RestChannelConsumer prepareRequest(RestRequest restRequest, NodeClient client) throws IOException {
         GetPipelineRequest request = new GetPipelineRequest(Strings.splitStringByCommaToArray(restRequest.param("id")));
         request.masterNodeTimeout(restRequest.paramAsTime("master_timeout", request.masterNodeTimeout()));
-        client.admin().cluster().getPipeline(request, new RestStatusToXContentListener<>(channel));
+        return channel -> client.admin().cluster().getPipeline(request, new RestStatusToXContentListener<>(channel));
     }
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/ingest/RestPutPipelineAction.java b/core/src/main/java/org/elasticsearch/rest/action/ingest/RestPutPipelineAction.java
index b6d34a6c25..d4388f6bea 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/ingest/RestPutPipelineAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/ingest/RestPutPipelineAction.java
@@ -24,12 +24,13 @@ import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.AcknowledgedRestListener;
 import org.elasticsearch.rest.action.RestActions;
 
+import java.io.IOException;
+
 
 public class RestPutPipelineAction extends BaseRestHandler {
 
@@ -40,11 +41,11 @@ public class RestPutPipelineAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(RestRequest restRequest, RestChannel channel, NodeClient client) throws Exception {
+    public RestChannelConsumer prepareRequest(RestRequest restRequest, NodeClient client) throws IOException {
         PutPipelineRequest request = new PutPipelineRequest(restRequest.param("id"), RestActions.getRestContent(restRequest));
         request.masterNodeTimeout(restRequest.paramAsTime("master_timeout", request.masterNodeTimeout()));
         request.timeout(restRequest.paramAsTime("timeout", request.timeout()));
-        client.admin().cluster().putPipeline(request, new AcknowledgedRestListener<>(channel));
+        return channel -> client.admin().cluster().putPipeline(request, new AcknowledgedRestListener<>(channel));
     }
 
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/ingest/RestSimulatePipelineAction.java b/core/src/main/java/org/elasticsearch/rest/action/ingest/RestSimulatePipelineAction.java
index a51bdf5fef..07a5845958 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/ingest/RestSimulatePipelineAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/ingest/RestSimulatePipelineAction.java
@@ -24,13 +24,13 @@ import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.RestActions;
-import org.elasticsearch.rest.action.RestStatusToXContentListener;
 import org.elasticsearch.rest.action.RestToXContentListener;
 
+import java.io.IOException;
+
 public class RestSimulatePipelineAction extends BaseRestHandler {
 
     @Inject
@@ -43,10 +43,10 @@ public class RestSimulatePipelineAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(RestRequest restRequest, RestChannel channel, NodeClient client) throws Exception {
+    public RestChannelConsumer prepareRequest(RestRequest restRequest, NodeClient client) throws IOException {
         SimulatePipelineRequest request = new SimulatePipelineRequest(RestActions.getRestContent(restRequest));
         request.setId(restRequest.param("id"));
         request.setVerbose(restRequest.paramAsBoolean("verbose", false));
-        client.admin().cluster().simulatePipeline(request, new RestToXContentListener<>(channel));
+        return channel -> client.admin().cluster().simulatePipeline(request, new RestToXContentListener<>(channel));
     }
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/search/RestClearScrollAction.java b/core/src/main/java/org/elasticsearch/rest/action/search/RestClearScrollAction.java
index 4c8f84a222..bc49bffe44 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/search/RestClearScrollAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/search/RestClearScrollAction.java
@@ -30,7 +30,6 @@ import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.RestActions;
@@ -52,7 +51,7 @@ public class RestClearScrollAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         String scrollIds = request.param("scroll_id");
         ClearScrollRequest clearRequest = new ClearScrollRequest();
         clearRequest.setScrollIds(Arrays.asList(splitScrollIds(scrollIds)));
@@ -68,7 +67,7 @@ public class RestClearScrollAction extends BaseRestHandler {
            }
         }
 
-        client.clearScroll(clearRequest, new RestStatusToXContentListener<ClearScrollResponse>(channel));
+        return channel -> client.clearScroll(clearRequest, new RestStatusToXContentListener<ClearScrollResponse>(channel));
     }
 
     public static String[] splitScrollIds(String scrollIds) {
diff --git a/core/src/main/java/org/elasticsearch/rest/action/search/RestExplainAction.java b/core/src/main/java/org/elasticsearch/rest/action/search/RestExplainAction.java
index 597bf3db61..16de4fb9ac 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/search/RestExplainAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/search/RestExplainAction.java
@@ -33,7 +33,6 @@ import org.elasticsearch.index.query.QueryBuilder;
 import org.elasticsearch.indices.query.IndicesQueriesRegistry;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
@@ -64,7 +63,7 @@ public class RestExplainAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         final ExplainRequest explainRequest = new ExplainRequest(request.param("index"), request.param("type"), request.param("id"));
         explainRequest.parent(request.param("parent"));
         explainRequest.routing(request.param("routing"));
@@ -92,7 +91,7 @@ public class RestExplainAction extends BaseRestHandler {
 
         explainRequest.fetchSourceContext(FetchSourceContext.parseFromRestRequest(request));
 
-        client.explain(explainRequest, new RestBuilderListener<ExplainResponse>(channel) {
+        return channel -> client.explain(explainRequest, new RestBuilderListener<ExplainResponse>(channel) {
             @Override
             public RestResponse buildResponse(ExplainResponse response, XContentBuilder builder) throws Exception {
                 builder.startObject();
diff --git a/core/src/main/java/org/elasticsearch/rest/action/search/RestMultiSearchAction.java b/core/src/main/java/org/elasticsearch/rest/action/search/RestMultiSearchAction.java
index a54e40be73..8117024f17 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/search/RestMultiSearchAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/search/RestMultiSearchAction.java
@@ -34,7 +34,6 @@ import org.elasticsearch.common.xcontent.XContentFactory;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.index.query.QueryParseContext;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.RestActions;
@@ -75,9 +74,9 @@ public class RestMultiSearchAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) throws Exception {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         MultiSearchRequest multiSearchRequest = parseRequest(request, allowExplicitIndex, searchRequestParsers, parseFieldMatcher);
-        client.multiSearch(multiSearchRequest, new RestToXContentListener<>(channel));
+        return channel -> client.multiSearch(multiSearchRequest, new RestToXContentListener<>(channel));
     }
 
     /**
diff --git a/core/src/main/java/org/elasticsearch/rest/action/search/RestSearchAction.java b/core/src/main/java/org/elasticsearch/rest/action/search/RestSearchAction.java
index 8acfc72dfe..6980bf3ed3 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/search/RestSearchAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/search/RestSearchAction.java
@@ -33,7 +33,6 @@ import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.index.query.QueryBuilder;
 import org.elasticsearch.index.query.QueryParseContext;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.RestActions;
@@ -76,11 +75,12 @@ public class RestSearchAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) throws IOException {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         SearchRequest searchRequest = new SearchRequest();
         BytesReference restContent = RestActions.hasBodyContent(request) ? RestActions.getRestContent(request) : null;
         parseSearchRequest(searchRequest, request, searchRequestParsers, parseFieldMatcher, restContent);
-        client.search(searchRequest, new RestStatusToXContentListener<>(channel));
+
+        return channel -> client.search(searchRequest, new RestStatusToXContentListener<>(channel));
     }
 
     /**
diff --git a/core/src/main/java/org/elasticsearch/rest/action/search/RestSearchScrollAction.java b/core/src/main/java/org/elasticsearch/rest/action/search/RestSearchScrollAction.java
index 9b9ddd3a93..abe27bd41f 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/search/RestSearchScrollAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/search/RestSearchScrollAction.java
@@ -19,7 +19,6 @@
 
 package org.elasticsearch.rest.action.search;
 
-import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.action.search.SearchScrollRequest;
 import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.bytes.BytesReference;
@@ -31,7 +30,6 @@ import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.RestActions;
@@ -57,7 +55,7 @@ public class RestSearchScrollAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         String scrollId = request.param("scroll_id");
         SearchScrollRequest searchScrollRequest = new SearchScrollRequest();
         searchScrollRequest.scrollId(scrollId);
@@ -78,7 +76,7 @@ public class RestSearchScrollAction extends BaseRestHandler {
                 buildFromContent(RestActions.getRestContent(request), searchScrollRequest);
             }
         }
-        client.searchScroll(searchScrollRequest, new RestStatusToXContentListener<SearchResponse>(channel));
+        return channel -> client.searchScroll(searchScrollRequest, new RestStatusToXContentListener<>(channel));
     }
 
     public static void buildFromContent(BytesReference content, SearchScrollRequest searchScrollRequest) {
diff --git a/core/src/main/java/org/elasticsearch/rest/action/search/RestSuggestAction.java b/core/src/main/java/org/elasticsearch/rest/action/search/RestSuggestAction.java
index b55a1590f3..4f61d7f4c8 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/search/RestSuggestAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/search/RestSuggestAction.java
@@ -19,8 +19,6 @@
 
 package org.elasticsearch.rest.action.search;
 
-import java.io.IOException;
-
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.action.support.IndicesOptions;
@@ -35,7 +33,6 @@ import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.index.query.QueryParseContext;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
@@ -47,6 +44,8 @@ import org.elasticsearch.search.builder.SearchSourceBuilder;
 import org.elasticsearch.search.suggest.Suggest;
 import org.elasticsearch.search.suggest.SuggestBuilder;
 
+import java.io.IOException;
+
 import static org.elasticsearch.rest.RestRequest.Method.GET;
 import static org.elasticsearch.rest.RestRequest.Method.POST;
 import static org.elasticsearch.rest.action.RestActions.buildBroadcastShardsHeader;
@@ -67,7 +66,7 @@ public class RestSuggestAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) throws IOException {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         final SearchRequest searchRequest = new SearchRequest(
                 Strings.splitStringByCommaToArray(request.param("index")), new SearchSourceBuilder());
         searchRequest.indicesOptions(IndicesOptions.fromRequest(request, searchRequest.indicesOptions()));
@@ -82,7 +81,7 @@ public class RestSuggestAction extends BaseRestHandler {
         }
         searchRequest.routing(request.param("routing"));
         searchRequest.preference(request.param("preference"));
-        client.search(searchRequest, new RestBuilderListener<SearchResponse>(channel) {
+        return channel -> client.search(searchRequest, new RestBuilderListener<SearchResponse>(channel) {
             @Override
             public RestResponse buildResponse(SearchResponse response, XContentBuilder builder) throws Exception {
                 RestStatus restStatus = RestStatus.status(response.getSuccessfulShards(),
diff --git a/core/src/test/java/org/elasticsearch/action/admin/cluster/node/tasks/TransportTasksActionTests.java b/core/src/test/java/org/elasticsearch/action/admin/cluster/node/tasks/TransportTasksActionTests.java
index f964eeba9f..3952ab2046 100644
--- a/core/src/test/java/org/elasticsearch/action/admin/cluster/node/tasks/TransportTasksActionTests.java
+++ b/core/src/test/java/org/elasticsearch/action/admin/cluster/node/tasks/TransportTasksActionTests.java
@@ -47,7 +47,6 @@ import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentFactory;
 import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.common.xcontent.XContentType;
-import org.elasticsearch.rest.action.admin.cluster.RestListTasksAction;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.tasks.TaskId;
 import org.elasticsearch.tasks.TaskInfo;
@@ -66,7 +65,6 @@ import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
-import java.util.function.Consumer;
 
 import static org.elasticsearch.action.support.PlainActionFuture.newFuture;
 import static org.hamcrest.Matchers.containsString;
@@ -536,7 +534,7 @@ public class TransportTasksActionTests extends TaskManagerTestCase {
 
         // Try to cancel main task using action name
         CancelTasksRequest request = new CancelTasksRequest();
-        request.setNodesIds(testNodes[0].discoveryNode.getId());
+        request.setNodes(testNodes[0].discoveryNode.getId());
         request.setReason("Testing Cancellation");
         request.setActions(actionName);
         CancelTasksResponse response = testNodes[randomIntBetween(0, testNodes.length - 1)].transportCancelTasksAction.execute(request)
diff --git a/core/src/test/java/org/elasticsearch/common/network/NetworkModuleTests.java b/core/src/test/java/org/elasticsearch/common/network/NetworkModuleTests.java
index 2daea85967..d86773d2d9 100644
--- a/core/src/test/java/org/elasticsearch/common/network/NetworkModuleTests.java
+++ b/core/src/test/java/org/elasticsearch/common/network/NetworkModuleTests.java
@@ -34,7 +34,6 @@ import org.elasticsearch.http.HttpStats;
 import org.elasticsearch.indices.breaker.CircuitBreakerService;
 import org.elasticsearch.plugins.NetworkPlugin;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.cat.AbstractCatAction;
 import org.elasticsearch.test.transport.AssertingLocalTransport;
@@ -42,6 +41,7 @@ import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.Transport;
 import org.elasticsearch.transport.TransportInterceptor;
 
+import java.io.IOException;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
@@ -89,7 +89,7 @@ public class NetworkModuleTests extends ModuleTestCase {
             super(null);
         }
         @Override
-        public void handleRequest(RestRequest request, RestChannel channel, NodeClient client) throws Exception {}
+        public RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException { return channel -> {}; }
     }
 
     static class FakeCatRestHandler extends AbstractCatAction {
@@ -97,7 +97,7 @@ public class NetworkModuleTests extends ModuleTestCase {
             super(null);
         }
         @Override
-        protected void doRequest(RestRequest request, RestChannel channel, NodeClient client) {}
+        protected RestChannelConsumer doCatRequest(RestRequest request, NodeClient client) { return channel -> {}; }
         @Override
         protected void documentation(StringBuilder sb) {}
         @Override
diff --git a/core/src/test/java/org/elasticsearch/rest/BaseRestHandlerTests.java b/core/src/test/java/org/elasticsearch/rest/BaseRestHandlerTests.java
new file mode 100644
index 0000000000..6019ab8aac
--- /dev/null
+++ b/core/src/test/java/org/elasticsearch/rest/BaseRestHandlerTests.java
@@ -0,0 +1,141 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.rest;
+
+import org.elasticsearch.client.node.NodeClient;
+import org.elasticsearch.common.Table;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.rest.action.cat.AbstractCatAction;
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.test.rest.FakeRestChannel;
+import org.elasticsearch.test.rest.FakeRestRequest;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.hamcrest.core.StringContains.containsString;
+import static org.hamcrest.object.HasToString.hasToString;
+import static org.mockito.Mockito.mock;
+
+public class BaseRestHandlerTests extends ESTestCase {
+
+    public void testUnconsumedParameters() throws Exception {
+        final AtomicBoolean executed = new AtomicBoolean();
+        BaseRestHandler handler = new BaseRestHandler(Settings.EMPTY) {
+            @Override
+            protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
+                request.param("consumed");
+                return channel -> executed.set(true);
+            }
+        };
+
+        final HashMap<String, String> params = new HashMap<>();
+        params.put("consumed", randomAsciiOfLength(8));
+        params.put("unconsumed", randomAsciiOfLength(8));
+        RestRequest request = new FakeRestRequest.Builder().withParams(params).build();
+        RestChannel channel = new FakeRestChannel(request, randomBoolean(), 1);
+        final IllegalArgumentException e =
+            expectThrows(IllegalArgumentException.class, () -> handler.handleRequest(request, channel, mock(NodeClient.class)));
+        assertThat(e, hasToString(containsString("request [/] contains unused params: [unconsumed]")));
+        assertFalse(executed.get());
+    }
+
+    public void testUnconsumedResponseParameters() throws Exception {
+        final AtomicBoolean executed = new AtomicBoolean();
+        BaseRestHandler handler = new BaseRestHandler(Settings.EMPTY) {
+            @Override
+            protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
+                request.param("consumed");
+                return channel -> executed.set(true);
+            }
+
+            @Override
+            protected Set<String> responseParams() {
+                return Collections.singleton("response_param");
+            }
+        };
+
+        final HashMap<String, String> params = new HashMap<>();
+        params.put("consumed", randomAsciiOfLength(8));
+        params.put("response_param", randomAsciiOfLength(8));
+        RestRequest request = new FakeRestRequest.Builder().withParams(params).build();
+        RestChannel channel = new FakeRestChannel(request, randomBoolean(), 1);
+        handler.handleRequest(request, channel, mock(NodeClient.class));
+        assertTrue(executed.get());
+    }
+
+    public void testDefaultResponseParameters() throws Exception {
+        final AtomicBoolean executed = new AtomicBoolean();
+        BaseRestHandler handler = new BaseRestHandler(Settings.EMPTY) {
+            @Override
+            protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
+                return channel -> executed.set(true);
+            }
+        };
+
+        final HashMap<String, String> params = new HashMap<>();
+        params.put("format", randomAsciiOfLength(8));
+        params.put("filter_path", randomAsciiOfLength(8));
+        params.put("pretty", randomAsciiOfLength(8));
+        params.put("human", randomAsciiOfLength(8));
+        RestRequest request = new FakeRestRequest.Builder().withParams(params).build();
+        RestChannel channel = new FakeRestChannel(request, randomBoolean(), 1);
+        handler.handleRequest(request, channel, mock(NodeClient.class));
+        assertTrue(executed.get());
+    }
+
+    public void testCatResponseParameters() throws Exception {
+        final AtomicBoolean executed = new AtomicBoolean();
+        AbstractCatAction handler = new AbstractCatAction(Settings.EMPTY) {
+            @Override
+            protected RestChannelConsumer doCatRequest(RestRequest request, NodeClient client) {
+                return channel -> executed.set(true);
+            }
+
+            @Override
+            protected void documentation(StringBuilder sb) {
+
+            }
+
+            @Override
+            protected Table getTableWithHeader(RestRequest request) {
+                return null;
+            }
+        };
+
+        final HashMap<String, String> params = new HashMap<>();
+        params.put("format", randomAsciiOfLength(8));
+        params.put("h", randomAsciiOfLength(8));
+        params.put("v", randomAsciiOfLength(8));
+        params.put("ts", randomAsciiOfLength(8));
+        params.put("pri", randomAsciiOfLength(8));
+        params.put("bytes", randomAsciiOfLength(8));
+        params.put("size", randomAsciiOfLength(8));
+        params.put("time", randomAsciiOfLength(8));
+        RestRequest request = new FakeRestRequest.Builder().withParams(params).build();
+        RestChannel channel = new FakeRestChannel(request, randomBoolean(), 1);
+        handler.handleRequest(request, channel, mock(NodeClient.class));
+        assertTrue(executed.get());
+    }
+
+}
diff --git a/docs/reference/cluster/tasks.asciidoc b/docs/reference/cluster/tasks.asciidoc
index dec4fb8f69..2176a43229 100644
--- a/docs/reference/cluster/tasks.asciidoc
+++ b/docs/reference/cluster/tasks.asciidoc
@@ -62,7 +62,7 @@ It is also possible to retrieve information for a particular task:
 
 [source,js]
 --------------------------------------------------
-GET _tasks/taskId:1 <1>
+GET _tasks/task_id:1 <1>
 --------------------------------------------------
 // CONSOLE
 // TEST[catch:missing]
@@ -114,7 +114,7 @@ If a long-running task supports cancellation, it can be cancelled by the followi
 
 [source,js]
 --------------------------------------------------
-POST _tasks/taskId:1/_cancel
+POST _tasks/task_id:1/_cancel
 --------------------------------------------------
 // CONSOLE
 
@@ -124,7 +124,7 @@ nodes `nodeId1` and `nodeId2`.
 
 [source,js]
 --------------------------------------------------
-POST _tasks/_cancel?node_id=nodeId1,nodeId2&actions=*reindex
+POST _tasks/_cancel?nodes=nodeId1,nodeId2&actions=*reindex
 --------------------------------------------------
 // CONSOLE
 
diff --git a/docs/reference/docs/delete-by-query.asciidoc b/docs/reference/docs/delete-by-query.asciidoc
index 468f454562..fa0d610f2b 100644
--- a/docs/reference/docs/delete-by-query.asciidoc
+++ b/docs/reference/docs/delete-by-query.asciidoc
@@ -233,7 +233,7 @@ You can fetch the status of any running delete-by-query requests with the
 
 [source,js]
 --------------------------------------------------
-GET _tasks?detailed=true&action=*/delete/byquery
+GET _tasks?detailed=true&actions=*/delete/byquery
 --------------------------------------------------
 // CONSOLE
 
@@ -308,7 +308,7 @@ Any Delete By Query can be canceled using the <<tasks,Task Cancel API>>:
 
 [source,js]
 --------------------------------------------------
-POST _tasks/taskid:1/_cancel
+POST _tasks/task_id:1/_cancel
 --------------------------------------------------
 // CONSOLE
 
@@ -327,7 +327,7 @@ using the `_rethrottle` API:
 
 [source,js]
 --------------------------------------------------
-POST _delete_by_query/taskid:1/_rethrottle?requests_per_second=-1
+POST _delete_by_query/task_id:1/_rethrottle?requests_per_second=-1
 --------------------------------------------------
 // CONSOLE
 
diff --git a/docs/reference/docs/reindex.asciidoc b/docs/reference/docs/reindex.asciidoc
index 46786c0b6d..d04ef0ee09 100644
--- a/docs/reference/docs/reindex.asciidoc
+++ b/docs/reference/docs/reindex.asciidoc
@@ -595,7 +595,7 @@ Any Reindex can be canceled using the <<tasks,Task Cancel API>>:
 
 [source,js]
 --------------------------------------------------
-POST _tasks/taskid:1/_cancel
+POST _tasks/task_id:1/_cancel
 --------------------------------------------------
 // CONSOLE
 
@@ -614,7 +614,7 @@ the `_rethrottle` API:
 
 [source,js]
 --------------------------------------------------
-POST _reindex/taskid:1/_rethrottle?requests_per_second=-1
+POST _reindex/task_id:1/_rethrottle?requests_per_second=-1
 --------------------------------------------------
 // CONSOLE
 
diff --git a/docs/reference/docs/update-by-query.asciidoc b/docs/reference/docs/update-by-query.asciidoc
index 7299d398e6..39a61791fd 100644
--- a/docs/reference/docs/update-by-query.asciidoc
+++ b/docs/reference/docs/update-by-query.asciidoc
@@ -295,7 +295,7 @@ You can fetch the status of all running update-by-query requests with the
 
 [source,js]
 --------------------------------------------------
-GET _tasks?detailed=true&action=*byquery
+GET _tasks?detailed=true&actions=*byquery
 --------------------------------------------------
 // CONSOLE
 
@@ -373,7 +373,7 @@ Any Update By Query can be canceled using the <<tasks,Task Cancel API>>:
 
 [source,js]
 --------------------------------------------------
-POST _tasks/taskid:1/_cancel
+POST _tasks/task_id:1/_cancel
 --------------------------------------------------
 // CONSOLE
 
@@ -392,7 +392,7 @@ using the `_rethrottle` API:
 
 [source,js]
 --------------------------------------------------
-POST _update_by_query/taskid:1/_rethrottle?requests_per_second=-1
+POST _update_by_query/task_id:1/_rethrottle?requests_per_second=-1
 --------------------------------------------------
 // CONSOLE
 
diff --git a/docs/reference/setup/sysconfig/file-descriptors.asciidoc b/docs/reference/setup/sysconfig/file-descriptors.asciidoc
index 25a2214146..b60dd3ce61 100644
--- a/docs/reference/setup/sysconfig/file-descriptors.asciidoc
+++ b/docs/reference/setup/sysconfig/file-descriptors.asciidoc
@@ -1,6 +1,10 @@
 [[file-descriptors]]
 === File Descriptors
 
+[NOTE]
+This is only a problem for Linux and macOS and can be safely ignored if running
+Elasticsearch on Windows.
+
 Elasticsearch uses a lot of file descriptors or file handles.  Running out of
 file descriptors can be disastrous and will most probably lead to data loss.
 Make sure to increase the limit on the number of open files descriptors for
diff --git a/modules/lang-mustache/src/main/java/org/elasticsearch/script/mustache/RestMultiSearchTemplateAction.java b/modules/lang-mustache/src/main/java/org/elasticsearch/script/mustache/RestMultiSearchTemplateAction.java
index a7b8615372..818e170a90 100644
--- a/modules/lang-mustache/src/main/java/org/elasticsearch/script/mustache/RestMultiSearchTemplateAction.java
+++ b/modules/lang-mustache/src/main/java/org/elasticsearch/script/mustache/RestMultiSearchTemplateAction.java
@@ -25,7 +25,6 @@ import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.RestActions;
@@ -55,13 +54,13 @@ public class RestMultiSearchTemplateAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(RestRequest request, RestChannel channel, NodeClient client) throws Exception {
+    public RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
         if (RestActions.hasBodyContent(request) == false) {
             throw new ElasticsearchException("request body is required");
         }
 
         MultiSearchTemplateRequest multiRequest = parseRequest(request, allowExplicitIndex);
-        client.execute(MultiSearchTemplateAction.INSTANCE, multiRequest, new RestToXContentListener<>(channel));
+        return channel -> client.execute(MultiSearchTemplateAction.INSTANCE, multiRequest, new RestToXContentListener<>(channel));
     }
 
     /**
diff --git a/modules/lang-mustache/src/main/java/org/elasticsearch/script/mustache/RestRenderSearchTemplateAction.java b/modules/lang-mustache/src/main/java/org/elasticsearch/script/mustache/RestRenderSearchTemplateAction.java
index 74dc6363f3..489d7f83f3 100644
--- a/modules/lang-mustache/src/main/java/org/elasticsearch/script/mustache/RestRenderSearchTemplateAction.java
+++ b/modules/lang-mustache/src/main/java/org/elasticsearch/script/mustache/RestRenderSearchTemplateAction.java
@@ -23,13 +23,14 @@ import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.RestActions;
 import org.elasticsearch.rest.action.RestToXContentListener;
 import org.elasticsearch.script.ScriptService;
 
+import java.io.IOException;
+
 import static org.elasticsearch.rest.RestRequest.Method.GET;
 import static org.elasticsearch.rest.RestRequest.Method.POST;
 
@@ -45,7 +46,7 @@ public class RestRenderSearchTemplateAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(RestRequest request, RestChannel channel, NodeClient client) throws Exception {
+    public RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
         // Creates the render template request
         SearchTemplateRequest renderRequest = RestSearchTemplateAction.parse(RestActions.getRestContent(request));
         renderRequest.setSimulate(true);
@@ -56,6 +57,6 @@ public class RestRenderSearchTemplateAction extends BaseRestHandler {
             renderRequest.setScript(id);
         }
 
-        client.execute(SearchTemplateAction.INSTANCE, renderRequest, new RestToXContentListener<>(channel));
+        return channel -> client.execute(SearchTemplateAction.INSTANCE, renderRequest, new RestToXContentListener<>(channel));
     }
 }
diff --git a/modules/lang-mustache/src/main/java/org/elasticsearch/script/mustache/RestSearchTemplateAction.java b/modules/lang-mustache/src/main/java/org/elasticsearch/script/mustache/RestSearchTemplateAction.java
index 6d830a21f5..2feeb2d328 100644
--- a/modules/lang-mustache/src/main/java/org/elasticsearch/script/mustache/RestSearchTemplateAction.java
+++ b/modules/lang-mustache/src/main/java/org/elasticsearch/script/mustache/RestSearchTemplateAction.java
@@ -34,9 +34,7 @@ import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentFactory;
 import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.common.xcontent.XContentParser;
-import org.elasticsearch.indices.query.IndicesQueriesRegistry;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.RestActions;
@@ -44,8 +42,6 @@ import org.elasticsearch.rest.action.RestStatusToXContentListener;
 import org.elasticsearch.rest.action.search.RestSearchAction;
 import org.elasticsearch.script.ScriptService;
 import org.elasticsearch.search.SearchRequestParsers;
-import org.elasticsearch.search.aggregations.AggregatorParsers;
-import org.elasticsearch.search.suggest.Suggesters;
 
 import java.io.IOException;
 
@@ -100,7 +96,7 @@ public class RestSearchTemplateAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(RestRequest request, RestChannel channel, NodeClient client) throws Exception {
+    public RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
         if (RestActions.hasBodyContent(request) == false) {
             throw new ElasticsearchException("request body is required");
         }
@@ -113,7 +109,7 @@ public class RestSearchTemplateAction extends BaseRestHandler {
         SearchTemplateRequest searchTemplateRequest = parse(RestActions.getRestContent(request));
         searchTemplateRequest.setRequest(searchRequest);
 
-        client.execute(SearchTemplateAction.INSTANCE, searchTemplateRequest, new RestStatusToXContentListener<>(channel));
+        return channel -> client.execute(SearchTemplateAction.INSTANCE, searchTemplateRequest, new RestStatusToXContentListener<>(channel));
     }
 
     public static SearchTemplateRequest parse(BytesReference bytes) throws IOException {
diff --git a/modules/percolator/src/main/java/org/elasticsearch/percolator/MultiPercolateRequest.java b/modules/percolator/src/main/java/org/elasticsearch/percolator/MultiPercolateRequest.java
index a18c12bb76..d8bb91f2fb 100644
--- a/modules/percolator/src/main/java/org/elasticsearch/percolator/MultiPercolateRequest.java
+++ b/modules/percolator/src/main/java/org/elasticsearch/percolator/MultiPercolateRequest.java
@@ -89,7 +89,7 @@ public class MultiPercolateRequest extends ActionRequest<MultiPercolateRequest>
     /**
      * Embeds a percolate request which request body is defined as raw bytes to this multi percolate request
      */
-    public MultiPercolateRequest add(BytesReference data, boolean allowExplicitIndex) throws Exception {
+    public MultiPercolateRequest add(BytesReference data, boolean allowExplicitIndex) throws IOException {
         XContent xContent = XContentFactory.xContent(data);
         int from = 0;
         int length = data.length();
diff --git a/modules/percolator/src/main/java/org/elasticsearch/percolator/RestMultiPercolateAction.java b/modules/percolator/src/main/java/org/elasticsearch/percolator/RestMultiPercolateAction.java
index 3d8631624e..a7a8e51862 100644
--- a/modules/percolator/src/main/java/org/elasticsearch/percolator/RestMultiPercolateAction.java
+++ b/modules/percolator/src/main/java/org/elasticsearch/percolator/RestMultiPercolateAction.java
@@ -16,6 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
+
 package org.elasticsearch.percolator;
 
 import org.elasticsearch.action.support.IndicesOptions;
@@ -24,12 +25,13 @@ import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.RestActions;
 import org.elasticsearch.rest.action.RestToXContentListener;
 
+import java.io.IOException;
+
 import static org.elasticsearch.rest.RestRequest.Method.GET;
 import static org.elasticsearch.rest.RestRequest.Method.POST;
 
@@ -53,14 +55,13 @@ public class RestMultiPercolateAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest restRequest, final RestChannel restChannel, final NodeClient client) throws Exception {
+    public RestChannelConsumer prepareRequest(final RestRequest restRequest, final NodeClient client) throws IOException {
         MultiPercolateRequest multiPercolateRequest = new MultiPercolateRequest();
         multiPercolateRequest.indicesOptions(IndicesOptions.fromRequest(restRequest, multiPercolateRequest.indicesOptions()));
         multiPercolateRequest.indices(Strings.splitStringByCommaToArray(restRequest.param("index")));
         multiPercolateRequest.documentType(restRequest.param("type"));
         multiPercolateRequest.add(RestActions.getRestContent(restRequest), allowExplicitIndex);
-        client.execute(MultiPercolateAction.INSTANCE, multiPercolateRequest,
-                new RestToXContentListener<MultiPercolateResponse>(restChannel));
+        return channel -> client.execute(MultiPercolateAction.INSTANCE, multiPercolateRequest, new RestToXContentListener<>(channel));
     }
 
 }
diff --git a/modules/percolator/src/main/java/org/elasticsearch/percolator/RestPercolateAction.java b/modules/percolator/src/main/java/org/elasticsearch/percolator/RestPercolateAction.java
index 35d9f2c604..0c9eaa4d12 100644
--- a/modules/percolator/src/main/java/org/elasticsearch/percolator/RestPercolateAction.java
+++ b/modules/percolator/src/main/java/org/elasticsearch/percolator/RestPercolateAction.java
@@ -16,6 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
+
 package org.elasticsearch.percolator;
 
 import org.elasticsearch.action.get.GetRequest;
@@ -32,11 +33,14 @@ import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.RestActions;
 import org.elasticsearch.rest.action.RestToXContentListener;
 
+import java.io.IOException;
+
 import static org.elasticsearch.rest.RestRequest.Method.GET;
 import static org.elasticsearch.rest.RestRequest.Method.POST;
 
 @Deprecated
 public class RestPercolateAction extends BaseRestHandler {
+
     @Inject
     public RestPercolateAction(Settings settings, RestController controller) {
         super(settings);
@@ -56,7 +60,7 @@ public class RestPercolateAction extends BaseRestHandler {
         controller.registerHandler(POST, "/{index}/{type}/{id}/_percolate/count", countExistingDocHandler);
     }
 
-    void parseDocPercolate(PercolateRequest percolateRequest, RestRequest restRequest, RestChannel restChannel, NodeClient client) {
+    private RestChannelConsumer parseDocPercolate(PercolateRequest percolateRequest, RestRequest restRequest, NodeClient client) {
         percolateRequest.indices(Strings.splitStringByCommaToArray(restRequest.param("index")));
         percolateRequest.documentType(restRequest.param("type"));
         percolateRequest.routing(restRequest.param("routing"));
@@ -64,11 +68,10 @@ public class RestPercolateAction extends BaseRestHandler {
         percolateRequest.source(RestActions.getRestContent(restRequest));
 
         percolateRequest.indicesOptions(IndicesOptions.fromRequest(restRequest, percolateRequest.indicesOptions()));
-        executePercolate(client, percolateRequest, restChannel);
+        return channel -> executePercolate(client, percolateRequest, channel);
     }
 
-    void parseExistingDocPercolate(PercolateRequest percolateRequest, RestRequest restRequest, RestChannel restChannel,
-            NodeClient client) {
+    private RestChannelConsumer parseExistingDocPercolate(PercolateRequest percolateRequest, RestRequest restRequest, NodeClient client) {
         String index = restRequest.param("index");
         String type = restRequest.param("type");
         percolateRequest.indices(Strings.splitStringByCommaToArray(restRequest.param("percolate_index", index)));
@@ -89,57 +92,61 @@ public class RestPercolateAction extends BaseRestHandler {
         percolateRequest.source(RestActions.getRestContent(restRequest));
 
         percolateRequest.indicesOptions(IndicesOptions.fromRequest(restRequest, percolateRequest.indicesOptions()));
-        executePercolate(client, percolateRequest, restChannel);
+        return channel -> executePercolate(client, percolateRequest, channel);
     }
 
-    void executePercolate(final NodeClient client, final PercolateRequest percolateRequest, final RestChannel restChannel) {
+    private void executePercolate(final NodeClient client, final PercolateRequest percolateRequest, final RestChannel restChannel) {
         client.execute(PercolateAction.INSTANCE, percolateRequest, new RestToXContentListener<>(restChannel));
     }
 
     @Override
-    public void handleRequest(RestRequest restRequest, RestChannel restChannel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(RestRequest restRequest, final NodeClient client) throws IOException {
         PercolateRequest percolateRequest = new PercolateRequest();
-        parseDocPercolate(percolateRequest, restRequest, restChannel, client);
+        return parseDocPercolate(percolateRequest, restRequest, client);
     }
 
-    final class RestCountPercolateDocHandler extends BaseRestHandler {
+    private final class RestCountPercolateDocHandler extends BaseRestHandler {
 
         private RestCountPercolateDocHandler(Settings settings) {
             super(settings);
         }
 
         @Override
-        public void handleRequest(RestRequest restRequest, RestChannel restChannel, final NodeClient client) {
+        public RestChannelConsumer prepareRequest(RestRequest restRequest, final NodeClient client) throws IOException {
             PercolateRequest percolateRequest = new PercolateRequest();
             percolateRequest.onlyCount(true);
-            parseDocPercolate(percolateRequest, restRequest, restChannel, client);
+            return parseDocPercolate(percolateRequest, restRequest, client);
         }
+
     }
 
-    final class RestPercolateExistingDocHandler extends BaseRestHandler {
+    private final class RestPercolateExistingDocHandler extends BaseRestHandler {
 
-        protected RestPercolateExistingDocHandler(Settings settings) {
+        RestPercolateExistingDocHandler(Settings settings) {
             super(settings);
         }
 
         @Override
-        public void handleRequest(RestRequest restRequest, RestChannel restChannel, final NodeClient client) {
+        public RestChannelConsumer prepareRequest(RestRequest restRequest, final NodeClient client) throws IOException {
             PercolateRequest percolateRequest = new PercolateRequest();
-            parseExistingDocPercolate(percolateRequest, restRequest, restChannel, client);
+            return parseExistingDocPercolate(percolateRequest, restRequest, client);
         }
+
     }
 
-    final class RestCountPercolateExistingDocHandler extends BaseRestHandler {
+    private final class RestCountPercolateExistingDocHandler extends BaseRestHandler {
 
-        protected RestCountPercolateExistingDocHandler(Settings settings) {
+        RestCountPercolateExistingDocHandler(Settings settings) {
             super(settings);
         }
 
         @Override
-        public void handleRequest(RestRequest restRequest, RestChannel restChannel, final NodeClient client) {
+        public RestChannelConsumer prepareRequest(RestRequest restRequest, final NodeClient client) throws IOException {
             PercolateRequest percolateRequest = new PercolateRequest();
             percolateRequest.onlyCount(true);
-            parseExistingDocPercolate(percolateRequest, restRequest, restChannel, client);
+            return parseExistingDocPercolate(percolateRequest, restRequest, client);
         }
+
     }
+
 }
diff --git a/modules/reindex/src/main/java/org/elasticsearch/index/reindex/AbstractBaseReindexRestHandler.java b/modules/reindex/src/main/java/org/elasticsearch/index/reindex/AbstractBaseReindexRestHandler.java
index faa2607ce2..75eb8fdafa 100644
--- a/modules/reindex/src/main/java/org/elasticsearch/index/reindex/AbstractBaseReindexRestHandler.java
+++ b/modules/reindex/src/main/java/org/elasticsearch/index/reindex/AbstractBaseReindexRestHandler.java
@@ -59,8 +59,8 @@ public abstract class AbstractBaseReindexRestHandler<
         this.action = action;
     }
 
-    protected void handleRequest(RestRequest request, RestChannel channel, NodeClient client,
-                                 boolean includeCreated, boolean includeUpdated) throws IOException {
+    protected RestChannelConsumer doPrepareRequest(RestRequest request, NodeClient client,
+                                                   boolean includeCreated, boolean includeUpdated) throws IOException {
         // Build the internal request
         Request internal = setCommonOptions(request, buildRequest(request));
 
@@ -70,8 +70,7 @@ public abstract class AbstractBaseReindexRestHandler<
             params.put(BulkByScrollTask.Status.INCLUDE_CREATED, Boolean.toString(includeCreated));
             params.put(BulkByScrollTask.Status.INCLUDE_UPDATED, Boolean.toString(includeUpdated));
 
-            client.executeLocally(action, internal, new BulkIndexByScrollResponseContentListener(channel, params));
-            return;
+            return channel -> client.executeLocally(action, internal, new BulkIndexByScrollResponseContentListener(channel, params));
         } else {
             internal.setShouldStoreResult(true);
         }
@@ -83,10 +82,9 @@ public abstract class AbstractBaseReindexRestHandler<
          */
         ActionRequestValidationException validationException = internal.validate();
         if (validationException != null) {
-            channel.sendResponse(new BytesRestResponse(channel, validationException));
-            return;
+            throw validationException;
         }
-        sendTask(channel, client.executeLocally(action, internal, LoggingTaskListener.instance()));
+        return sendTask(client.executeLocally(action, internal, LoggingTaskListener.instance()));
     }
 
     /**
@@ -116,13 +114,15 @@ public abstract class AbstractBaseReindexRestHandler<
         return request;
     }
 
-    private void sendTask(RestChannel channel, Task task) throws IOException {
-        try (XContentBuilder builder = channel.newBuilder()) {
-            builder.startObject();
-            builder.field("task", clusterService.localNode().getId() + ":" + task.getId());
-            builder.endObject();
-            channel.sendResponse(new BytesRestResponse(RestStatus.OK, builder));
-        }
+    private RestChannelConsumer sendTask(Task task) throws IOException {
+        return channel -> {
+            try (XContentBuilder builder = channel.newBuilder()) {
+                builder.startObject();
+                builder.field("task", clusterService.localNode().getId() + ":" + task.getId());
+                builder.endObject();
+                channel.sendResponse(new BytesRestResponse(RestStatus.OK, builder));
+            }
+        };
     }
 
     /**
diff --git a/modules/reindex/src/main/java/org/elasticsearch/index/reindex/RestDeleteByQueryAction.java b/modules/reindex/src/main/java/org/elasticsearch/index/reindex/RestDeleteByQueryAction.java
index 92a96a880e..d3cdf9fe09 100644
--- a/modules/reindex/src/main/java/org/elasticsearch/index/reindex/RestDeleteByQueryAction.java
+++ b/modules/reindex/src/main/java/org/elasticsearch/index/reindex/RestDeleteByQueryAction.java
@@ -25,13 +25,9 @@ import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.indices.query.IndicesQueriesRegistry;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.search.SearchRequestParsers;
-import org.elasticsearch.search.aggregations.AggregatorParsers;
-import org.elasticsearch.search.suggest.Suggesters;
 
 import java.io.IOException;
 import java.util.HashMap;
@@ -51,11 +47,11 @@ public class RestDeleteByQueryAction extends AbstractBulkByQueryRestHandler<Dele
     }
 
     @Override
-    public void handleRequest(RestRequest request, RestChannel channel, NodeClient client) throws Exception {
+    public RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
         if (false == request.hasContent()) {
             throw new ElasticsearchException("_delete_by_query requires a request body");
         }
-        handleRequest(request, channel, client, false, false);
+        return doPrepareRequest(request, client, false, false);
     }
 
     @Override
diff --git a/modules/reindex/src/main/java/org/elasticsearch/index/reindex/RestReindexAction.java b/modules/reindex/src/main/java/org/elasticsearch/index/reindex/RestReindexAction.java
index abec32de99..14fd6cb903 100644
--- a/modules/reindex/src/main/java/org/elasticsearch/index/reindex/RestReindexAction.java
+++ b/modules/reindex/src/main/java/org/elasticsearch/index/reindex/RestReindexAction.java
@@ -42,7 +42,6 @@ import org.elasticsearch.common.xcontent.json.JsonXContent;
 import org.elasticsearch.index.VersionType;
 import org.elasticsearch.index.query.QueryParseContext;
 import org.elasticsearch.index.reindex.remote.RemoteInfo;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.script.Script;
@@ -118,11 +117,11 @@ public class RestReindexAction extends AbstractBaseReindexRestHandler<ReindexReq
     }
 
     @Override
-    public void handleRequest(RestRequest request, RestChannel channel, NodeClient client) throws IOException {
+    public RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
         if (false == request.hasContent()) {
             throw new ElasticsearchException("_reindex requires a request body");
         }
-        handleRequest(request, channel, client, true, true);
+        return doPrepareRequest(request, client, true, true);
     }
 
     @Override
diff --git a/modules/reindex/src/main/java/org/elasticsearch/index/reindex/RestRethrottleAction.java b/modules/reindex/src/main/java/org/elasticsearch/index/reindex/RestRethrottleAction.java
index 26c2aad8ce..7ada5b9c9c 100644
--- a/modules/reindex/src/main/java/org/elasticsearch/index/reindex/RestRethrottleAction.java
+++ b/modules/reindex/src/main/java/org/elasticsearch/index/reindex/RestRethrottleAction.java
@@ -24,11 +24,12 @@ import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.tasks.TaskId;
 
+import java.io.IOException;
+
 import static org.elasticsearch.rest.RestRequest.Method.POST;
 import static org.elasticsearch.rest.action.admin.cluster.RestListTasksAction.listTasksResponseListener;
 
@@ -45,7 +46,7 @@ public class RestRethrottleAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         RethrottleRequest internalRequest = new RethrottleRequest();
         internalRequest.setTaskId(new TaskId(request.param("taskId")));
         Float requestsPerSecond = AbstractBaseReindexRestHandler.parseRequestsPerSecond(request);
@@ -53,6 +54,8 @@ public class RestRethrottleAction extends BaseRestHandler {
             throw new IllegalArgumentException("requests_per_second is a required parameter");
         }
         internalRequest.setRequestsPerSecond(requestsPerSecond);
-        client.execute(RethrottleAction.INSTANCE, internalRequest, listTasksResponseListener(clusterService, channel));
+        final String groupBy = request.param("group_by", "nodes");
+        return channel ->
+            client.execute(RethrottleAction.INSTANCE, internalRequest, listTasksResponseListener(clusterService, groupBy, channel));
     }
 }
diff --git a/modules/reindex/src/main/java/org/elasticsearch/index/reindex/RestUpdateByQueryAction.java b/modules/reindex/src/main/java/org/elasticsearch/index/reindex/RestUpdateByQueryAction.java
index 06adb76df7..8fe9cec14b 100644
--- a/modules/reindex/src/main/java/org/elasticsearch/index/reindex/RestUpdateByQueryAction.java
+++ b/modules/reindex/src/main/java/org/elasticsearch/index/reindex/RestUpdateByQueryAction.java
@@ -26,15 +26,11 @@ import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.ParseFieldMatcher;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.indices.query.IndicesQueriesRegistry;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.script.Script;
 import org.elasticsearch.script.ScriptService;
 import org.elasticsearch.search.SearchRequestParsers;
-import org.elasticsearch.search.aggregations.AggregatorParsers;
-import org.elasticsearch.search.suggest.Suggesters;
 
 import java.io.IOException;
 import java.util.HashMap;
@@ -56,8 +52,8 @@ public class RestUpdateByQueryAction extends AbstractBulkByQueryRestHandler<Upda
     }
 
     @Override
-    public void handleRequest(RestRequest request, RestChannel channel, NodeClient client) throws Exception {
-        handleRequest(request, channel, client, false, true);
+    public RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
+        return doPrepareRequest(request, client, false, true);
     }
 
     @Override
diff --git a/modules/reindex/src/main/java/org/elasticsearch/index/reindex/remote/RemoteRequestBuilders.java b/modules/reindex/src/main/java/org/elasticsearch/index/reindex/remote/RemoteRequestBuilders.java
index b2700618f0..107bfaf8e1 100644
--- a/modules/reindex/src/main/java/org/elasticsearch/index/reindex/remote/RemoteRequestBuilders.java
+++ b/modules/reindex/src/main/java/org/elasticsearch/index/reindex/remote/RemoteRequestBuilders.java
@@ -86,7 +86,7 @@ final class RemoteRequestBuilders {
                 for (int i = 1; i < searchRequest.source().sorts().size(); i++) {
                     sorts.append(',').append(sortToUri(searchRequest.source().sorts().get(i)));
                 }
-                params.put("sorts", sorts.toString());
+                params.put("sort", sorts.toString());
             }
         }
         if (searchRequest.source().storedFields() != null && false == searchRequest.source().storedFields().fieldNames().isEmpty()) {
diff --git a/modules/reindex/src/test/java/org/elasticsearch/index/reindex/remote/RemoteRequestBuildersTests.java b/modules/reindex/src/test/java/org/elasticsearch/index/reindex/remote/RemoteRequestBuildersTests.java
index 9bbfd175a7..aa262f0011 100644
--- a/modules/reindex/src/test/java/org/elasticsearch/index/reindex/remote/RemoteRequestBuildersTests.java
+++ b/modules/reindex/src/test/java/org/elasticsearch/index/reindex/remote/RemoteRequestBuildersTests.java
@@ -96,7 +96,7 @@ public class RemoteRequestBuildersTests extends ESTestCase {
         // Test sort:_doc for versions that support it.
         Version remoteVersion = Version.fromId(between(Version.V_2_1_0_ID, Version.CURRENT.id));
         searchRequest.source().sort("_doc");
-        assertThat(initialSearchParams(searchRequest, remoteVersion), hasEntry("sorts", "_doc:asc"));
+        assertThat(initialSearchParams(searchRequest, remoteVersion), hasEntry("sort", "_doc:asc"));
 
         // Test search_type scan for versions that don't support sort:_doc.
         remoteVersion = Version.fromId(between(0, Version.V_2_1_0_ID - 1));
@@ -106,7 +106,7 @@ public class RemoteRequestBuildersTests extends ESTestCase {
         remoteVersion = Version.fromId(between(0, Version.CURRENT.id));
         searchRequest.source().sorts().clear();
         searchRequest.source().sort("foo");
-        assertThat(initialSearchParams(searchRequest, remoteVersion), hasEntry("sorts", "foo:asc"));
+        assertThat(initialSearchParams(searchRequest, remoteVersion), hasEntry("sort", "foo:asc"));
     }
 
     public void testInitialSearchParamsFields() {
diff --git a/plugins/jvm-example/src/main/java/org/elasticsearch/plugin/example/ExampleCatAction.java b/plugins/jvm-example/src/main/java/org/elasticsearch/plugin/example/ExampleCatAction.java
index 68ec25f220..88ff7609ff 100644
--- a/plugins/jvm-example/src/main/java/org/elasticsearch/plugin/example/ExampleCatAction.java
+++ b/plugins/jvm-example/src/main/java/org/elasticsearch/plugin/example/ExampleCatAction.java
@@ -23,7 +23,6 @@ import org.elasticsearch.common.Table;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.cat.AbstractCatAction;
@@ -45,21 +44,18 @@ public class ExampleCatAction extends AbstractCatAction {
     }
 
     @Override
-    protected void doRequest(final RestRequest request, final RestChannel channel, final NodeClient client) {
+    protected RestChannelConsumer doCatRequest(final RestRequest request, final NodeClient client) {
         Table table = getTableWithHeader(request);
         table.startRow();
         table.addCell(config.getTestConfig());
         table.endRow();
-        try {
-            channel.sendResponse(RestTable.buildResponse(table, channel));
-        } catch (Exception e) {
+        return channel -> {
             try {
+                channel.sendResponse(RestTable.buildResponse(table, channel));
+            } catch (final Exception e) {
                 channel.sendResponse(new BytesRestResponse(channel, e));
-            } catch (Exception inner) {
-                inner.addSuppressed(e);
-                logger.error("failed to send failure response", inner);
             }
-        }
+        };
     }
 
     @Override
diff --git a/qa/smoke-test-http/src/test/java/org/elasticsearch/http/TestDeprecationHeaderRestAction.java b/qa/smoke-test-http/src/test/java/org/elasticsearch/http/TestDeprecationHeaderRestAction.java
index cff7b55f9d..8a998a4e53 100644
--- a/qa/smoke-test-http/src/test/java/org/elasticsearch/http/TestDeprecationHeaderRestAction.java
+++ b/qa/smoke-test-http/src/test/java/org/elasticsearch/http/TestDeprecationHeaderRestAction.java
@@ -27,11 +27,11 @@ import org.elasticsearch.common.xcontent.XContentFactory;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestStatus;
 
+import java.io.IOException;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
@@ -80,7 +80,7 @@ public class TestDeprecationHeaderRestAction extends BaseRestHandler {
 
     @SuppressWarnings("unchecked") // List<String> casts
     @Override
-    public void handleRequest(RestRequest request, RestChannel channel, NodeClient client) throws Exception {
+    public RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
         final List<String> settings;
 
         try (XContentParser parser = XContentFactory.xContent(request.content()).createParser(request.content())) {
@@ -95,14 +95,15 @@ public class TestDeprecationHeaderRestAction extends BaseRestHandler {
             }
         }
 
-        final XContentBuilder builder = channel.newBuilder();
+        return channel -> {
+            final XContentBuilder builder = channel.newBuilder();
 
-        builder.startObject().startArray("settings");
-        for (String setting : settings) {
-            builder.startObject().field(setting, SETTINGS.get(setting).getRaw(this.settings)).endObject();
-        }
-        builder.endArray().endObject();
-
-        channel.sendResponse(new BytesRestResponse(RestStatus.OK, builder));
+            builder.startObject().startArray("settings");
+            for (String setting : settings) {
+                builder.startObject().field(setting, SETTINGS.get(setting).getRaw(this.settings)).endObject();
+            }
+            builder.endArray().endObject();
+            channel.sendResponse(new BytesRestResponse(RestStatus.OK, builder));
+        };
     }
 }
diff --git a/qa/smoke-test-http/src/test/java/org/elasticsearch/http/TestResponseHeaderRestAction.java b/qa/smoke-test-http/src/test/java/org/elasticsearch/http/TestResponseHeaderRestAction.java
index a321fc0a45..a38fd3d015 100644
--- a/qa/smoke-test-http/src/test/java/org/elasticsearch/http/TestResponseHeaderRestAction.java
+++ b/qa/smoke-test-http/src/test/java/org/elasticsearch/http/TestResponseHeaderRestAction.java
@@ -23,12 +23,13 @@ import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
-import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
 import org.elasticsearch.rest.RestStatus;
 
+import java.io.IOException;
+
 public class TestResponseHeaderRestAction extends BaseRestHandler {
 
     @Inject
@@ -38,15 +39,15 @@ public class TestResponseHeaderRestAction extends BaseRestHandler {
     }
 
     @Override
-    public void handleRequest(RestRequest request, RestChannel channel, NodeClient client) {
+    public RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
         if ("password".equals(request.header("Secret"))) {
             RestResponse response = new BytesRestResponse(RestStatus.OK, "Access granted");
             response.addHeader("Secret", "granted");
-            channel.sendResponse(response);
+            return channel -> channel.sendResponse(response);
         } else {
             RestResponse response = new BytesRestResponse(RestStatus.UNAUTHORIZED, "Access denied");
             response.addHeader("Secret", "required");
-            channel.sendResponse(response);
+            return channel -> channel.sendResponse(response);
         }
     }
 }
diff --git a/rest-api-spec/src/main/resources/rest-api-spec/api/cat.snapshots.json b/rest-api-spec/src/main/resources/rest-api-spec/api/cat.snapshots.json
index 68468dd542..dc743893b5 100644
--- a/rest-api-spec/src/main/resources/rest-api-spec/api/cat.snapshots.json
+++ b/rest-api-spec/src/main/resources/rest-api-spec/api/cat.snapshots.json
@@ -3,8 +3,10 @@
     "documentation": "http://www.elastic.co/guide/en/elasticsearch/reference/master/cat-snapshots.html",
     "methods": ["GET"],
     "url": {
-      "path": "/_cat/snapshots/{repository}",
-      "paths": ["/_cat/snapshots/{repository}"],
+      "path": "/_cat/snapshots",
+      "paths": [
+        "/_cat/snapshots",
+        "/_cat/snapshots/{repository}"],
       "parts": {
         "repository": {
           "type" : "list",
diff --git a/rest-api-spec/src/main/resources/rest-api-spec/test/cat.snapshots/10_basic.yaml b/rest-api-spec/src/main/resources/rest-api-spec/test/cat.snapshots/10_basic.yaml
index c1a1fc9701..6e03ceb98c 100755
--- a/rest-api-spec/src/main/resources/rest-api-spec/test/cat.snapshots/10_basic.yaml
+++ b/rest-api-spec/src/main/resources/rest-api-spec/test/cat.snapshots/10_basic.yaml
@@ -2,7 +2,6 @@
 "Help":
   - do:
       cat.snapshots:
-        repository: test_cat_snapshots_1
         help: true
 
   - match: