Tests

server/src/test/java/org/elasticsearch/action/bulk/BulkRequestParserTests.java

@@ -9,6 +9,7 @@
 
 package org.elasticsearch.action.bulk;
 
+import org.elasticsearch.action.DocWriteRequest;
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.core.RestApiVersion;
@@ -21,8 +22,56 @@
 import java.util.List;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import static org.hamcrest.Matchers.equalTo;
+
 public class BulkRequestParserTests extends ESTestCase {
 
+    public void testIncrementalParsing() throws IOException {
+        ArrayList<DocWriteRequest<?>> indexRequests = new ArrayList<>();
+        ArrayList<DocWriteRequest<?>> updateRequests = new ArrayList<>();
+        ArrayList<DocWriteRequest<?>> deleteRequests = new ArrayList<>();
+
+        BulkRequestParser parser = new BulkRequestParser(randomBoolean(), RestApiVersion.current());
+        BulkRequestParser.IncrementalParser incrementalParser = parser.incrementalParser(
+            null,
+            null,
+            null,
+            null,
+            null,
+            null,
+            null,
+            false,
+            XContentType.JSON,
+            (r, t) -> indexRequests.add(r),
+            updateRequests::add,
+            deleteRequests::add
+        );
+
+        BytesArray request = new BytesArray("""
+            { "index":{ "_id": "bar", "pipeline": "foo" } }
+            { "field": "value"}
+            { "index":{ "require_alias": false } }
+            { "field": "value" }
+            { "update":{ "_id": "bus", "require_alias": true } }
+            { "doc": {"field": "value" }}
+            { "delete":{ "_id": "baz" } }
+            { "index": { } }
+            { "field": "value"}
+            { "delete":{ "_id": "bop" } }
+            """);
+
+        int consumed = 0;
+        for (int i = 0; i < request.length() - 1; ++i) {
+            consumed += incrementalParser.parse(request.slice(consumed, i - consumed + 1), false);
+        }
+        consumed += incrementalParser.parse(request.slice(consumed, request.length() - consumed), true);
+        assertThat(consumed, equalTo(request.length()));
+
+        assertThat(indexRequests.size(), equalTo(3));
+        assertThat(updateRequests.size(), equalTo(1));
+        assertThat(deleteRequests.size(), equalTo(2));
+    }
+
     public void testIndexRequest() throws IOException {
         BytesArray request = new BytesArray("""
             { "index":{ "_id": "bar" } }
@@ -126,7 +175,7 @@ public void testUpdateRequest() throws IOException {
         }, req -> fail());
     }
 
-    public void testBarfOnLackOfTrailingNewline() {
+    public void testBarfOnLackOfTrailingNewline() throws IOException {
         BytesArray request = new BytesArray("""
             { "index":{ "_id": "bar" } }
             {}""");
@@ -150,6 +199,27 @@ public void testBarfOnLackOfTrailingNewline() {
             )
         );
         assertEquals("The bulk request must be terminated by a newline [\\n]", e.getMessage());
+
+        BulkRequestParser.IncrementalParser incrementalParser = parser.incrementalParser(
+            "foo",
+            null,
+            null,
+            null,
+            null,
+            null,
+            null,
+            false,
+            XContentType.JSON,
+            (req, type) -> {},
+            req -> {},
+            req -> {}
+        );
+
+        // Should not throw because not last
+        incrementalParser.parse(request, false);
+
+        IllegalArgumentException e2 = expectThrows(IllegalArgumentException.class, () -> incrementalParser.parse(request, true));
+        assertEquals("The bulk request must be terminated by a newline [\\n]", e2.getMessage());
     }
 
     public void testFailOnExplicitIndex() {