Changes

Tim-Brooks · Nov 4, 2024 · 1c3466f · 1c3466f
1 parent 905532d
commit 1c3466f
Show file tree

Hide file tree

Showing 2 changed files with 46 additions and 0 deletions.
diff --git a/server/src/main/java/org/elasticsearch/action/bulk/BulkRequestParser.java b/server/src/main/java/org/elasticsearch/action/bulk/BulkRequestParser.java
@@ -210,6 +210,7 @@ public class IncrementalParser {
         private final Consumer<UpdateRequest> updateRequestConsumer;
         private final Consumer<DeleteRequest> deleteRequestConsumer;
 
+        private Exception failure = null;
         private int incrementalFromOffset = 0;
         private int line = 0;
 
@@ -249,6 +250,19 @@ private IncrementalParser(
         }
 
         public int parse(BytesReference data, boolean lastData) throws IOException {
+            if (failure != null) {
+                assert false : failure.getMessage();
+                throw new IllegalStateException("Parser has already encountered exception", failure);
+            }
+            try {
+                return tryParse(data, lastData);
+            } catch (Exception e) {
+                failure = e;
+                throw e;
+            }
+        }
+
+        private int tryParse(BytesReference data, boolean lastData) throws IOException {
             int from = 0;
             int consumed = 0;
 

diff --git a/server/src/test/java/org/elasticsearch/action/bulk/BulkRequestParserTests.java b/server/src/test/java/org/elasticsearch/action/bulk/BulkRequestParserTests.java
@@ -33,6 +33,38 @@ public class BulkRequestParserTests extends ESTestCase {
         .filter(v -> v.compareTo(RestApiVersion.V_8) > 0)
         .toList();
 
+    public void testParserCannotBeReusedAfterFailure() {
+        BytesArray request = new BytesArray("""
+            { "index":{ }, "something": "unexpected" }
+            {}
+            """);
+
+        BulkRequestParser parser = new BulkRequestParser(randomBoolean(), RestApiVersion.current());
+        BulkRequestParser.IncrementalParser incrementalParser = parser.incrementalParser(
+            null,
+            null,
+            null,
+            null,
+            null,
+            null,
+            null,
+            false,
+            XContentType.JSON,
+            (req, type) -> fail("expected failure before we got this far"),
+            req -> fail("expected failure before we got this far"),
+            req -> fail("expected failure before we got this far")
+        );
+
+        IllegalArgumentException ex = expectThrows(IllegalArgumentException.class, () -> incrementalParser.parse(request, false));
+        assertEquals("Malformed action/metadata line [1], expected END_OBJECT but found [FIELD_NAME]", ex.getMessage());
+
+        BytesArray valid = new BytesArray("""
+            { "index":{ "_id": "bar" } }
+            {}
+            """);
+        expectThrows(AssertionError.class, () -> incrementalParser.parse(valid, false));
+    }
+
     public void testIncrementalParsing() throws IOException {
         ArrayList<DocWriteRequest<?>> indexRequests = new ArrayList<>();
         ArrayList<DocWriteRequest<?>> updateRequests = new ArrayList<>();