From 1c3466f3065c66c53eb3085b8b36875d0fe709a5 Mon Sep 17 00:00:00 2001
From: Tim Brooks <tim@timbrooks.org>
Date: Mon, 4 Nov 2024 11:09:36 -0700
Subject: [PATCH] Changes

---
 .../action/bulk/BulkRequestParser.java        | 14 ++++++++
 .../action/bulk/BulkRequestParserTests.java   | 32 +++++++++++++++++++
 2 files changed, 46 insertions(+)

diff --git a/server/src/main/java/org/elasticsearch/action/bulk/BulkRequestParser.java b/server/src/main/java/org/elasticsearch/action/bulk/BulkRequestParser.java
index 42c11586731b5..49c25abf1fae6 100644
--- 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
index 6fc71f0d7e056..5785d076693e7 100644
--- 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<>();