Fixed NPEs caused by requests without content.

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

@@ -73,8 +73,8 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
         }
         bulkRequest.timeout(request.paramAsTime("timeout", BulkShardRequest.DEFAULT_TIMEOUT));
         bulkRequest.setRefreshPolicy(request.param("refresh"));
-        bulkRequest.add(request.content(), defaultIndex, defaultType, defaultRouting, defaultFields, null, defaultPipeline, null, true,
-            request.getXContentType());
+        bulkRequest.add(request.requiredContent(), defaultIndex, defaultType, defaultRouting, defaultFields,
+            null, defaultPipeline, null, true, request.getXContentType());
 
         // short circuit the call to the transport layer
         return channel -> {

core/src/main/java/org/elasticsearch/action/bulk/BulkRequest.java

@@ -41,7 +41,6 @@
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 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.VersionType;
@@ -300,10 +299,16 @@ public BulkRequest add(BytesReference data, @Nullable String defaultIndex, @Null
                 if (token == null) {
                     continue;
                 }
-                assert token == XContentParser.Token.START_OBJECT;
+                if (token != XContentParser.Token.START_OBJECT) {
+                    throw new IllegalArgumentException("Malformed action/metadata line [" + line + "], expected "
+                        + XContentParser.Token.START_OBJECT + " but found [" + token + "]");
+                }
                 // Move to FIELD_NAME, that's the action
                 token = parser.nextToken();
-                assert token == XContentParser.Token.FIELD_NAME;
+                if (token != XContentParser.Token.FIELD_NAME) {
+                    throw new IllegalArgumentException("Malformed action/metadata line [" + line + "], expected "
+                        + XContentParser.Token.FIELD_NAME + " but found [" + token + "]");
+                }
                 String action = parser.currentName();
 
                 String index = defaultIndex;

core/src/main/java/org/elasticsearch/rest/RestRequest.java

@@ -136,6 +136,18 @@ public final String path() {
 
     public abstract BytesReference content();
 
+    /**
+     * @return content of the request body or throw an exception if the body or content type is missing
+     */
+    public final BytesReference requiredContent() {
+        if (hasContent() == false) {
+            throw new ElasticsearchParseException("request body is required");
+        } else if (xContentType.get() == null) {
+            throw new IllegalStateException("unknown content type");
+        }
+        return content();
+    }
+
     /**
      * 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
@@ -329,12 +341,7 @@ public NamedXContentRegistry getXContentRegistry() {
      * {@link #contentOrSourceParamParser()} for requests that support specifying the request body in the {@code source} param.
      */
     public final XContentParser contentParser() throws IOException {
-        BytesReference content = content();
-        if (content.length() == 0) {
-            throw new ElasticsearchParseException("Body required");
-        } else if (xContentType.get() == null) {
-            throw new IllegalStateException("unknown content type");
-        }
+        BytesReference content = requiredContent(); // will throw exception if body or content type missing
         return xContentType.get().xContent().createParser(xContentRegistry, content);
     }
 
@@ -364,11 +371,7 @@ public final boolean hasContentOrSourceParam() {
      */
     public final XContentParser contentOrSourceParamParser() throws IOException {
         Tuple<XContentType, BytesReference> tuple = contentOrSourceParam();
-        BytesReference content = tuple.v2();
-        if (content.length() == 0) {
-            throw new ElasticsearchParseException("Body required");
-        }
-        return tuple.v1().xContent().createParser(xContentRegistry, content);
+        return tuple.v1().xContent().createParser(xContentRegistry, tuple.v2());
     }
 
     /**
@@ -377,10 +380,10 @@ public final XContentParser contentOrSourceParamParser() throws IOException {
      * back to the user when there isn't request content.
      */
     public final void withContentOrSourceParamParserOrNull(CheckedConsumer<XContentParser, IOException> withParser) throws IOException {
-        Tuple<XContentType, BytesReference> tuple = contentOrSourceParam();
-        BytesReference content = tuple.v2();
-        XContentType xContentType = tuple.v1();
-        if (content.length() > 0) {
+        if (hasContentOrSourceParam()) {
+            Tuple<XContentType, BytesReference> tuple = contentOrSourceParam();
+            BytesReference content = tuple.v2();
+            XContentType xContentType = tuple.v1();
             try (XContentParser parser = xContentType.xContent().createParser(xContentRegistry, content)) {
                 withParser.accept(parser);
             }
@@ -390,28 +393,26 @@ public final void withContentOrSourceParamParserOrNull(CheckedConsumer<XContentP
     }
 
     /**
-     * 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.
+     * Get the content of the request or the contents of the {@code source} param or throw an exception if both are missing.
+     * Prefer {@link #contentOrSourceParamParser()} or {@link #withContentOrSourceParamParserOrNull(CheckedConsumer)} if you need a parser.
      */
     public final Tuple<XContentType, BytesReference> contentOrSourceParam() {
-        if (hasContent()) {
-            if (xContentType.get() == null) {
-                throw new IllegalStateException("unknown content type");
-            }
-            return new Tuple<>(xContentType.get(), content());
+        if (hasContentOrSourceParam() == false) {
+            throw new ElasticsearchParseException("request body or source parameter is required");
+        } else if (hasContent()) {
+            return new Tuple<>(xContentType.get(), requiredContent());
         }
-
         String source = param("source");
         String typeParam = param("source_content_type");
-        if (source != null && typeParam != null) {
-            BytesArray bytes = new BytesArray(source);
-            final XContentType xContentType = parseContentType(Collections.singletonList(typeParam));
-            if (xContentType == null) {
-                throw new IllegalStateException("Unknown value for source_content_type [" + typeParam + "]");
-            }
-            return new Tuple<>(xContentType, bytes);
+        if (source == null || typeParam == null) {
+            throw new IllegalStateException("source and source_content_type parameters are required");
+        }
+        BytesArray bytes = new BytesArray(source);
+        final XContentType xContentType = parseContentType(Collections.singletonList(typeParam));
+        if (xContentType == null) {
+            throw new IllegalStateException("Unknown value for source_content_type [" + typeParam + "]");
         }
-        return new Tuple<>(XContentType.JSON, BytesArray.EMPTY);
+        return new Tuple<>(xContentType, bytes);
     }
 
     /**

core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestPutStoredScriptAction.java

@@ -58,7 +58,7 @@ public RestChannelConsumer prepareRequest(RestRequest request, NodeClient client
             lang = null;
         }
 
-        BytesReference content = request.content();
+        BytesReference content = request.requiredContent();
 
         if (lang != null) {
             deprecationLogger.deprecated(

core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestPutIndexTemplateAction.java

@@ -57,7 +57,7 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
         putRequest.masterNodeTimeout(request.paramAsTime("master_timeout", putRequest.masterNodeTimeout()));
         putRequest.create(request.paramAsBoolean("create", false));
         putRequest.cause(request.param("cause", ""));
-        putRequest.source(request.content(), request.getXContentType());
+        putRequest.source(request.requiredContent(), request.getXContentType());
         return channel -> client.admin().indices().putTemplate(putRequest, new AcknowledgedRestListener<>(channel));
     }
 

core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestPutMappingAction.java

@@ -64,7 +64,7 @@ public RestPutMappingAction(Settings settings, RestController controller) {
     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(), request.getXContentType());
+        putMappingRequest.source(request.requiredContent(), 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()));

core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestUpdateSettingsAction.java

@@ -54,18 +54,16 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
         updateSettingsRequest.indicesOptions(IndicesOptions.fromRequest(request, updateSettingsRequest.indicesOptions()));
 
         Map<String, Object> settings = new HashMap<>();
-        if (request.hasContent()) {
-            try (XContentParser parser = request.contentParser()) {
-                Map<String, Object> bodySettings = parser.map();
-                Object innerBodySettings = bodySettings.get("settings");
-                // clean up in case the body is wrapped with "settings" : { ... }
-                if (innerBodySettings instanceof Map) {
-                    @SuppressWarnings("unchecked")
-                    Map<String, Object> innerBodySettingsMap = (Map<String, Object>) innerBodySettings;
-                    settings.putAll(innerBodySettingsMap);
-                } else {
-                    settings.putAll(bodySettings);
-                }
+        try (XContentParser parser = request.contentParser()) {
+            Map<String, Object> bodySettings = parser.map();
+            Object innerBodySettings = bodySettings.get("settings");
+            // clean up in case the body is wrapped with "settings" : { ... }
+            if (innerBodySettings instanceof Map) {
+                @SuppressWarnings("unchecked")
+                Map<String, Object> innerBodySettingsMap = (Map<String, Object>) innerBodySettings;
+                settings.putAll(innerBodySettingsMap);
+            } else {
+                settings.putAll(bodySettings);
             }
         }
         updateSettingsRequest.settings(settings);

core/src/main/java/org/elasticsearch/rest/action/document/RestBulkAction.java

@@ -86,7 +86,7 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
         }
         bulkRequest.timeout(request.paramAsTime("timeout", BulkShardRequest.DEFAULT_TIMEOUT));
         bulkRequest.setRefreshPolicy(request.param("refresh"));
-        bulkRequest.add(request.content(), defaultIndex, defaultType, defaultRouting, defaultFields,
+        bulkRequest.add(request.requiredContent(), defaultIndex, defaultType, defaultRouting, defaultFields,
             defaultFetchSourceContext, defaultPipeline, null, allowExplicitIndex, request.getXContentType());
 
         return channel -> client.bulk(bulkRequest, new RestStatusToXContentListener<>(channel));

core/src/main/java/org/elasticsearch/rest/action/document/RestIndexAction.java

@@ -64,7 +64,7 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
         indexRequest.routing(request.param("routing"));
         indexRequest.parent(request.param("parent"));
         indexRequest.setPipeline(request.param("pipeline"));
-        indexRequest.source(request.content(), request.getXContentType());
+        indexRequest.source(request.requiredContent(), request.getXContentType());
         indexRequest.timeout(request.paramAsTime("timeout", IndexRequest.DEFAULT_TIMEOUT));
         indexRequest.setRefreshPolicy(request.param("refresh"));
         indexRequest.version(RestActions.parseVersion(request));

core/src/test/java/org/elasticsearch/action/bulk/BulkRequestTests.java

@@ -177,6 +177,31 @@ public void testSimpleBulk10() throws Exception {
         assertThat(bulkRequest.numberOfActions(), equalTo(9));
     }
 
+    public void testBulkEmptyObject() throws Exception {
+        String bulkIndexAction = "{ \"index\":{\"_index\":\"test\",\"_type\":\"type1\",\"_id\":\"1\"} }\r\n";
+        String bulkIndexSource = "{ \"field1\" : \"value1\" }\r\n";
+        String emptyObject = "{}\r\n";
+        StringBuilder bulk = new StringBuilder();
+        int emptyLine;
+        if (randomBoolean()) {
+            bulk.append(emptyObject);
+            emptyLine = 1;
+        } else {
+            int actions = randomIntBetween(1, 10);
+            int emptyAction = randomIntBetween(1, actions);
+            emptyLine = emptyAction * 2 - 1;
+            for (int i = 1; i <= actions; i++) {
+                bulk.append(i == emptyAction ? emptyObject : bulkIndexAction);
+                bulk.append(randomBoolean() ? emptyObject : bulkIndexSource);
+            }
+        }
+        String bulkAction = bulk.toString();
+        BulkRequest bulkRequest = new BulkRequest();
+        IllegalArgumentException exc = expectThrows(IllegalArgumentException.class,
+            () -> bulkRequest.add(bulkAction.getBytes(StandardCharsets.UTF_8), 0, bulkAction.length(), null, null, XContentType.JSON));
+        assertThat(exc.getMessage(), containsString("Malformed action/metadata line [" + emptyLine + "], expected FIELD_NAME but found [END_OBJECT]"));
+    }
+
     // issue 7361
     public void testBulkRequestWithRefresh() throws Exception {
         BulkRequest bulkRequest = new BulkRequest();

core/src/test/java/org/elasticsearch/rest/RestRequestTests.java

@@ -43,11 +43,14 @@ public class RestRequestTests extends ESTestCase {
     public void testContentParser() throws IOException {
         Exception e = expectThrows(ElasticsearchParseException.class, () ->
             new ContentRestRequest("", emptyMap()).contentParser());
-        assertEquals("Body required", e.getMessage());
+        assertEquals("request body is required", e.getMessage());
         e = expectThrows(ElasticsearchParseException.class, () ->
             new ContentRestRequest("", singletonMap("source", "{}")).contentParser());
-        assertEquals("Body required", e.getMessage());
+        assertEquals("request body is required", e.getMessage());
         assertEquals(emptyMap(), new ContentRestRequest("{}", emptyMap()).contentParser().map());
+        e = expectThrows(ElasticsearchParseException.class, () ->
+            new ContentRestRequest("", emptyMap(), emptyMap()).contentParser());
+        assertEquals("request body is required", e.getMessage());
     }
 
     public void testApplyContentParser() throws IOException {
@@ -59,7 +62,9 @@ public void testApplyContentParser() throws IOException {
     }
 
     public void testContentOrSourceParam() throws IOException {
-        assertEquals(BytesArray.EMPTY, new ContentRestRequest("", emptyMap()).contentOrSourceParam().v2());
+        Exception e = expectThrows(ElasticsearchParseException.class, () ->
+            new ContentRestRequest("", emptyMap()).contentOrSourceParam());
+        assertEquals("request body or source parameter is required", e.getMessage());
         assertEquals(new BytesArray("stuff"), new ContentRestRequest("stuff", emptyMap()).contentOrSourceParam().v2());
         assertEquals(new BytesArray("stuff"),
             new ContentRestRequest("stuff", MapBuilder.<String, String>newMapBuilder()
@@ -68,6 +73,10 @@ public void testContentOrSourceParam() throws IOException {
             new ContentRestRequest("", MapBuilder.<String, String>newMapBuilder()
                 .put("source", "{\"foo\": \"stuff\"}").put("source_content_type", "application/json").immutableMap())
                 .contentOrSourceParam().v2());
+        e = expectThrows(IllegalStateException.class, () ->
+            new ContentRestRequest("", MapBuilder.<String, String>newMapBuilder()
+                .put("source", "stuff2").immutableMap()).contentOrSourceParam());
+        assertEquals("source and source_content_type parameters are required", e.getMessage());
     }
 
     public void testHasContentOrSourceParam() throws IOException {
@@ -80,7 +89,7 @@ public void testHasContentOrSourceParam() throws IOException {
     public void testContentOrSourceParamParser() throws IOException {
         Exception e = expectThrows(ElasticsearchParseException.class, () ->
             new ContentRestRequest("", emptyMap()).contentOrSourceParamParser());
-        assertEquals("Body required", e.getMessage());
+        assertEquals("request body or source parameter is required", e.getMessage());
         assertEquals(emptyMap(), new ContentRestRequest("{}", emptyMap()).contentOrSourceParamParser().map());
         assertEquals(emptyMap(), new ContentRestRequest("{}", singletonMap("source", "stuff2")).contentOrSourceParamParser().map());
         assertEquals(emptyMap(), new ContentRestRequest("", MapBuilder.<String, String>newMapBuilder()
@@ -138,6 +147,24 @@ public void testMultipleContentTypeHeaders() {
         assertEquals("only one Content-Type header should be provided", e.getMessage());
     }
 
+    public void testRequiredContent() {
+        Exception e = expectThrows(ElasticsearchParseException.class, () ->
+            new ContentRestRequest("", emptyMap()).requiredContent());
+        assertEquals("request body is required", e.getMessage());
+        assertEquals(new BytesArray("stuff"), new ContentRestRequest("stuff", emptyMap()).requiredContent());
+        assertEquals(new BytesArray("stuff"),
+            new ContentRestRequest("stuff", MapBuilder.<String, String>newMapBuilder()
+                .put("source", "stuff2").put("source_content_type", "application/json").immutableMap()).requiredContent());
+        e = expectThrows(ElasticsearchParseException.class, () ->
+            new ContentRestRequest("", MapBuilder.<String, String>newMapBuilder()
+                .put("source", "{\"foo\": \"stuff\"}").put("source_content_type", "application/json").immutableMap())
+                .requiredContent());
+        assertEquals("request body is required", e.getMessage());
+        e = expectThrows(IllegalStateException.class, () ->
+            new ContentRestRequest("test", null, Collections.emptyMap()).requiredContent());
+        assertEquals("unknown content type", e.getMessage());
+    }
+
     private static final class ContentRestRequest extends RestRequest {
         private final BytesArray content;
 

modules/ingest-common/src/test/resources/rest-api-spec/test/ingest/20_crud.yml

@@ -203,3 +203,16 @@ teardown:
       catch: missing
       ingest.get_pipeline:
         id: "my_pipeline"
+
+---
+"missing body":
+
+  - skip:
+      version: " - 5.99.99"
+      reason: NPE caused by missing body fixed in 6.0.0
+
+  - do:
+      catch: /request body or source parameter is required/
+      raw:
+        method: PUT
+        path: _ingest/pipeline/my_pipeline

modules/ingest-common/src/test/resources/rest-api-spec/test/ingest/90_simulate.yml

@@ -605,3 +605,16 @@ teardown:
   - length: { docs.0.processor_results.1: 2 }
   - match: { docs.0.processor_results.1.tag: "rename-1" }
   - match: { docs.0.processor_results.1.doc._source.new_status: 200 }
+
+---
+"missing body":
+
+  - skip:
+      version: " - 5.99.99"
+      reason: NPE caused by missing body fixed in 6.0.0
+
+  - do:
+      catch: /request body or source parameter is required/
+      raw:
+        method: POST
+        path: _ingest/pipeline/my_pipeline/_simulate

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

@@ -57,10 +57,6 @@ public RestMultiSearchTemplateAction(Settings settings, RestController controlle
 
     @Override
     public RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
-        if (request.hasContentOrSourceParam() == false) {
-            throw new ElasticsearchException("request body is required");
-        }
-
         MultiSearchTemplateRequest multiRequest = parseRequest(request, allowExplicitIndex);
         return channel -> client.execute(MultiSearchTemplateAction.INSTANCE, multiRequest, new RestToXContentListener<>(channel));
     }