Add source context to index request (#134474)

Add a source context to manage lifecycle of source. This commit is only
a first step adding the context.

Author: Tim-Brooks

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

@@ -324,6 +324,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion MAPPINGS_IN_DATA_STREAMS = def(9_112_0_00);
     public static final TransportVersion INFERENCE_API_EIS_DIAGNOSTICS = def(9_156_0_00);
     public static final TransportVersion ML_INFERENCE_ENDPOINT_CACHE = def(9_157_0_00);
+    public static final TransportVersion INDEX_SOURCE = def(9_158_0_00);
 
     /*
      * STOP! READ THIS FIRST! No, really,

server/src/main/java/org/elasticsearch/action/DocWriteRequest.java

@@ -9,6 +9,7 @@
 package org.elasticsearch.action;
 
 import org.apache.lucene.util.Accountable;
+import org.elasticsearch.action.bulk.TransportAbstractBulkAction;
 import org.elasticsearch.action.delete.DeleteRequest;
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.support.IndicesOptions;
@@ -20,6 +21,7 @@
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.lucene.uid.Versions;
 import org.elasticsearch.core.Nullable;
+import org.elasticsearch.core.Releasable;
 import org.elasticsearch.index.Index;
 import org.elasticsearch.index.VersionType;
 import org.elasticsearch.index.shard.ShardId;
@@ -37,7 +39,7 @@
  * Generic interface to group ActionRequest, which perform writes to a single document
  * Action requests implementing this can be part of {@link org.elasticsearch.action.bulk.BulkRequest}
  */
-public interface DocWriteRequest<T> extends IndicesRequest, Accountable {
+public interface DocWriteRequest<T> extends IndicesRequest, Accountable, Releasable {
 
     // Flag set for disallowing index auto creation for an individual write request.
     String REQUIRE_ALIAS = "require_alias";
@@ -346,4 +348,12 @@ static ActionRequestValidationException validateDocIdLength(String id, ActionReq
         }
         return validationException;
     }
+
+    @Override
+    default void close() {
+        IndexRequest indexRequest = TransportAbstractBulkAction.getIndexWriteRequest(this);
+        if (indexRequest != null) {
+            indexRequest.indexSource().close();
+        }
+    }
 }

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

@@ -182,7 +182,7 @@ BulkRequest internalAdd(IndexRequest request) {
 
         requests.add(request);
         // lack of source is validated in validate() method
-        sizeInBytes += (request.source() != null ? request.source().length() : 0) + REQUEST_OVERHEAD;
+        sizeInBytes += request.indexSource().byteLength() + REQUEST_OVERHEAD;
         indices.add(request.index());
         return this;
     }
@@ -200,10 +200,10 @@ BulkRequest internalAdd(UpdateRequest request) {
 
         requests.add(request);
         if (request.doc() != null) {
-            sizeInBytes += request.doc().source().length();
+            sizeInBytes += request.doc().indexSource().byteLength();
         }
         if (request.upsertRequest() != null) {
-            sizeInBytes += request.upsertRequest().source().length();
+            sizeInBytes += request.upsertRequest().indexSource().byteLength();
         }
         if (request.script() != null) {
             sizeInBytes += request.script().getIdOrCode().length() * 2;

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

@@ -296,7 +296,7 @@ public void markItemForFailureStore(int slot, String targetIndexName, Exception
                     () -> "Encountered exception while attempting to redirect a failed ingest operation: index ["
                         + targetIndexName
                         + "], source: ["
-                        + indexRequest.source().utf8ToString()
+                        + indexRequest.indexSource().bytes().utf8ToString()
                         + "]",
                     ioException
                 );

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

@@ -90,13 +90,11 @@ public long totalSizeInBytes() {
         for (int i = 0; i < items.length; i++) {
             DocWriteRequest<?> request = items[i].request();
             if (request instanceof IndexRequest) {
-                if (((IndexRequest) request).source() != null) {
-                    totalSizeInBytes += ((IndexRequest) request).source().length();
-                }
+                totalSizeInBytes += ((IndexRequest) request).indexSource().byteLength();
             } else if (request instanceof UpdateRequest) {
                 IndexRequest doc = ((UpdateRequest) request).doc();
-                if (doc != null && doc.source() != null) {
-                    totalSizeInBytes += ((UpdateRequest) request).doc().source().length();
+                if (doc != null) {
+                    totalSizeInBytes += ((UpdateRequest) request).doc().indexSource().byteLength();
                 }
             }
         }
@@ -108,13 +106,11 @@ public long maxOperationSizeInBytes() {
         for (int i = 0; i < items.length; i++) {
             DocWriteRequest<?> request = items[i].request();
             if (request instanceof IndexRequest) {
-                if (((IndexRequest) request).source() != null) {
-                    maxOperationSizeInBytes = Math.max(maxOperationSizeInBytes, ((IndexRequest) request).source().length());
-                }
+                maxOperationSizeInBytes = Math.max(maxOperationSizeInBytes, ((IndexRequest) request).indexSource().byteLength());
             } else if (request instanceof UpdateRequest) {
                 IndexRequest doc = ((UpdateRequest) request).doc();
-                if (doc != null && doc.source() != null) {
-                    maxOperationSizeInBytes = Math.max(maxOperationSizeInBytes, ((UpdateRequest) request).doc().source().length());
+                if (doc != null) {
+                    maxOperationSizeInBytes = Math.max(maxOperationSizeInBytes, ((UpdateRequest) request).doc().indexSource().byteLength());
                 }
             }
         }