Optionally require a valid content type for all rest requests with content (#22691)

This change adds a strict mode for xcontent parsing on the rest layer. The strict mode will be off by default for 5.x and in a separate commit will be enabled by default for 6.0. The strict mode, which can be enabled by setting `http.content_type.required: true` in 5.x, will require that all incoming rest requests have a valid and supported content type header before the request is dispatched. In the non-strict mode, the Content-Type header will be inspected and if it is not present or not valid, we will continue with auto detection of content like we have done previously.

The content type header is parsed to the matching XContentType value with the only exception being for plain text requests. This value is then passed on with the content bytes so that we can reduce the number of places where we need to auto-detect the content type.

As part of this, many transport requests and builders were updated to provide methods that
accepted the XContentType along with the bytes and the methods that would rely on auto-detection have been deprecated.

In the non-strict mode, deprecation warnings are issued whenever a request with body doesn't provide the Content-Type header.

See #19388 

19 files changed, 377 insertions, 110 deletions

diff --git a/core/src/main/java/org/elasticsearch/rest/AbstractRestChannel.java b/core/src/main/java/org/elasticsearch/rest/AbstractRestChannel.java
index 4b3505e97e..bdc78c82dd 100644
--- a/core/src/main/java/org/elasticsearch/rest/AbstractRestChannel.java
+++ b/core/src/main/java/org/elasticsearch/rest/AbstractRestChannel.java
@@ -20,7 +20,6 @@ package org.elasticsearch.rest;
 
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.Strings;
-import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.io.stream.BytesStreamOutput;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentFactory;
@@ -58,28 +57,37 @@ public abstract class AbstractRestChannel implements RestChannel {
 
     @Override
     public XContentBuilder newBuilder() throws IOException {
-        return newBuilder(request.hasContent() ? request.content() : null, true);
+        return newBuilder(request.getXContentType(), true);
     }
 
     @Override
     public XContentBuilder newErrorBuilder() throws IOException {
         // Disable filtering when building error responses
-        return newBuilder(request.hasContent() ? request.content() : null, false);
+        return newBuilder(request.getXContentType(), false);
     }
 
+    /**
+     * Creates a new {@link XContentBuilder} for a response to be sent using this channel. The builder's type is determined by the following
+     * logic. If the request has a format parameter that will be used to attempt to map to an {@link XContentType}. If there is no format
+     * parameter, the HTTP Accept header is checked to see if it can be matched to a {@link XContentType}. If this first attempt to map
+     * fails, the request content type will be used if the value is not {@code null}; if the value is {@code null} the output format falls
+     * back to JSON.
+     */
     @Override
-    public XContentBuilder newBuilder(@Nullable BytesReference autoDetectSource, boolean useFiltering) throws IOException {
-        XContentType contentType = XContentType.fromMediaTypeOrFormat(format);
-        if (contentType == null) {
-            // try and guess it from the auto detect source
-            if (autoDetectSource != null) {
-                contentType = XContentFactory.xContentType(autoDetectSource);
+    public XContentBuilder newBuilder(@Nullable XContentType requestContentType, boolean useFiltering) throws IOException {
+        // try to determine the response content type from the media type or the format query string parameter, with the format parameter
+        // taking precedence over the Accept header
+        XContentType responseContentType = XContentType.fromMediaTypeOrFormat(format);
+        if (responseContentType == null) {
+            if (requestContentType != null) {
+                // if there was a parsed content-type for the incoming request use that since no format was specified using the query
+                // string parameter or the HTTP Accept header
+                responseContentType = requestContentType;
+            } else {
+                // default to JSON output when all else fails
+                responseContentType = XContentType.JSON;
             }
         }
-        if (contentType == null) {
-            // default to JSON
-            contentType = XContentType.JSON;
-        }
 
         Set<String> includes = Collections.emptySet();
         Set<String> excludes = Collections.emptySet();
@@ -89,7 +97,7 @@ public abstract class AbstractRestChannel implements RestChannel {
             excludes = filters.stream().filter(EXCLUDE_FILTER).map(f -> f.substring(1)).collect(toSet());
         }
 
-        XContentBuilder builder = new XContentBuilder(XContentFactory.xContent(contentType), bytesOutput(), includes, excludes);
+        XContentBuilder builder = new XContentBuilder(XContentFactory.xContent(responseContentType), bytesOutput(), includes, excludes);
         if (pretty) {
             builder.prettyPrint().lfAtEnd();
         }
@@ -125,5 +133,4 @@ public abstract class AbstractRestChannel implements RestChannel {
     public boolean detailedErrorsEnabled() {
         return detailedErrorsEnabled;
     }
-
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/BytesRestResponse.java b/core/src/main/java/org/elasticsearch/rest/BytesRestResponse.java
index 2d45bc1765..a7b709625c 100644
--- a/core/src/main/java/org/elasticsearch/rest/BytesRestResponse.java
+++ b/core/src/main/java/org/elasticsearch/rest/BytesRestResponse.java
@@ -30,6 +30,7 @@ import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.logging.ESLoggerFactory;
 import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.json.JsonXContent;
 import org.elasticsearch.common.xcontent.XContentParser;
 
 import java.io.IOException;
@@ -146,6 +147,13 @@ public class BytesRestResponse extends RestResponse {
         return builder;
     }
 
+    static BytesRestResponse createSimpleErrorResponse(RestStatus status, String errorMessage) throws IOException {
+        return new BytesRestResponse(status, JsonXContent.contentBuilder().startObject()
+            .field("error", errorMessage)
+            .field("status", status.getStatus())
+            .endObject());
+    }
+
     public static ElasticsearchStatusException errorFromXContent(XContentParser parser) throws IOException {
         XContentParser.Token token = parser.nextToken();
         ensureExpectedToken(XContentParser.Token.START_OBJECT, token, parser::getTokenLocation);
diff --git a/core/src/main/java/org/elasticsearch/rest/RestChannel.java b/core/src/main/java/org/elasticsearch/rest/RestChannel.java
index 2a56313fd8..8c8346f0ef 100644
--- a/core/src/main/java/org/elasticsearch/rest/RestChannel.java
+++ b/core/src/main/java/org/elasticsearch/rest/RestChannel.java
@@ -20,9 +20,9 @@
 package org.elasticsearch.rest;
 
 import org.elasticsearch.common.Nullable;
-import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.io.stream.BytesStreamOutput;
 import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentType;
 
 import java.io.IOException;
 
@@ -35,7 +35,7 @@ public interface RestChannel {
 
     XContentBuilder newErrorBuilder() throws IOException;
 
-    XContentBuilder newBuilder(@Nullable BytesReference autoDetectSource, boolean useFiltering) throws IOException;
+    XContentBuilder newBuilder(@Nullable XContentType xContentType, boolean useFiltering) throws IOException;
 
     BytesStreamOutput bytesOutput();
 
@@ -47,5 +47,4 @@ public interface RestChannel {
     boolean detailedErrorsEnabled();
 
     void sendResponse(RestResponse response);
-
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/RestController.java b/core/src/main/java/org/elasticsearch/rest/RestController.java
index 5ac82b7e45..fd3587df2a 100644
--- a/core/src/main/java/org/elasticsearch/rest/RestController.java
+++ b/core/src/main/java/org/elasticsearch/rest/RestController.java
@@ -22,18 +22,19 @@ package org.elasticsearch.rest;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
+import java.util.List;
 import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
 import java.util.function.UnaryOperator;
 
 import org.apache.logging.log4j.message.ParameterizedMessage;
-import org.apache.logging.log4j.util.Supplier;
 import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.Nullable;
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.breaker.CircuitBreaker;
 import org.elasticsearch.common.bytes.BytesArray;
-import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.component.AbstractComponent;
 import org.elasticsearch.common.io.Streams;
 import org.elasticsearch.common.io.stream.BytesStreamOutput;
@@ -42,12 +43,15 @@ import org.elasticsearch.common.path.PathTrie;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.common.xcontent.XContentBuilder;
-import org.elasticsearch.http.HttpServerTransport;
+import org.elasticsearch.common.xcontent.XContentFactory;
+import org.elasticsearch.http.HttpTransportSettings;
 import org.elasticsearch.indices.breaker.CircuitBreakerService;
+import org.elasticsearch.common.xcontent.XContentType;
 
 import static org.elasticsearch.rest.RestStatus.BAD_REQUEST;
 import static org.elasticsearch.rest.RestStatus.FORBIDDEN;
 import static org.elasticsearch.rest.RestStatus.INTERNAL_SERVER_ERROR;
+import static org.elasticsearch.rest.RestStatus.NOT_ACCEPTABLE;
 import static org.elasticsearch.rest.RestStatus.OK;
 
 public class RestController extends AbstractComponent {
@@ -67,6 +71,10 @@ public class RestController extends AbstractComponent {
     /** Rest headers that are copied to internal requests made during a rest request. */
     private final Set<String> headersToCopy;
 
+    private final boolean isContentTypeRequired;
+
+    private final DeprecationLogger deprecationLogger;
+
     public RestController(Settings settings, Set<String> headersToCopy, UnaryOperator<RestHandler> handlerWrapper,
                           NodeClient client, CircuitBreakerService circuitBreakerService) {
         super(settings);
@@ -77,10 +85,10 @@ public class RestController extends AbstractComponent {
         this.handlerWrapper = handlerWrapper;
         this.client = client;
         this.circuitBreakerService = circuitBreakerService;
+        this.isContentTypeRequired = HttpTransportSettings.SETTING_HTTP_CONTENT_TYPE_REQUIRED.get(settings);
+        this.deprecationLogger = new DeprecationLogger(logger);
     }
 
-
-
     /**
      * Registers a REST handler to be executed when the provided {@code method} and {@code path} match the request.
      *
@@ -165,15 +173,22 @@ public class RestController extends AbstractComponent {
         }
         RestChannel responseChannel = channel;
         try {
-            int contentLength = request.content().length();
-            if (canTripCircuitBreaker(request)) {
-                inFlightRequestsBreaker(circuitBreakerService).addEstimateBytesAndMaybeBreak(contentLength, "<http_request>");
+            final int contentLength = request.hasContent() ? request.content().length() : 0;
+            assert contentLength >= 0 : "content length was negative, how is that possible?";
+            final RestHandler handler = getHandler(request);
+
+            if (contentLength > 0 && hasContentTypeOrCanAutoDetect(request, handler) == false) {
+                sendContentTypeErrorMessage(request, responseChannel);
             } else {
-                inFlightRequestsBreaker(circuitBreakerService).addWithoutBreaking(contentLength);
+                if (canTripCircuitBreaker(request)) {
+                    inFlightRequestsBreaker(circuitBreakerService).addEstimateBytesAndMaybeBreak(contentLength, "<http_request>");
+                } else {
+                    inFlightRequestsBreaker(circuitBreakerService).addWithoutBreaking(contentLength);
+                }
+                // iff we could reserve bytes for the request we need to send the response also over this channel
+                responseChannel = new ResourceHandlingHttpChannel(channel, circuitBreakerService, contentLength);
+                dispatchRequest(request, responseChannel, client, threadContext, handler);
             }
-            // iff we could reserve bytes for the request we need to send the response also over this channel
-            responseChannel = new ResourceHandlingHttpChannel(channel, circuitBreakerService, contentLength);
-            dispatchRequest(request, responseChannel, client, threadContext);
         } catch (Exception e) {
             try {
                 responseChannel.sendResponse(new BytesRestResponse(channel, e));
@@ -185,33 +200,75 @@ public class RestController extends AbstractComponent {
         }
     }
 
-    void dispatchRequest(final RestRequest request, final RestChannel channel, final NodeClient client, ThreadContext threadContext) throws Exception {
-        if (!checkRequestParameters(request, channel)) {
-            return;
-        }
-        try (ThreadContext.StoredContext ignored = threadContext.stashContext()) {
-            for (String key : headersToCopy) {
-                String httpHeader = request.header(key);
-                if (httpHeader != null) {
-                    threadContext.putHeader(key, httpHeader);
+    void dispatchRequest(final RestRequest request, final RestChannel channel, final NodeClient client, ThreadContext threadContext,
+                         final RestHandler handler) throws Exception {
+        if (checkRequestParameters(request, channel) == false) {
+            channel.sendResponse(BytesRestResponse.createSimpleErrorResponse(BAD_REQUEST, "error traces in responses are disabled."));
+        } else {
+            try (ThreadContext.StoredContext ignored = threadContext.stashContext()) {
+                for (String key : headersToCopy) {
+                    String httpHeader = request.header(key);
+                    if (httpHeader != null) {
+                        threadContext.putHeader(key, httpHeader);
+                    }
                 }
-            }
 
-            final RestHandler handler = getHandler(request);
+                if (handler == null) {
+                    if (request.method() == RestRequest.Method.OPTIONS) {
+                        // when we have OPTIONS request, simply send OK by default (with the Access Control Origin header which gets automatically added)
 
-            if (handler == null) {
-                if (request.method() == RestRequest.Method.OPTIONS) {
-                    // when we have OPTIONS request, simply send OK by default (with the Access Control Origin header which gets automatically added)
-                    channel.sendResponse(new BytesRestResponse(OK, BytesRestResponse.TEXT_CONTENT_TYPE, BytesArray.EMPTY));
+                        channel.sendResponse(new BytesRestResponse(OK, BytesRestResponse.TEXT_CONTENT_TYPE, BytesArray.EMPTY));
+                    } else {
+                        final String msg = "No handler found for uri [" + request.uri() + "] and method [" + request.method() + "]";
+                        channel.sendResponse(new BytesRestResponse(BAD_REQUEST, msg));
+                    }
                 } else {
-                    final String msg = "No handler found for uri [" + request.uri() + "] and method [" + request.method() + "]";
-                    channel.sendResponse(new BytesRestResponse(BAD_REQUEST, msg));
+                    final RestHandler wrappedHandler = Objects.requireNonNull(handlerWrapper.apply(handler));
+                    wrappedHandler.handleRequest(request, channel, client);
                 }
+            }
+        }
+    }
+
+    /**
+     * If a request contains content, this method will return {@code true} if the {@code Content-Type} header is present, matches an
+     * {@link XContentType} or the request is plain text, and content type is required. If content type is not required then this method
+     * returns true unless a content type could not be inferred from the body and the rest handler does not support plain text
+     */
+    private boolean hasContentTypeOrCanAutoDetect(final RestRequest restRequest, final RestHandler restHandler) {
+        if (restRequest.getXContentType() == null) {
+            if (restHandler != null && restHandler.supportsPlainText()) {
+                // content type of null with a handler that supports plain text gets through for now. Once we remove plain text this can
+                // be removed!
+                deprecationLogger.deprecated("Plain text request bodies are deprecated. Use request parameters or body " +
+                    "in a supported format.");
+            } else if (isContentTypeRequired) {
+                return false;
             } else {
-                final RestHandler wrappedHandler = Objects.requireNonNull(handlerWrapper.apply(handler));
-                wrappedHandler.handleRequest(request, channel, client);
+                deprecationLogger.deprecated("Content type detection for rest requests is deprecated. Specify the content type using " +
+                    "the [Content-Type] header.");
+                XContentType xContentType = XContentFactory.xContentType(restRequest.content());
+                if (xContentType == null) {
+                    return false;
+                } else {
+                    restRequest.setXContentType(xContentType);
+                }
             }
         }
+        return true;
+    }
+
+    private void sendContentTypeErrorMessage(RestRequest restRequest, RestChannel channel) throws IOException {
+        final List<String> contentTypeHeader = restRequest.getAllHeaderValues("Content-Type");
+        final String errorMessage;
+        if (contentTypeHeader == null) {
+            errorMessage = "Content-Type header is missing";
+        } else {
+            errorMessage = "Content-Type header [" +
+                Strings.collectionToCommaDelimitedString(restRequest.getAllHeaderValues("Content-Type")) + "] is not supported";
+        }
+
+        channel.sendResponse(BytesRestResponse.createSimpleErrorResponse(NOT_ACCEPTABLE, errorMessage));
     }
 
     /**
@@ -222,15 +279,6 @@ public class RestController extends AbstractComponent {
         // error_trace cannot be used when we disable detailed errors
         // 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();
-                RestResponse response = new BytesRestResponse(BAD_REQUEST, builder);
-                response.addHeader("Content-Type", "application/json");
-                channel.sendResponse(response);
-            } catch (IOException e) {
-                logger.warn("Failed to send response", e);
-            }
             return false;
         }
 
@@ -312,8 +360,8 @@ public class RestController extends AbstractComponent {
         }
 
         @Override
-        public XContentBuilder newBuilder(@Nullable BytesReference autoDetectSource, boolean useFiltering) throws IOException {
-            return delegate.newBuilder(autoDetectSource, useFiltering);
+        public XContentBuilder newBuilder(@Nullable XContentType xContentType, boolean useFiltering) throws IOException {
+            return delegate.newBuilder(xContentType, useFiltering);
         }
 
         @Override
diff --git a/core/src/main/java/org/elasticsearch/rest/RestHandler.java b/core/src/main/java/org/elasticsearch/rest/RestHandler.java
index 393e425baf..740b98ae73 100644
--- a/core/src/main/java/org/elasticsearch/rest/RestHandler.java
+++ b/core/src/main/java/org/elasticsearch/rest/RestHandler.java
@@ -28,7 +28,6 @@ public interface RestHandler {
 
     /**
      * Handles a rest request.
-     *
      * @param request The request to handle
      * @param channel The channel to write the request response to
      * @param client A client to use to make internal requests on behalf of the original request
@@ -38,4 +37,13 @@ public interface RestHandler {
     default boolean canTripCircuitBreaker() {
         return true;
     }
+
+    /**
+     * Indicates if a RestHandler supports plain text bodies
+     * @deprecated use request parameters or bodies that can be parsed with XContent!
+     */
+    @Deprecated
+    default boolean supportsPlainText() {
+        return false;
+    }
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/RestRequest.java b/core/src/main/java/org/elasticsearch/rest/RestRequest.java
index b4b0614f8a..7d41faf12e 100644
--- a/core/src/main/java/org/elasticsearch/rest/RestRequest.java
+++ b/core/src/main/java/org/elasticsearch/rest/RestRequest.java
@@ -19,6 +19,7 @@
 
 package org.elasticsearch.rest;
 
+import org.apache.lucene.util.SetOnce;
 import org.elasticsearch.ElasticsearchParseException;
 import org.elasticsearch.common.Booleans;
 import org.elasticsearch.common.CheckedConsumer;
@@ -26,20 +27,26 @@ import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.collect.Tuple;
+import org.elasticsearch.common.logging.DeprecationLogger;
+import org.elasticsearch.common.logging.Loggers;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.XContentFactory;
 import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.common.xcontent.XContentType;
 
 import java.io.IOException;
 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.regex.Pattern;
 import java.util.stream.Collectors;
 
 import static org.elasticsearch.common.unit.ByteSizeValue.parseBytesSizeValue;
@@ -47,12 +54,25 @@ import static org.elasticsearch.common.unit.TimeValue.parseTimeValue;
 
 public abstract class RestRequest implements ToXContent.Params {
 
+    private static final DeprecationLogger DEPRECATION_LOGGER = new DeprecationLogger(Loggers.getLogger(RestRequest.class));
+    // tchar pattern as defined by RFC7230 section 3.2.6
+    private static final Pattern TCHAR_PATTERN = Pattern.compile("[a-zA-z0-9!#$%&'*+\\-.\\^_`|~]+");
+
     private final NamedXContentRegistry xContentRegistry;
     private final Map<String, String> params;
+    private final Map<String, List<String>> headers;
     private final String rawPath;
     private final Set<String> consumedParams = new HashSet<>();
+    private final SetOnce<XContentType> xContentType = new SetOnce<>();
 
-    public RestRequest(NamedXContentRegistry xContentRegistry, String uri) {
+    /**
+     * Creates a new RestRequest
+     * @param xContentRegistry the xContentRegistry to use when parsing XContent
+     * @param uri the URI of the request that potentially contains request parameters
+     * @param headers a map of the headers. This map should implement a Case-Insensitive hashing for keys as HTTP header names are case
+     *                insensitive
+     */
+    public RestRequest(NamedXContentRegistry xContentRegistry, String uri, Map<String, List<String>> headers) {
         this.xContentRegistry = xContentRegistry;
         final Map<String, String> params = new HashMap<>();
         int pathEndPos = uri.indexOf('?');
@@ -63,12 +83,32 @@ public abstract class RestRequest implements ToXContent.Params {
             RestUtils.decodeQueryString(uri, pathEndPos + 1, params);
         }
         this.params = params;
+        this.headers = Collections.unmodifiableMap(headers);
+        final List<String> contentType = getAllHeaderValues("Content-Type");
+        final XContentType xContentType = parseContentType(contentType);
+        if (xContentType != null) {
+            this.xContentType.set(xContentType);
+        }
     }
 
-    public RestRequest(NamedXContentRegistry xContentRegistry, Map<String, String> params, String path) {
+    /**
+     * Creates a new RestRequest
+     * @param xContentRegistry the xContentRegistry to use when parsing XContent
+     * @param params the parameters of the request
+     * @param path the path of the request. This should not contain request parameters
+     * @param headers a map of the headers. This map should implement a Case-Insensitive hashing for keys as HTTP header names are case
+     *                insensitive
+     */
+    public RestRequest(NamedXContentRegistry xContentRegistry, Map<String, String> params, String path, Map<String, List<String>> headers) {
         this.xContentRegistry = xContentRegistry;
         this.params = params;
         this.rawPath = path;
+        this.headers = Collections.unmodifiableMap(headers);
+        final List<String> contentType = getAllHeaderValues("Content-Type");
+        final XContentType xContentType = parseContentType(contentType);
+        if (xContentType != null) {
+            this.xContentType.set(xContentType);
+        }
     }
 
     public enum Method {
@@ -100,9 +140,53 @@ public abstract class RestRequest implements ToXContent.Params {
 
     public abstract BytesReference content();
 
-    public abstract String header(String name);
+    /**
+     * Get the value of the header or {@code null} if not found. This method only retrieves the first header value if multiple values are
+     * sent. Use of {@link #getAllHeaderValues(String)} should be preferred
+     */
+    public final String header(String name) {
+        List<String> values = headers.get(name);
+        if (values != null && values.isEmpty() == false) {
+            return values.get(0);
+        }
+        return null;
+    }
+
+    /**
+     * Get all values for the header or {@code null} if the header was not found
+     */
+    public final List<String> getAllHeaderValues(String name) {
+        List<String> values = headers.get(name);
+        if (values != null) {
+            return Collections.unmodifiableList(values);
+        }
+        return null;
+    }
+
+    /**
+     * Get all of the headers and values associated with the headers. Modifications of this map are not supported.
+     */
+    public final Map<String, List<String>> getHeaders() {
+        return headers;
+    }
 
-    public abstract Iterable<Map.Entry<String, String>> headers();
+    /**
+     * The {@link XContentType} that was parsed from the {@code Content-Type} header. This value will be {@code null} in the case of
+     * a request without a valid {@code Content-Type} header, a request without content ({@link #hasContent()}, or a plain text request
+     */
+    @Nullable
+    public final XContentType getXContentType() {
+        return xContentType.get();
+    }
+
+    /**
+     * Sets the {@link XContentType}
+     * @deprecated this is only used to allow BWC with content-type detection
+     */
+    @Deprecated
+    final void setXContentType(XContentType xContentType) {
+        this.xContentType.set(xContentType);
+    }
 
     @Nullable
     public SocketAddress getRemoteAddress() {
@@ -254,8 +338,10 @@ public abstract class RestRequest implements ToXContent.Params {
         BytesReference content = content();
         if (content.length() == 0) {
             throw new ElasticsearchParseException("Body required");
+        } else if (xContentType.get() == null) {
+            throw new IllegalStateException("unknown content type");
         }
-        return XContentFactory.xContent(content).createParser(xContentRegistry, content);
+        return xContentType.get().xContent().createParser(xContentRegistry, content);
     }
 
     /**
@@ -283,11 +369,12 @@ public abstract class RestRequest implements ToXContent.Params {
      * if you need to handle the absence request content gracefully.
      */
     public final XContentParser contentOrSourceParamParser() throws IOException {
-        BytesReference content = contentOrSourceParam();
+        Tuple<XContentType, BytesReference> tuple = contentOrSourceParam();
+        BytesReference content = tuple.v2();
         if (content.length() == 0) {
             throw new ElasticsearchParseException("Body required");
         }
-        return XContentFactory.xContent(content).createParser(xContentRegistry, content);
+        return tuple.v1().xContent().createParser(xContentRegistry, content);
     }
 
     /**
@@ -296,9 +383,11 @@ public abstract class RestRequest implements ToXContent.Params {
      * back to the user when there isn't request content.
      */
     public final void withContentOrSourceParamParserOrNull(CheckedConsumer<XContentParser, IOException> withParser) throws IOException {
-        BytesReference content = contentOrSourceParam();
+        Tuple<XContentType, BytesReference> tuple = contentOrSourceParam();
+        BytesReference content = tuple.v2();
+        XContentType xContentType = tuple.v1();
         if (content.length() > 0) {
-            try (XContentParser parser = XContentFactory.xContent(content).createParser(xContentRegistry, content)) {
+            try (XContentParser parser = xContentType.xContent().createParser(xContentRegistry, content)) {
                 withParser.accept(parser);
             }
         } else {
@@ -310,7 +399,66 @@ public abstract class RestRequest implements ToXContent.Params {
      * Get the content of the request or the contents of the {@code source} param. Prefer {@link #contentOrSourceParamParser()} or
      * {@link #withContentOrSourceParamParserOrNull(CheckedConsumer)} if you need a parser.
      */
-    public final BytesReference contentOrSourceParam() {
+    public final Tuple<XContentType, BytesReference> contentOrSourceParam() {
+        if (hasContent()) {
+            if (xContentType.get() == null) {
+                throw new IllegalStateException("unknown content type");
+            }
+            return new Tuple<>(xContentType.get(), content());
+        }
+
+        String source = param("source");
+        String typeParam = param("source_content_type");
+        if (source != null) {
+            BytesArray bytes = new BytesArray(source);
+            final XContentType xContentType;
+            if (typeParam != null) {
+                xContentType = parseContentType(Collections.singletonList(typeParam));
+            } else {
+                DEPRECATION_LOGGER.deprecated("Deprecated use of the [source] parameter without the [source_content_type] parameter. Use " +
+                    "the [source_content_type] parameter to specify the content type of the source such as [application/json]");
+                xContentType = XContentFactory.xContentType(bytes);
+            }
+
+            if (xContentType == null) {
+                throw new IllegalStateException("could not determine source content type");
+            }
+            return new Tuple<>(xContentType, bytes);
+        }
+        return new Tuple<>(XContentType.JSON, BytesArray.EMPTY);
+    }
+
+    /**
+     * Call a consumer with the parser for the contents of this request if it has contents, otherwise with a parser for the {@code source}
+     * parameter if there is one, otherwise with {@code null}. Use {@link #contentOrSourceParamParser()} if you should throw an exception
+     * back to the user when there isn't request content. This version allows for plain text content
+     */
+    @Deprecated
+    public final void withContentOrSourceParamParserOrNullLenient(CheckedConsumer<XContentParser, IOException> withParser)
+        throws IOException {
+        if (hasContent() && xContentType.get() == null) {
+            withParser.accept(null);
+        }
+
+        Tuple<XContentType, BytesReference> tuple = contentOrSourceParam();
+        BytesReference content = tuple.v2();
+        XContentType xContentType = tuple.v1();
+        if (content.length() > 0) {
+            try (XContentParser parser = xContentType.xContent().createParser(xContentRegistry, content)) {
+                withParser.accept(parser);
+            }
+        } else {
+            withParser.accept(null);
+        }
+    }
+
+    /**
+     * Get the content of the request or the contents of the {@code source} param without the xcontent type. This is useful the request can
+     * accept non xcontent values.
+     * @deprecated we should only take xcontent
+     */
+    @Deprecated
+    public final BytesReference getContentOrSourceParamOnly() {
         if (hasContent()) {
             return content();
         }
@@ -320,4 +468,29 @@ public abstract class RestRequest implements ToXContent.Params {
         }
         return BytesArray.EMPTY;
     }
+
+    /**
+     * Parses the given content type string for the media type. This method currently ignores parameters.
+     */
+    // TODO stop ignoring parameters such as charset...
+    private static XContentType parseContentType(List<String> header) {
+        if (header == null || header.isEmpty()) {
+            return null;
+        } else if (header.size() > 1) {
+            throw new IllegalArgumentException("only one Content-Type header should be provided");
+        }
+
+        String rawContentType = header.get(0);
+        final String[] elements = rawContentType.split("[ \t]*;");
+        if (elements.length > 0) {
+            final String[] splitMediaType = elements[0].split("/");
+            if (splitMediaType.length == 2 && TCHAR_PATTERN.matcher(splitMediaType[0]).matches()
+                && TCHAR_PATTERN.matcher(splitMediaType[1].trim()).matches()) {
+                return XContentType.fromMediaType(elements[0]);
+            } else {
+                throw new IllegalArgumentException("invalid Content-Type header [" + rawContentType + "]");
+            }
+        }
+        throw new IllegalArgumentException("empty Content-Type header");
+    }
 }
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 cee46eba04..358c3656ce 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
@@ -65,7 +65,7 @@ public class RestPutStoredScriptAction extends BaseRestHandler {
                 "specifying lang [" + lang + "] as part of the url path is deprecated, use request content instead");
         }
 
-        PutStoredScriptRequest putRequest = new PutStoredScriptRequest(id, lang, content);
+        PutStoredScriptRequest putRequest = new PutStoredScriptRequest(id, lang, content, request.getXContentType());
         return channel -> client.admin().cluster().putStoredScript(putRequest, 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 ea0cf7bf0c..8c314efee6 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
@@ -42,7 +42,7 @@ public class RestCreateIndexAction extends BaseRestHandler {
     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());
+            createIndexRequest.source(request.content(), request.getXContentType());
         }
         createIndexRequest.updateAllTypes(request.paramAsBoolean("update_all_types", false));
         createIndexRequest.timeout(request.paramAsTime("timeout", createIndexRequest.timeout()));
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 c48ed3716d..b376f3dab3 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
@@ -57,7 +57,7 @@ public class RestPutIndexTemplateAction extends BaseRestHandler {
         putRequest.masterNodeTimeout(request.paramAsTime("master_timeout", putRequest.masterNodeTimeout()));
         putRequest.create(request.paramAsBoolean("create", false));
         putRequest.cause(request.param("cause", ""));
-        putRequest.source(request.content());
+        putRequest.source(request.content(), request.getXContentType());
         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 60c434c6b0..06fe679264 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
@@ -20,7 +20,6 @@
 package org.elasticsearch.rest.action.admin.indices;
 
 import org.elasticsearch.action.admin.indices.mapping.put.PutMappingRequest;
-import org.elasticsearch.action.admin.indices.mapping.put.PutMappingResponse;
 import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.Strings;
@@ -65,11 +64,11 @@ public class RestPutMappingAction extends BaseRestHandler {
     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());
+        putMappingRequest.source(request.content(), request.getXContentType());
         putMappingRequest.updateAllTypes(request.paramAsBoolean("update_all_types", false));
         putMappingRequest.timeout(request.paramAsTime("timeout", putMappingRequest.timeout()));
         putMappingRequest.masterNodeTimeout(request.paramAsTime("master_timeout", putMappingRequest.masterNodeTimeout()));
         putMappingRequest.indicesOptions(IndicesOptions.fromRequest(request, putMappingRequest.indicesOptions()));
-        return channel -> client.admin().indices().putMapping(putMappingRequest, new AcknowledgedRestListener<PutMappingResponse>(channel));
+        return channel -> client.admin().indices().putMapping(putMappingRequest, new AcknowledgedRestListener<>(channel));
     }
 }
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 9e38fb5e07..47037460e2 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
@@ -66,7 +66,9 @@ public class RestUpdateSettingsAction extends BaseRestHandler {
         Settings.Builder updateSettings = Settings.builder();
         String bodySettingsStr = request.content().utf8ToString();
         if (Strings.hasText(bodySettingsStr)) {
-            Settings buildSettings = Settings.builder().loadFromSource(bodySettingsStr).build();
+            Settings buildSettings = Settings.builder()
+                .loadFromSource(bodySettingsStr, request.getXContentType())
+                .build();
             for (Map.Entry<String, String> entry : buildSettings.getAsMap().entrySet()) {
                 String key = entry.getKey();
                 String value = entry.getValue();
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 af0faac553..8fdf2792a4 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
@@ -93,7 +93,7 @@ public class RestBulkAction extends BaseRestHandler {
         bulkRequest.timeout(request.paramAsTime("timeout", BulkShardRequest.DEFAULT_TIMEOUT));
         bulkRequest.setRefreshPolicy(request.param("refresh"));
         bulkRequest.add(request.content(), defaultIndex, defaultType, defaultRouting, defaultFields,
-            defaultFetchSourceContext, defaultPipeline, null, allowExplicitIndex);
+            defaultFetchSourceContext, defaultPipeline, null, allowExplicitIndex, request.getXContentType());
 
         return channel -> client.bulk(bulkRequest, new RestBuilderListener<BulkResponse>(channel) {
             @Override
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 728e1ff599..83d424ed74 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
@@ -66,7 +66,7 @@ public class RestGetSourceAction extends BaseRestHandler {
                 client.get(getRequest, new RestResponseListener<GetResponse>(channel) {
                     @Override
                     public RestResponse buildResponse(GetResponse response) throws Exception {
-                        XContentBuilder builder = channel.newBuilder(response.getSourceInternal(), false);
+                        XContentBuilder builder = channel.newBuilder(request.getXContentType(), false);
                         if (response.isSourceEmpty()) { // check if doc source (or doc itself) is missing
                             return new BytesRestResponse(NOT_FOUND, builder);
                         } else {
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 042b0d57b5..ddaf226875 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
@@ -65,7 +65,7 @@ public class RestIndexAction extends BaseRestHandler {
         indexRequest.routing(request.param("routing"));
         indexRequest.parent(request.param("parent")); // order is important, set it after routing, so it will set the routing
         indexRequest.setPipeline(request.param("pipeline"));
-        indexRequest.source(request.content());
+        indexRequest.source(request.content(), request.getXContentType());
         indexRequest.timeout(request.paramAsTime("timeout", IndexRequest.DEFAULT_TIMEOUT));
         indexRequest.setRefreshPolicy(request.param("refresh"));
         indexRequest.version(RestActions.parseVersion(request));
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 b7a8e6366b..2496c9b4a2 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
@@ -21,7 +21,10 @@ package org.elasticsearch.rest.action.ingest;
 
 import org.elasticsearch.action.ingest.PutPipelineRequest;
 import org.elasticsearch.client.node.NodeClient;
+import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
@@ -38,7 +41,8 @@ public class RestPutPipelineAction extends BaseRestHandler {
 
     @Override
     public RestChannelConsumer prepareRequest(RestRequest restRequest, NodeClient client) throws IOException {
-        PutPipelineRequest request = new PutPipelineRequest(restRequest.param("id"), restRequest.contentOrSourceParam());
+        Tuple<XContentType, BytesReference> sourceTuple = restRequest.contentOrSourceParam();
+        PutPipelineRequest request = new PutPipelineRequest(restRequest.param("id"), sourceTuple.v2(), sourceTuple.v1());
         request.masterNodeTimeout(restRequest.paramAsTime("master_timeout", request.masterNodeTimeout()));
         request.timeout(restRequest.paramAsTime("timeout", request.timeout()));
         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 5885dc9df6..9dbe1808a8 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
@@ -21,7 +21,10 @@ package org.elasticsearch.rest.action.ingest;
 
 import org.elasticsearch.action.ingest.SimulatePipelineRequest;
 import org.elasticsearch.client.node.NodeClient;
+import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
@@ -40,7 +43,8 @@ public class RestSimulatePipelineAction extends BaseRestHandler {
 
     @Override
     public RestChannelConsumer prepareRequest(RestRequest restRequest, NodeClient client) throws IOException {
-        SimulatePipelineRequest request = new SimulatePipelineRequest(restRequest.contentOrSourceParam());
+        Tuple<XContentType, BytesReference> sourceTuple = restRequest.contentOrSourceParam();
+        SimulatePipelineRequest request = new SimulatePipelineRequest(sourceTuple.v2(), sourceTuple.v1());
         request.setId(restRequest.param("id"));
         request.setVerbose(restRequest.paramAsBoolean("verbose", false));
         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 e7ee7c7397..47252f5a10 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
@@ -20,12 +20,10 @@
 package org.elasticsearch.rest.action.search;
 
 import org.elasticsearch.action.search.ClearScrollRequest;
-import org.elasticsearch.action.search.ClearScrollResponse;
 import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesReference;
 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.RestController;
@@ -50,26 +48,34 @@ public class RestClearScrollAction extends BaseRestHandler {
         String scrollIds = request.param("scroll_id");
         ClearScrollRequest clearRequest = new ClearScrollRequest();
         clearRequest.setScrollIds(Arrays.asList(splitScrollIds(scrollIds)));
-        BytesReference body = request.contentOrSourceParam();
-        if (body.length() > 0) {
-            if (XContentFactory.xContentType(body) == null) {
-                scrollIds = body.utf8ToString();
-                clearRequest.setScrollIds(Arrays.asList(splitScrollIds(scrollIds)));
+        request.withContentOrSourceParamParserOrNullLenient((xContentParser -> {
+            if (xContentParser == null) {
+                if (request.hasContent()) {
+                    // TODO: why do we accept this plain text value? maybe we can just use the scroll params?
+                    BytesReference body = request.content();
+                    String bodyScrollIds = body.utf8ToString();
+                    clearRequest.setScrollIds(Arrays.asList(splitScrollIds(bodyScrollIds)));
+                }
             } else {
                 // NOTE: if rest request with xcontent body has request parameters, these parameters does not override xcontent value
                 clearRequest.setScrollIds(null);
-                try (XContentParser parser = request.contentOrSourceParamParser()) {
-                    buildFromContent(parser, clearRequest);
+                try {
+                    buildFromContent(xContentParser, clearRequest);
                 } catch (IOException e) {
                     throw new IllegalArgumentException("Failed to parse request body", e);
                 }
             }
-        }
+        }));
 
-        return channel -> client.clearScroll(clearRequest, new RestStatusToXContentListener<ClearScrollResponse>(channel));
+        return channel -> client.clearScroll(clearRequest, new RestStatusToXContentListener<>(channel));
+    }
+
+    @Override
+    public boolean supportsPlainText() {
+        return true;
     }
 
-    public static String[] splitScrollIds(String scrollIds) {
+    private static String[] splitScrollIds(String scrollIds) {
         if (scrollIds == null) {
             return Strings.EMPTY_ARRAY;
         }
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 567823a6e2..a803958618 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
@@ -26,10 +26,11 @@ import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContent;
-import org.elasticsearch.common.xcontent.XContentFactory;
 import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.index.query.QueryParseContext;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.RestController;
@@ -93,7 +94,7 @@ public class RestMultiSearchAction extends BaseRestHandler {
     }
 
     /**
-     * Parses a multi-line {@link RestRequest} body, instanciating a {@link SearchRequest} for each line and applying the given consumer.
+     * Parses a multi-line {@link RestRequest} body, instantiating a {@link SearchRequest} for each line and applying the given consumer.
      */
     public static void parseMultiLineRequest(RestRequest request, IndicesOptions indicesOptions, boolean allowExplicitIndex,
             BiConsumer<SearchRequest, XContentParser> consumer) throws IOException {
@@ -103,9 +104,10 @@ public class RestMultiSearchAction extends BaseRestHandler {
         String searchType = request.param("search_type");
         String routing = request.param("routing");
 
-        final BytesReference data = request.contentOrSourceParam();
+        final Tuple<XContentType, BytesReference> sourceTuple = request.contentOrSourceParam();
+        final XContent xContent = sourceTuple.v1().xContent();
+        final BytesReference data = sourceTuple.v2();
 
-        XContent xContent = XContentFactory.xContent(data);
         int from = 0;
         int length = data.length();
         byte marker = xContent.streamSeparator();
@@ -176,7 +178,7 @@ public class RestMultiSearchAction extends BaseRestHandler {
                 break;
             }
             BytesReference bytes = data.slice(from, nextMarker - from);
-            try (XContentParser parser = XContentFactory.xContent(bytes).createParser(request.getXContentRegistry(), bytes)) {
+            try (XContentParser parser = xContent.createParser(request.getXContentRegistry(), bytes)) {
                 consumer.accept(searchRequest, parser);
             }
             // move pointers
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 87a6f464cc..2a60fc6317 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
@@ -24,7 +24,6 @@ import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
-import org.elasticsearch.common.xcontent.XContentFactory;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.RestController;
@@ -58,25 +57,33 @@ public class RestSearchScrollAction extends BaseRestHandler {
             searchScrollRequest.scroll(new Scroll(parseTimeValue(scroll, null, "scroll")));
         }
 
-        BytesReference body = request.contentOrSourceParam();
-        if (body.length() > 0) {
-            if (XContentFactory.xContentType(body) == null) {
-                if (scrollId == null) {
-                    scrollId = body.utf8ToString();
-                    searchScrollRequest.scrollId(scrollId);
+        request.withContentOrSourceParamParserOrNull(xContentParser -> {
+            if (xContentParser == null) {
+                if (request.hasContent()) {
+                    // TODO: why do we accept this plain text value? maybe we can just use the scroll params?
+                    BytesReference body = request.getContentOrSourceParamOnly();
+                    if (scrollId == null) {
+                        String bodyScrollId = body.utf8ToString();
+                        searchScrollRequest.scrollId(bodyScrollId);
+                    }
                 }
             } else {
                 // NOTE: if rest request with xcontent body has request parameters, these parameters override xcontent values
-                try (XContentParser parser = request.contentOrSourceParamParser()) {
-                    buildFromContent(parser, searchScrollRequest);
+                try {
+                    buildFromContent(xContentParser, searchScrollRequest);
                 } catch (IOException e) {
                     throw new IllegalArgumentException("Failed to parse request body", e);
                 }
             }
-        }
+        });
         return channel -> client.searchScroll(searchScrollRequest, new RestStatusToXContentListener<>(channel));
     }
 
+    @Override
+    public boolean supportsPlainText() {
+        return true;
+    }
+
     public static void buildFromContent(XContentParser parser, SearchScrollRequest searchScrollRequest) throws IOException {
         if (parser.nextToken() != XContentParser.Token.START_OBJECT) {
             throw new IllegalArgumentException("Malformed content, must start with an object");