Add query param include_source_on_error for ingest requests

docs/changelog/120725.yaml

@@ -0,0 +1,7 @@
+pr: 120725
+summary: |-
+  A new query parameter `?include_source_on_error` was added for create / index, update and bulk REST APIs to control
+  if to include the document source in the error response in case of parsing errors. The default value is `true`.
+area: Infra/REST API
+type: enhancement
+issues: []

libs/x-content/impl/src/main/java/org/elasticsearch/xcontent/provider/XContentParserConfigurationImpl.java

@@ -31,7 +31,8 @@ public class XContentParserConfigurationImpl implements XContentParserConfigurat
         RestApiVersion.current(),
         null,
         null,
-        false
+        false,
+        true
     );
 
     final NamedXContentRegistry registry;
@@ -40,21 +41,45 @@ public class XContentParserConfigurationImpl implements XContentParserConfigurat
     final FilterPath[] includes;
     final FilterPath[] excludes;
     final boolean filtersMatchFieldNamesWithDots;
+    final boolean includeSourceOnError;
 
     private XContentParserConfigurationImpl(
         NamedXContentRegistry registry,
         DeprecationHandler deprecationHandler,
         RestApiVersion restApiVersion,
         FilterPath[] includes,
         FilterPath[] excludes,
-        boolean filtersMatchFieldNamesWithDots
+        boolean filtersMatchFieldNamesWithDots,
+        boolean includeSourceOnError
     ) {
         this.registry = registry;
         this.deprecationHandler = deprecationHandler;
         this.restApiVersion = restApiVersion;
         this.includes = includes;
         this.excludes = excludes;
         this.filtersMatchFieldNamesWithDots = filtersMatchFieldNamesWithDots;
+        this.includeSourceOnError = includeSourceOnError;
+    }
+
+    @Override
+    public boolean includeSourceOnError() {
+        return includeSourceOnError;
+    }
+
+    @Override
+    public XContentParserConfiguration withIncludeSourceOnError(boolean includeSourceOnError) {
+        if (includeSourceOnError == this.includeSourceOnError) {
+            return this;
+        }
+        return new XContentParserConfigurationImpl(
+            registry,
+            deprecationHandler,
+            restApiVersion,
+            includes,
+            excludes,
+            filtersMatchFieldNamesWithDots,
+            includeSourceOnError
+        );
     }
 
     @Override
@@ -65,7 +90,8 @@ public XContentParserConfigurationImpl withRegistry(NamedXContentRegistry regist
             restApiVersion,
             includes,
             excludes,
-            filtersMatchFieldNamesWithDots
+            filtersMatchFieldNamesWithDots,
+            includeSourceOnError
         );
     }
 
@@ -80,7 +106,8 @@ public XContentParserConfiguration withDeprecationHandler(DeprecationHandler dep
             restApiVersion,
             includes,
             excludes,
-            filtersMatchFieldNamesWithDots
+            filtersMatchFieldNamesWithDots,
+            includeSourceOnError
         );
     }
 
@@ -95,7 +122,8 @@ public XContentParserConfiguration withRestApiVersion(RestApiVersion restApiVers
             restApiVersion,
             includes,
             excludes,
-            filtersMatchFieldNamesWithDots
+            filtersMatchFieldNamesWithDots,
+            includeSourceOnError
         );
     }
 
@@ -143,7 +171,8 @@ public XContentParserConfiguration withFiltering(
             restApiVersion,
             includePaths,
             excludePaths,
-            filtersMatchFieldNamesWithDots
+            filtersMatchFieldNamesWithDots,
+            includeSourceOnError
         );
     }
 

libs/x-content/impl/src/main/java/org/elasticsearch/xcontent/provider/json/JsonXContentImpl.java

@@ -87,23 +87,30 @@ public XContentGenerator createGenerator(OutputStream os, Set<String> includes,
         return new JsonXContentGenerator(jsonFactory.createGenerator(os, JsonEncoding.UTF8), os, includes, excludes);
     }
 
+    private XContentParser createParser(XContentParserConfiguration config, JsonParser parser) {
+        if (config.includeSourceOnError() == false) {
+            parser.disable(JsonParser.Feature.INCLUDE_SOURCE_IN_LOCATION); // enabled by default, disable if requested
+        }
+        return new JsonXContentParser(config, parser);
+    }
+
     @Override
     public XContentParser createParser(XContentParserConfiguration config, String content) throws IOException {
-        return new JsonXContentParser(config, jsonFactory.createParser(content));
+        return createParser(config, jsonFactory.createParser(content));
     }
 
     @Override
     public XContentParser createParser(XContentParserConfiguration config, InputStream is) throws IOException {
-        return new JsonXContentParser(config, jsonFactory.createParser(is));
+        return createParser(config, jsonFactory.createParser(is));
     }
 
     @Override
     public XContentParser createParser(XContentParserConfiguration config, byte[] data, int offset, int length) throws IOException {
-        return new JsonXContentParser(config, jsonFactory.createParser(data, offset, length));
+        return createParser(config, jsonFactory.createParser(data, offset, length));
     }
 
     @Override
     public XContentParser createParser(XContentParserConfiguration config, Reader reader) throws IOException {
-        return new JsonXContentParser(config, jsonFactory.createParser(reader));
+        return createParser(config, jsonFactory.createParser(reader));
     }
 }

libs/x-content/src/main/java/org/elasticsearch/xcontent/XContentParserConfiguration.java

@@ -26,6 +26,13 @@ public interface XContentParserConfiguration {
      */
     XContentParserConfiguration EMPTY = XContentProvider.provider().empty();
 
+    /**
+     *  Disable to not include the source in case of parsing errors (defaults to true).
+     */
+    XContentParserConfiguration withIncludeSourceOnError(boolean includeSourceOnError);
+
+    boolean includeSourceOnError();
+
     /**
      * Replace the registry backing {@link XContentParser#namedObject}.
      */

libs/x-content/src/test/java/org/elasticsearch/xcontent/XContentParserTests.java

@@ -16,6 +16,7 @@
 import org.elasticsearch.xcontent.json.JsonXContent;
 
 import java.io.IOException;
+import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
@@ -33,6 +34,7 @@
 import static org.hamcrest.Matchers.in;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.not;
 import static org.hamcrest.Matchers.nullValue;
 import static org.junit.internal.matchers.ThrowableMessageMatcher.hasMessage;
 
@@ -655,6 +657,25 @@ public void testCreateRootSubParser() throws IOException {
 
     }
 
+    public void testJsonIncludeSourceOnParserError() throws IOException {
+        var xContent = XContentFactory.xContent(XContentType.JSON);
+        var source = "{\"field\": invalid}"; // causes parse exception
+        var sourceEnabled = XContentParserConfiguration.EMPTY;
+        var sourceDisabled = XContentParserConfiguration.EMPTY.withIncludeSourceOnError(false);
+
+        var parseException = expectThrows(XContentParseException.class, () -> createParser(xContent, sourceEnabled, source).map());
+        assertThat(parseException.getMessage(), containsString(source));
+
+        parseException = expectThrows(XContentParseException.class, () -> createParser(xContent, sourceDisabled, source).map());
+        assertThat(parseException.getMessage(), not(containsString(source)));
+    }
+
+    private XContentParser createParser(XContent xContent, XContentParserConfiguration config, String content) throws IOException {
+        return randomBoolean()
+            ? xContent.createParser(config, content)
+            : xContent.createParser(config, content.getBytes(StandardCharsets.UTF_8));
+    }
+
     /**
      * Generates a random object {"first_field": "foo", "marked_field": {...random...}, "last_field": "bar}
      *

rest-api-spec/src/main/resources/rest-api-spec/api/bulk.json

@@ -83,6 +83,10 @@
       "list_executed_pipelines": {
         "type": "boolean",
         "description": "Sets list_executed_pipelines for all incoming documents. Defaults to unset (false)"
+      },
+      "include_source_on_error": {
+        "type": "boolean",
+        "description": "True or false if to include the document source in the error message in case of parsing errors. Defaults to true."
       }
     },
     "body":{

rest-api-spec/src/main/resources/rest-api-spec/api/create.json

@@ -69,6 +69,10 @@
       "pipeline":{
         "type":"string",
         "description":"The pipeline id to preprocess incoming documents with"
+      },
+      "include_source_on_error": {
+        "type": "boolean",
+        "description": "True or false if to include the document source in the error message in case of parsing errors. Defaults to true."
       }
     },
     "body":{

rest-api-spec/src/main/resources/rest-api-spec/api/index.json

@@ -105,6 +105,10 @@
       "require_data_stream": {
         "type": "boolean",
         "description": "When true, requires the destination to be a data stream (existing or to-be-created). Default is false"
+      },
+      "include_source_on_error": {
+        "type": "boolean",
+        "description": "True or false if to include the document source in the error message in case of parsing errors. Defaults to true."
       }
     },
     "body":{

rest-api-spec/src/main/resources/rest-api-spec/api/update.json

@@ -83,6 +83,10 @@
       "require_alias": {
         "type": "boolean",
         "description": "When true, requires destination is an alias. Default is false"
+      },
+      "include_source_on_error": {
+        "type": "boolean",
+        "description": "True or false if to include the document source in the error message in case of parsing errors. Defaults to true."
       }
     },
     "body":{

server/src/internalClusterTest/java/org/elasticsearch/rest/action/document/RestBulkActionIT.java

@@ -0,0 +1,76 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.rest.action.document;
+
+import org.elasticsearch.client.Request;
+import org.elasticsearch.client.Response;
+import org.elasticsearch.client.ResponseException;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.io.Streams;
+import org.elasticsearch.rest.RestUtils;
+import org.elasticsearch.test.ESIntegTestCase;
+
+import java.io.InputStreamReader;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.hamcrest.Matchers.both;
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.not;
+
+public class RestBulkActionIT extends ESIntegTestCase {
+    @Override
+    protected boolean addMockHttpTransport() {
+        return false;
+    }
+
+    public void testBulkIndexWithSourceOnErrorDisabled() throws Exception {
+        var source = "{\"field\": \"index\",}";
+        var sourceEscaped = "{\\\"field\\\": \\\"index\\\",}";
+
+        var request = new Request("PUT", "/test_index/_bulk");
+        request.setJsonEntity(Strings.format("{\"index\":{\"_id\":\"1\"}}\n%s\n", source));
+
+        Response response = getRestClient().performRequest(request);
+        String responseContent = Streams.copyToString(new InputStreamReader(response.getEntity().getContent(), UTF_8));
+        assertThat(responseContent, containsString(sourceEscaped));
+
+        request.addParameter(RestUtils.INCLUDE_SOURCE_ON_ERROR_PARAMETER, "false");
+
+        response = getRestClient().performRequest(request);
+        responseContent = Streams.copyToString(new InputStreamReader(response.getEntity().getContent(), UTF_8));
+        assertThat(
+            responseContent,
+            both(not(containsString(sourceEscaped))).and(
+                containsString("REDACTED (`StreamReadFeature.INCLUDE_SOURCE_IN_LOCATION` disabled)")
+            )
+        );
+    }
+
+    public void testBulkUpdateWithSourceOnErrorDisabled() throws Exception {
+        var source = "{\"field\": \"index\",}";
+        var sourceEscaped = "{\\\"field\\\": \\\"index\\\",}";
+
+        var request = new Request("PUT", "/test_index/_bulk");
+        request.addParameter(RestUtils.INCLUDE_SOURCE_ON_ERROR_PARAMETER, "false");
+        request.setJsonEntity(Strings.format("{\"update\":{\"_id\":\"1\"}}\n{\"doc\":%s}}\n", source));
+
+        // note: this behavior is not consistent with bulk index actions
+        // In case of updates by doc, the source is eagerly parsed and will fail the entire request if it cannot be parsed
+        var exception = assertThrows(ResponseException.class, () -> getRestClient().performRequest(request));
+        String response = Streams.copyToString(new InputStreamReader(exception.getResponse().getEntity().getContent(), UTF_8));
+
+        assertThat(
+            response,
+            both(not(containsString(sourceEscaped))).and(
+                containsString("REDACTED (`StreamReadFeature.INCLUDE_SOURCE_IN_LOCATION` disabled)")
+            )
+        );
+    }
+}

server/src/internalClusterTest/java/org/elasticsearch/rest/action/document/RestIndexActionIT.java

@@ -0,0 +1,53 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.rest.action.document;
+
+import org.elasticsearch.client.Request;
+import org.elasticsearch.client.ResponseException;
+import org.elasticsearch.common.io.Streams;
+import org.elasticsearch.rest.RestUtils;
+import org.elasticsearch.test.ESIntegTestCase;
+
+import java.io.InputStreamReader;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.hamcrest.Matchers.both;
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.not;
+
+public class RestIndexActionIT extends ESIntegTestCase {
+    @Override
+    protected boolean addMockHttpTransport() {
+        return false;
+    }
+
+    public void testIndexWithSourceOnErrorDisabled() throws Exception {
+        var source = "{\"field\": \"value}";
+        var sourceEscaped = "{\\\"field\\\": \\\"value}";
+
+        var request = new Request("POST", "/test_index/_doc/1");
+        request.setJsonEntity(source);
+
+        var exception = assertThrows(ResponseException.class, () -> getRestClient().performRequest(request));
+        String response = Streams.copyToString(new InputStreamReader(exception.getResponse().getEntity().getContent(), UTF_8));
+        assertThat(response, containsString(sourceEscaped));
+
+        // disable source on error
+        request.addParameter(RestUtils.INCLUDE_SOURCE_ON_ERROR_PARAMETER, "false");
+        exception = assertThrows(ResponseException.class, () -> getRestClient().performRequest(request));
+        response = Streams.copyToString(new InputStreamReader(exception.getResponse().getEntity().getContent(), UTF_8));
+        assertThat(
+            response,
+            both(not(containsString(sourceEscaped))).and(
+                containsString("REDACTED (`StreamReadFeature.INCLUDE_SOURCE_IN_LOCATION` disabled)")
+            )
+        );
+    }
+}