handle missing content in incremental bulk

Author: mhl-b

server/src/main/java/org/elasticsearch/http/HttpBody.java

@@ -27,6 +27,26 @@ static Full empty() {
         return new ByteRefHttpBody(ReleasableBytesReference.empty());
     }
 
+    class NoopStream implements Stream {
+
+        @Override
+        public ChunkHandler handler() {
+            return null;
+        }
+
+        @Override
+        public void addTracingHandler(ChunkHandler chunkHandler) {}
+
+        @Override
+        public void setHandler(ChunkHandler chunkHandler) {}
+
+        @Override
+        public void next() {}
+
+        @Override
+        public void close() {}
+    }
+
     default boolean isFull() {
         return this instanceof Full;
     }

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

@@ -329,14 +329,19 @@ public HttpBody.Stream contentStream() {
         return httpRequest.body().asStream();
     }
 
+    public void ensureContent() {
+        if (hasContent() == false) {
+            throw new ElasticsearchParseException("request body is required");
+        }
+    }
+
     /**
      * Returns reference to the network buffer of HTTP content or throw an exception if the body or content type is missing.
      * See {@link #content()}.
      */
     public ReleasableBytesReference requiredContent() {
-        if (hasContent() == false) {
-            throw new ElasticsearchParseException("request body is required");
-        } else if (xContentType.get() == null) {
+        ensureContent();
+        if (xContentType.get() == null) {
             throwValidationException("unknown content type");
         }
         return content();

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

@@ -137,6 +137,7 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
                 client.bulk(bulkRequest, ActionListener.releaseAfter(new RestRefCountedChunkedToXContentListener<>(channel), content));
             };
         } else {
+            request.ensureContent();
             String waitForActiveShards = request.param("wait_for_active_shards");
             TimeValue timeout = request.paramAsTime("timeout", BulkShardRequest.DEFAULT_TIMEOUT);
             String refresh = request.param("refresh");

server/src/test/java/org/elasticsearch/rest/action/document/RestBulkActionTests.java

@@ -10,6 +10,7 @@
 package org.elasticsearch.rest.action.document;
 
 import org.apache.lucene.util.SetOnce;
+import org.elasticsearch.ElasticsearchParseException;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.DocWriteRequest;
 import org.elasticsearch.action.bulk.BulkRequest;
@@ -18,6 +19,7 @@
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.update.UpdateRequest;
 import org.elasticsearch.client.internal.Client;
+import org.elasticsearch.client.internal.node.NodeClient;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.bytes.ReleasableBytesReference;
 import org.elasticsearch.common.settings.ClusterSettings;
@@ -201,28 +203,29 @@ public void bulk(BulkRequest request, ActionListener<BulkResponse> listener) {
         }
     }
 
+    public void testIncrementalBulkMissingContent() {
+        assertThrows(
+            ElasticsearchParseException.class,
+            () -> new RestBulkAction(
+                Settings.EMPTY,
+                ClusterSettings.createBuiltInClusterSettings(),
+                new IncrementalBulkService(mock(Client.class), mock(IndexingPressure.class), MeterRegistry.NOOP)
+            ).handleRequest(
+                new FakeRestRequest.Builder(xContentRegistry()).withPath("my_index/_bulk").withBody(new HttpBody.NoopStream()).build(),
+                mock(RestChannel.class),
+                mock(NodeClient.class)
+            )
+        );
+    }
+
     public void testIncrementalParsing() {
         ArrayList<DocWriteRequest<?>> docs = new ArrayList<>();
         AtomicBoolean isLast = new AtomicBoolean(false);
         AtomicBoolean next = new AtomicBoolean(false);
 
         FakeRestRequest request = new FakeRestRequest.Builder(xContentRegistry()).withPath("my_index/_bulk")
             .withMethod(RestRequest.Method.POST)
-            .withBody(new HttpBody.Stream() {
-                @Override
-                public void close() {}
-
-                @Override
-                public ChunkHandler handler() {
-                    return null;
-                }
-
-                @Override
-                public void addTracingHandler(ChunkHandler chunkHandler) {}
-
-                @Override
-                public void setHandler(ChunkHandler chunkHandler) {}
-
+            .withBody(new HttpBody.NoopStream() {
                 @Override
                 public void next() {
                     next.set(true);