[8.x] Added query param ?include_source_on_error for ingest requests (#120725) (#121010)

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`.

Relates to ES-9186.

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,6 +41,7 @@ public class XContentParserConfigurationImpl implements XContentParserConfigurat
     final FilterPath[] includes;
     final FilterPath[] excludes;
     final boolean filtersMatchFieldNamesWithDots;
+    final boolean includeSourceOnError;
 
     private XContentParserConfigurationImpl(
         NamedXContentRegistry registry,
@@ -47,7 +49,8 @@ public class XContentParserConfigurationImpl implements XContentParserConfigurat
         RestApiVersion restApiVersion,
         FilterPath[] includes,
         FilterPath[] excludes,
-        boolean filtersMatchFieldNamesWithDots
+        boolean filtersMatchFieldNamesWithDots,
+        boolean includeSourceOnError
     ) {
         this.registry = registry;
         this.deprecationHandler = deprecationHandler;
@@ -55,6 +58,28 @@ public class XContentParserConfigurationImpl implements XContentParserConfigurat
         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 class XContentParserConfigurationImpl implements XContentParserConfigurat
             restApiVersion,
             includes,
             excludes,
-            filtersMatchFieldNamesWithDots
+            filtersMatchFieldNamesWithDots,
+            includeSourceOnError
         );
     }
 
@@ -80,7 +106,8 @@ public class XContentParserConfigurationImpl implements XContentParserConfigurat
             restApiVersion,
             includes,
             excludes,
-            filtersMatchFieldNamesWithDots
+            filtersMatchFieldNamesWithDots,
+            includeSourceOnError
         );
     }
 
@@ -95,7 +122,8 @@ public class XContentParserConfigurationImpl implements XContentParserConfigurat
             restApiVersion,
             includes,
             excludes,
-            filtersMatchFieldNamesWithDots
+            filtersMatchFieldNamesWithDots,
+            includeSourceOnError
         );
     }
 
@@ -143,7 +171,8 @@ public class XContentParserConfigurationImpl implements XContentParserConfigurat
             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 class JsonXContentImpl implements XContent {
         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.test.ESTestCase;
 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.hasSize;
 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 class XContentParserTests extends ESTestCase {
 
     }
 
+    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

@@ -87,6 +87,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)")
+            )
+        );
+    }
+}

server/src/internalClusterTest/java/org/elasticsearch/rest/action/document/RestUpdateActionIT.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 RestUpdateActionIT extends ESIntegTestCase {
+    @Override
+    protected boolean addMockHttpTransport() {
+        return false;
+    }
+
+    public void testUpdateByDocWithSourceOnErrorDisabled() throws Exception {
+        var updateRequest = "{\"doc\":{\"field\": \"value}}";
+        var sourceEscaped = "{\\\"field\\\": \\\"value}";
+
+        var request = new Request("POST", "/test_index/_update/1");
+        request.setJsonEntity(updateRequest);
+
+        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)")
+            )
+        );
+    }
+}

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -172,6 +172,7 @@ public class TransportVersions {
     public static final TransportVersion ESQL_RESPONSE_PARTIAL = def(8_832_00_0);
     public static final TransportVersion RANK_DOC_OPTIONAL_METADATA_FOR_EXPLAIN = def(8_833_00_0);
     public static final TransportVersion ILM_ADD_SEARCHABLE_SNAPSHOT_ADD_REPLICATE_FOR = def(8_834_00_0);
+    public static final TransportVersion INGEST_REQUEST_INCLUDE_SOURCE_ON_ERROR = def(8_835_00_0);
 
     /*
      * STOP! READ THIS FIRST! No, really,

server/src/main/java/org/elasticsearch/action/bulk/BulkRequest.java

@@ -84,6 +84,7 @@ public class BulkRequest extends ActionRequest
     private String globalIndex;
     private Boolean globalRequireAlias;
     private Boolean globalRequireDatsStream;
+    private boolean includeSourceOnError = true;
 
     private long sizeInBytes = 0;
 
@@ -103,6 +104,9 @@ public class BulkRequest extends ActionRequest
         } else {
             incrementalState = BulkRequest.IncrementalState.EMPTY;
         }
+        if (in.getTransportVersion().onOrAfter(TransportVersions.INGEST_REQUEST_INCLUDE_SOURCE_ON_ERROR)) {
+            includeSourceOnError = in.readBoolean();
+        } // else default value is true
     }
 
     public BulkRequest(@Nullable String globalIndex) {
@@ -278,7 +282,7 @@ public class BulkRequest extends ActionRequest
         String pipeline = valueOrDefault(defaultPipeline, globalPipeline);
         Boolean requireAlias = valueOrDefault(defaultRequireAlias, globalRequireAlias);
         Boolean requireDataStream = valueOrDefault(defaultRequireDataStream, globalRequireDatsStream);
-        new BulkRequestParser(true, restApiVersion).parse(
+        new BulkRequestParser(true, includeSourceOnError, restApiVersion).parse(
             data,
             defaultIndex,
             routing,
@@ -341,6 +345,11 @@ public class BulkRequest extends ActionRequest
         this.incrementalState = incrementalState;
     }
 
+    public final BulkRequest includeSourceOnError(boolean includeSourceOnError) {
+        this.includeSourceOnError = includeSourceOnError;
+        return this;
+    }
+
     /**
      * Note for internal callers (NOT high level rest client),
      * the global parameter setting is ignored when used with:
@@ -399,6 +408,10 @@ public class BulkRequest extends ActionRequest
         return globalRequireDatsStream;
     }
 
+    public boolean includeSourceOnError() {
+        return includeSourceOnError;
+    }
+
     /**
      * Note for internal callers (NOT high level rest client),
      * the global parameter setting is ignored when used with:
@@ -457,6 +470,9 @@ public class BulkRequest extends ActionRequest
         if (out.getTransportVersion().onOrAfter(TransportVersions.V_8_16_0)) {
             incrementalState.writeTo(out);
         }
+        if (out.getTransportVersion().onOrAfter(TransportVersions.INGEST_REQUEST_INCLUDE_SOURCE_ON_ERROR)) {
+            out.writeBoolean(includeSourceOnError);
+        }
     }
 
     @Override

server/src/main/java/org/elasticsearch/action/bulk/BulkRequestParser.java

@@ -78,12 +78,14 @@ public final class BulkRequestParser {
      * Create a new parser.
      *
      * @param deprecateOrErrorOnType whether to allow _type information in the index line; used by BulkMonitoring
+     * @param includeSourceOnError if to include the source in parser error messages
      * @param restApiVersion
      */
-    public BulkRequestParser(boolean deprecateOrErrorOnType, RestApiVersion restApiVersion) {
+    public BulkRequestParser(boolean deprecateOrErrorOnType, boolean includeSourceOnError, RestApiVersion restApiVersion) {
         this.deprecateOrErrorOnType = deprecateOrErrorOnType;
         this.config = XContentParserConfiguration.EMPTY.withDeprecationHandler(LoggingDeprecationHandler.INSTANCE)
-            .withRestApiVersion(restApiVersion);
+            .withRestApiVersion(restApiVersion)
+            .withIncludeSourceOnError(includeSourceOnError);
     }
 
     private static int findNextMarker(byte marker, int from, BytesReference data, boolean lastData) {
@@ -485,7 +487,8 @@ public final class BulkRequestParser {
                             .setDynamicTemplates(dynamicTemplates)
                             .setRequireAlias(requireAlias)
                             .setRequireDataStream(requireDataStream)
-                            .setListExecutedPipelines(currentListExecutedPipelines);
+                            .setListExecutedPipelines(currentListExecutedPipelines)
+                            .setIncludeSourceOnError(config.includeSourceOnError());
                         if ("create".equals(action)) {
                             indexRequest = indexRequest.create(true);
                         } else if (opType != null) {

server/src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java

@@ -376,6 +376,7 @@ public class TransportShardBulkAction extends TransportWriteAction<BulkShardRequ
                 request.getContentType(),
                 request.routing(),
                 request.getDynamicTemplates(),
+                request.getIncludeSourceOnError(),
                 meteringParserDecorator
             );
             result = primary.applyIndexOperationOnPrimary(

server/src/main/java/org/elasticsearch/action/bulk/TransportSimulateBulkAction.java

@@ -196,6 +196,7 @@ public class TransportSimulateBulkAction extends TransportAbstractBulkAction {
             request.getContentType(),
             request.routing(),
             request.getDynamicTemplates(),
+            request.getIncludeSourceOnError(),
             XContentMeteringParserDecorator.NOOP
         );
 

server/src/main/java/org/elasticsearch/action/index/IndexRequest.java

@@ -116,6 +116,8 @@ public class IndexRequest extends ReplicatedWriteRequest<IndexRequest> implement
 
     private boolean requireDataStream;
 
+    private boolean includeSourceOnError = true;
+
     /**
      * Transient flag denoting that the local request should be routed to a failure store. Not persisted across the wire.
      */
@@ -210,6 +212,10 @@ public class IndexRequest extends ReplicatedWriteRequest<IndexRequest> implement
                 in.readBoolean(); // obsolete originatesFromUpdateByDoc
             }
         }
+
+        if (in.getTransportVersion().onOrAfter(TransportVersions.INGEST_REQUEST_INCLUDE_SOURCE_ON_ERROR)) {
+            includeSourceOnError = in.readBoolean();
+        } // else default value is true
     }
 
     public IndexRequest() {
@@ -806,6 +812,9 @@ public class IndexRequest extends ReplicatedWriteRequest<IndexRequest> implement
                 out.writeBoolean(false); // obsolete originatesFromUpdateByDoc
             }
         }
+        if (out.getTransportVersion().onOrAfter(TransportVersions.INGEST_REQUEST_INCLUDE_SOURCE_ON_ERROR)) {
+            out.writeBoolean(includeSourceOnError);
+        }
     }
 
     @Override
@@ -874,6 +883,15 @@ public class IndexRequest extends ReplicatedWriteRequest<IndexRequest> implement
         return this;
     }
 
+    public boolean getIncludeSourceOnError() {
+        return includeSourceOnError;
+    }
+
+    public IndexRequest setIncludeSourceOnError(boolean includeSourceOnError) {
+        this.includeSourceOnError = includeSourceOnError;
+        return this;
+    }
+
     @Override
     public Index getConcreteWriteIndex(IndexAbstraction ia, Metadata metadata) {
         if (DataStream.isFailureStoreFeatureFlagEnabled() && writeToFailureStore) {

server/src/main/java/org/elasticsearch/index/mapper/DocumentParser.java

@@ -85,7 +85,11 @@ public final class DocumentParser {
         XContentMeteringParserDecorator meteringParserDecorator = source.getDocumentSizeObserver();
         try (
             XContentParser parser = meteringParserDecorator.decorate(
-                XContentHelper.createParser(parserConfiguration, source.source(), xContentType)
+                XContentHelper.createParser(
+                    parserConfiguration.withIncludeSourceOnError(source.getIncludeSourceOnError()),
+                    source.source(),
+                    xContentType
+                )
             )
         ) {
             context = new RootDocumentParserContext(mappingLookup, mappingParserContext, source, parser);

server/src/main/java/org/elasticsearch/index/mapper/SourceToParse.java

@@ -29,6 +29,9 @@ public class SourceToParse {
     private final XContentType xContentType;
 
     private final Map<String, String> dynamicTemplates;
+
+    private final boolean includeSourceOnError;
+
     private final XContentMeteringParserDecorator meteringParserDecorator;
 
     public SourceToParse(
@@ -37,6 +40,7 @@ public class SourceToParse {
         XContentType xContentType,
         @Nullable String routing,
         Map<String, String> dynamicTemplates,
+        boolean includeSourceOnError,
         XContentMeteringParserDecorator meteringParserDecorator
     ) {
         this.id = id;
@@ -46,15 +50,26 @@ public class SourceToParse {
         this.xContentType = Objects.requireNonNull(xContentType);
         this.routing = routing;
         this.dynamicTemplates = Objects.requireNonNull(dynamicTemplates);
+        this.includeSourceOnError = includeSourceOnError;
         this.meteringParserDecorator = meteringParserDecorator;
     }
 
     public SourceToParse(String id, BytesReference source, XContentType xContentType) {
-        this(id, source, xContentType, null, Map.of(), XContentMeteringParserDecorator.NOOP);
+        this(id, source, xContentType, null, Map.of(), true, XContentMeteringParserDecorator.NOOP);
     }
 
     public SourceToParse(String id, BytesReference source, XContentType xContentType, String routing) {
-        this(id, source, xContentType, routing, Map.of(), XContentMeteringParserDecorator.NOOP);
+        this(id, source, xContentType, routing, Map.of(), true, XContentMeteringParserDecorator.NOOP);
+    }
+
+    public SourceToParse(
+        String id,
+        BytesReference source,
+        XContentType xContentType,
+        String routing,
+        Map<String, String> dynamicTemplates
+    ) {
+        this(id, source, xContentType, routing, dynamicTemplates, true, XContentMeteringParserDecorator.NOOP);
     }
 
     public BytesReference source() {
@@ -94,4 +109,8 @@ public class SourceToParse {
     public XContentMeteringParserDecorator getDocumentSizeObserver() {
         return meteringParserDecorator;
     }
+
+    public boolean getIncludeSourceOnError() {
+        return includeSourceOnError;
+    }
 }

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

@@ -543,9 +543,24 @@ public class RestRequest implements ToXContent.Params, Traceable {
      * {@link #contentOrSourceParamParser()} for requests that support specifying the request body in the {@code source} param.
      */
     public final XContentParser contentParser() throws IOException {
+        return contentParser(parserConfig);
+    }
+
+    private XContentParser contentParser(XContentParserConfiguration parserConfig) throws IOException {
         BytesReference content = requiredContent(); // will throw exception if body or content type missing
         return XContentHelper.createParserNotCompressed(parserConfig, content, xContentType.get());
+    }
 
+    /**
+     * If there is any content then call {@code applyParser} with the parser modified by {@code includeSourceOnError}, otherwise do nothing.
+     */
+    public final void applyContentParser(boolean includeSourceOnError, CheckedConsumer<XContentParser, IOException> applyParser)
+        throws IOException {
+        if (hasContent()) {
+            try (XContentParser parser = contentParser(parserConfig.withIncludeSourceOnError(includeSourceOnError))) {
+                applyParser.accept(parser);
+            }
+        }
     }
 
     /**
@@ -553,7 +568,7 @@ public class RestRequest implements ToXContent.Params, Traceable {
      */
     public final void applyContentParser(CheckedConsumer<XContentParser, IOException> applyParser) throws IOException {
         if (hasContent()) {
-            try (XContentParser parser = contentParser()) {
+            try (XContentParser parser = contentParser(parserConfig)) {
                 applyParser.accept(parser);
             }
         }

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

@@ -278,6 +278,12 @@ public class RestUtils {
      */
     public static final String REST_TIMEOUT_PARAM = "timeout";
 
+    /**
+     * The name of the common {@code ?include_source_on_error} query parameter.
+     * By default, the document source is included in the error response in case of parsing errors. This parameter allows to disable this.
+     */
+    public static final String INCLUDE_SOURCE_ON_ERROR_PARAMETER = "include_source_on_error";
+
     /**
      * Extract the {@code ?master_timeout} parameter from the request, imposing the common default of {@code 30s} in case the parameter is
      * missing.
@@ -315,4 +321,15 @@ public class RestUtils {
         assert restRequest != null;
         return restRequest.paramAsTime(REST_TIMEOUT_PARAM, null);
     }
+
+    /**
+     * Extract the {@code ?include_source_on_error} parameter from the request, returning {@code true} in case the parameter is missing.
+     *
+     * @param restRequest The request from which to extract the {@code ?include_source_on_error} parameter
+     * @return the value of the {@code ?include_source_on_error} parameter from the request, with a default of {@code true} if the request
+     */
+    public static boolean getIncludeSourceOnError(RestRequest restRequest) {
+        assert restRequest != null;
+        return restRequest.paramAsBoolean(INCLUDE_SOURCE_ON_ERROR_PARAMETER, true);
+    }
 }

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

@@ -31,6 +31,7 @@ import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestRequest;
+import org.elasticsearch.rest.RestUtils;
 import org.elasticsearch.rest.Scope;
 import org.elasticsearch.rest.ServerlessScope;
 import org.elasticsearch.rest.action.RestRefCountedChunkedToXContentListener;
@@ -114,6 +115,7 @@ public class RestBulkAction extends BaseRestHandler {
             boolean defaultRequireDataStream = request.paramAsBoolean(DocWriteRequest.REQUIRE_DATA_STREAM, false);
             bulkRequest.timeout(request.paramAsTime("timeout", BulkShardRequest.DEFAULT_TIMEOUT));
             bulkRequest.setRefreshPolicy(request.param("refresh"));
+            bulkRequest.includeSourceOnError(RestUtils.getIncludeSourceOnError(request));
             ReleasableBytesReference content = request.requiredContent();
 
             try {
@@ -175,20 +177,21 @@ public class RestBulkAction extends BaseRestHandler {
         ChunkHandler(boolean allowExplicitIndex, RestRequest request, Supplier<IncrementalBulkService.Handler> handlerSupplier) {
             this.request = request;
             this.handlerSupplier = handlerSupplier;
-            this.parser = new BulkRequestParser(true, request.getRestApiVersion()).incrementalParser(
-                request.param("index"),
-                request.param("routing"),
-                FetchSourceContext.parseFromRestRequest(request),
-                request.param("pipeline"),
-                request.paramAsBoolean(DocWriteRequest.REQUIRE_ALIAS, false),
-                request.paramAsBoolean(DocWriteRequest.REQUIRE_DATA_STREAM, false),
-                request.paramAsBoolean("list_executed_pipelines", false),
-                allowExplicitIndex,
-                request.getXContentType(),
-                (indexRequest, type) -> items.add(indexRequest),
-                items::add,
-                items::add
-            );
+            this.parser = new BulkRequestParser(true, RestUtils.getIncludeSourceOnError(request), request.getRestApiVersion())
+                .incrementalParser(
+                    request.param("index"),
+                    request.param("routing"),
+                    FetchSourceContext.parseFromRestRequest(request),
+                    request.param("pipeline"),
+                    request.paramAsBoolean(DocWriteRequest.REQUIRE_ALIAS, false),
+                    request.paramAsBoolean(DocWriteRequest.REQUIRE_DATA_STREAM, false),
+                    request.paramAsBoolean("list_executed_pipelines", false),
+                    allowExplicitIndex,
+                    request.getXContentType(),
+                    (indexRequest, type) -> items.add(indexRequest),
+                    items::add,
+                    items::add
+                );
         }
 
         @Override

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

@@ -22,6 +22,7 @@ import org.elasticsearch.core.RestApiVersion;
 import org.elasticsearch.index.VersionType;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.RestRequest;
+import org.elasticsearch.rest.RestUtils;
 import org.elasticsearch.rest.Scope;
 import org.elasticsearch.rest.ServerlessScope;
 import org.elasticsearch.rest.action.RestActions;
@@ -149,6 +150,7 @@ public class RestIndexAction extends BaseRestHandler {
         indexRequest.setIfPrimaryTerm(request.paramAsLong("if_primary_term", indexRequest.ifPrimaryTerm()));
         indexRequest.setRequireAlias(request.paramAsBoolean(DocWriteRequest.REQUIRE_ALIAS, indexRequest.isRequireAlias()));
         indexRequest.setRequireDataStream(request.paramAsBoolean(DocWriteRequest.REQUIRE_DATA_STREAM, indexRequest.isRequireDataStream()));
+        indexRequest.setIncludeSourceOnError(RestUtils.getIncludeSourceOnError(request));
         String sOpType = request.param("op_type");
         String waitForActiveShards = request.param("wait_for_active_shards");
         if (waitForActiveShards != null) {

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

@@ -21,6 +21,7 @@ import org.elasticsearch.core.RestApiVersion;
 import org.elasticsearch.index.VersionType;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.RestRequest;
+import org.elasticsearch.rest.RestUtils;
 import org.elasticsearch.rest.Scope;
 import org.elasticsearch.rest.ServerlessScope;
 import org.elasticsearch.rest.action.RestActions;
@@ -85,7 +86,7 @@ public class RestUpdateAction extends BaseRestHandler {
         updateRequest.setIfPrimaryTerm(request.paramAsLong("if_primary_term", updateRequest.ifPrimaryTerm()));
         updateRequest.setRequireAlias(request.paramAsBoolean(DocWriteRequest.REQUIRE_ALIAS, updateRequest.isRequireAlias()));
 
-        request.applyContentParser(parser -> {
+        request.applyContentParser(RestUtils.getIncludeSourceOnError(request), parser -> {
             updateRequest.fromXContent(parser);
             IndexRequest upsertRequest = updateRequest.upsertRequest();
             if (upsertRequest != null) {

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

@@ -32,7 +32,7 @@ public class BulkRequestParserTests extends ESTestCase {
             {}
             """);
 
-        BulkRequestParser parser = new BulkRequestParser(randomBoolean(), RestApiVersion.current());
+        BulkRequestParser parser = new BulkRequestParser(randomBoolean(), true, RestApiVersion.current());
         BulkRequestParser.IncrementalParser incrementalParser = parser.incrementalParser(
             null,
             null,
@@ -66,7 +66,7 @@ public class BulkRequestParserTests extends ESTestCase {
         ArrayList<DocWriteRequest<?>> updateRequests = new ArrayList<>();
         ArrayList<DocWriteRequest<?>> deleteRequests = new ArrayList<>();
 
-        BulkRequestParser parser = new BulkRequestParser(randomBoolean(), RestApiVersion.current());
+        BulkRequestParser parser = new BulkRequestParser(randomBoolean(), true, RestApiVersion.current());
         BulkRequestParser.IncrementalParser incrementalParser = parser.incrementalParser(
             null,
             null,
@@ -112,7 +112,7 @@ public class BulkRequestParserTests extends ESTestCase {
             { "index":{ "_id": "bar" } }
             {}
             """);
-        BulkRequestParser parser = new BulkRequestParser(randomBoolean(), RestApiVersion.current());
+        BulkRequestParser parser = new BulkRequestParser(randomBoolean(), true, RestApiVersion.current());
         final AtomicBoolean parsed = new AtomicBoolean();
         parser.parse(request, "foo", null, null, null, null, null, null, false, XContentType.JSON, (indexRequest, type) -> {
             assertFalse(parsed.get());
@@ -148,7 +148,7 @@ public class BulkRequestParserTests extends ESTestCase {
         BytesArray request = new BytesArray("""
             { "delete":{ "_id": "bar" } }
             """);
-        BulkRequestParser parser = new BulkRequestParser(randomBoolean(), RestApiVersion.current());
+        BulkRequestParser parser = new BulkRequestParser(randomBoolean(), true, RestApiVersion.current());
         final AtomicBoolean parsed = new AtomicBoolean();
         parser.parse(
             request,
@@ -178,7 +178,7 @@ public class BulkRequestParserTests extends ESTestCase {
             { "update":{ "_id": "bar" } }
             {}
             """);
-        BulkRequestParser parser = new BulkRequestParser(randomBoolean(), RestApiVersion.current());
+        BulkRequestParser parser = new BulkRequestParser(randomBoolean(), true, RestApiVersion.current());
         final AtomicBoolean parsed = new AtomicBoolean();
         parser.parse(request, "foo", null, null, null, null, null, null, false, XContentType.JSON, (req, type) -> fail(), updateRequest -> {
             assertFalse(parsed.get());
@@ -214,7 +214,7 @@ public class BulkRequestParserTests extends ESTestCase {
         BytesArray request = new BytesArray("""
             { "index":{ "_id": "bar" } }
             {}""");
-        BulkRequestParser parser = new BulkRequestParser(randomBoolean(), RestApiVersion.current());
+        BulkRequestParser parser = new BulkRequestParser(randomBoolean(), true, RestApiVersion.current());
         IllegalArgumentException e = expectThrows(
             IllegalArgumentException.class,
             () -> parser.parse(
@@ -262,7 +262,7 @@ public class BulkRequestParserTests extends ESTestCase {
             { "index":{ "_index": "foo", "_id": "bar" } }
             {}
             """);
-        BulkRequestParser parser = new BulkRequestParser(randomBoolean(), RestApiVersion.current());
+        BulkRequestParser parser = new BulkRequestParser(randomBoolean(), true, RestApiVersion.current());
 
         IllegalArgumentException ex = expectThrows(
             IllegalArgumentException.class,
@@ -290,7 +290,7 @@ public class BulkRequestParserTests extends ESTestCase {
             { "index":{ "_type": "quux", "_id": "bar" } }
             {}
             """);
-        BulkRequestParser parser = new BulkRequestParser(false, RestApiVersion.current());
+        BulkRequestParser parser = new BulkRequestParser(false, true, RestApiVersion.current());
         final AtomicBoolean parsed = new AtomicBoolean();
         parser.parse(request, "foo", null, null, null, null, null, null, false, XContentType.JSON, (indexRequest, type) -> {
             assertFalse(parsed.get());
@@ -309,7 +309,7 @@ public class BulkRequestParserTests extends ESTestCase {
             { "index":{ "_index": "bar", "pipeline": "foo", "routing": "blub" } }
             {}
             """);
-        BulkRequestParser parser = new BulkRequestParser(randomBoolean(), RestApiVersion.current());
+        BulkRequestParser parser = new BulkRequestParser(randomBoolean(), true, RestApiVersion.current());
         final List<IndexRequest> indexRequests = new ArrayList<>();
         parser.parse(
             request,
@@ -339,7 +339,7 @@ public class BulkRequestParserTests extends ESTestCase {
             { "invalidaction":{ } }
             {}
             """);
-        BulkRequestParser parser = new BulkRequestParser(randomBoolean(), randomFrom(RestApiVersion.values()));
+        BulkRequestParser parser = new BulkRequestParser(randomBoolean(), true, randomFrom(RestApiVersion.values()));
 
         IllegalArgumentException ex = expectThrows(
             IllegalArgumentException.class,
@@ -370,7 +370,7 @@ public class BulkRequestParserTests extends ESTestCase {
             { "index":{ "_id": "bar" } }
             {}
             """);
-        BulkRequestParser parser = new BulkRequestParser(randomBoolean(), RestApiVersion.current());
+        BulkRequestParser parser = new BulkRequestParser(randomBoolean(), true, RestApiVersion.current());
         parser.parse(request, "foo", null, null, null, null, null, null, false, XContentType.JSON, (indexRequest, type) -> {
             assertFalse(indexRequest.getListExecutedPipelines());
         }, req -> fail(), req -> fail());

server/src/test/java/org/elasticsearch/index/mapper/DynamicTemplatesTests.java

@@ -22,7 +22,6 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.index.IndexVersion;
 import org.elasticsearch.index.IndexVersions;
-import org.elasticsearch.plugins.internal.XContentMeteringParserDecorator;
 import org.elasticsearch.test.XContentTestUtils;
 import org.elasticsearch.test.index.IndexVersionUtils;
 import org.elasticsearch.xcontent.XContentBuilder;
@@ -733,16 +732,7 @@ public class DynamicTemplatesTests extends MapperServiceTestCase {
             {"foo": "41.12,-71.34", "bar": "41.12,-71.34"}
             """;
         ParsedDocument doc = mapperService.documentMapper()
-            .parse(
-                new SourceToParse(
-                    "1",
-                    new BytesArray(json),
-                    XContentType.JSON,
-                    null,
-                    Map.of("foo", "geo_point"),
-                    XContentMeteringParserDecorator.NOOP
-                )
-            );
+            .parse(new SourceToParse("1", new BytesArray(json), XContentType.JSON, null, Map.of("foo", "geo_point")));
         assertThat(doc.rootDoc().getFields("foo"), hasSize(1));
         assertThat(doc.rootDoc().getFields("bar"), hasSize(1));
     }

test/framework/src/main/java/org/elasticsearch/index/mapper/MapperServiceTestCase.java

@@ -61,7 +61,6 @@ import org.elasticsearch.indices.breaker.NoneCircuitBreakerService;
 import org.elasticsearch.plugins.MapperPlugin;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.plugins.TelemetryPlugin;
-import org.elasticsearch.plugins.internal.XContentMeteringParserDecorator;
 import org.elasticsearch.script.Script;
 import org.elasticsearch.script.ScriptCompiler;
 import org.elasticsearch.script.ScriptContext;
@@ -389,14 +388,7 @@ public abstract class MapperServiceTestCase extends FieldTypeTestCase {
         XContentBuilder builder = JsonXContent.contentBuilder().startObject();
         build.accept(builder);
         builder.endObject();
-        return new SourceToParse(
-            id,
-            BytesReference.bytes(builder),
-            XContentType.JSON,
-            routing,
-            dynamicTemplates,
-            XContentMeteringParserDecorator.NOOP
-        );
+        return new SourceToParse(id, BytesReference.bytes(builder), XContentType.JSON, routing, dynamicTemplates);
     }
 
     /**

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/monitoring/action/MonitoringBulkRequest.java

@@ -83,7 +83,7 @@ public class MonitoringBulkRequest extends ActionRequest {
     ) throws IOException {
 
         // MonitoringBulkRequest accepts a body request that has the same format as the BulkRequest
-        new BulkRequestParser(false, RestApiVersion.current()).parse(
+        new BulkRequestParser(false, true, RestApiVersion.current()).parse(
             content,
             null,
             null,

x-pack/qa/runtime-fields/src/main/java/org/elasticsearch/xpack/runtimefields/test/CoreTestTranslater.java

@@ -366,7 +366,7 @@ public abstract class CoreTestTranslater {
                     bos.write(JsonXContent.jsonXContent.bulkSeparator());
                 }
                 List<IndexRequest> indexRequests = new ArrayList<>();
-                new BulkRequestParser(false, RestApiVersion.current()).parse(
+                new BulkRequestParser(false, true, RestApiVersion.current()).parse(
                     bos.bytes(),
                     defaultIndex,
                     defaultRouting,