Release source bytes when source rewritten

client/client-benchmark-noop-api-plugin/src/main/java/org/elasticsearch/plugin/noop/action/bulk/RestNoopBulkAction.java

@@ -8,6 +8,7 @@
  */
 package org.elasticsearch.plugin.noop.action.bulk;
 
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.DocWriteRequest;
 import org.elasticsearch.action.DocWriteResponse;
 import org.elasticsearch.action.bulk.BulkItemResponse;
@@ -16,6 +17,7 @@
 import org.elasticsearch.action.support.ActiveShardCount;
 import org.elasticsearch.action.update.UpdateResponse;
 import org.elasticsearch.client.internal.node.NodeClient;
+import org.elasticsearch.core.Releasables;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.RestChannel;
@@ -25,6 +27,7 @@
 import org.elasticsearch.xcontent.XContentBuilder;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.List;
 
 import static org.elasticsearch.rest.RestRequest.Method.POST;
@@ -78,10 +81,12 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
             request.getRestApiVersion()
         );
 
+        // The actual bulk request items are mutable during the bulk process so we must create a copy
+        List<DocWriteRequest<?>> toClose = new ArrayList<>(bulkRequest.requests());
         // short circuit the call to the transport layer
         return channel -> {
             BulkRestBuilderListener listener = new BulkRestBuilderListener(channel, request);
-            listener.onResponse(bulkRequest);
+            ActionListener.releaseAfter(listener, () -> Releasables.close(toClose)).onResponse(bulkRequest);
         };
     }
 

modules/data-streams/src/internalClusterTest/java/org/elasticsearch/datastreams/FailureStoreMetricsWithIncrementalBulkIT.java

@@ -18,14 +18,14 @@
 import org.elasticsearch.action.datastreams.CreateDataStreamAction;
 import org.elasticsearch.action.datastreams.GetDataStreamAction;
 import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.action.index.IndexSource;
 import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.cluster.metadata.ComposableIndexTemplate;
 import org.elasticsearch.cluster.metadata.DataStream;
 import org.elasticsearch.cluster.metadata.DataStreamTestHelper;
 import org.elasticsearch.cluster.metadata.Template;
 import org.elasticsearch.common.compress.CompressedXContent;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.core.AbstractRefCounted;
 import org.elasticsearch.core.Releasable;
 import org.elasticsearch.core.Strings;
 import org.elasticsearch.index.Index;
@@ -89,16 +89,17 @@ public void testShortCircuitFailure() throws Exception {
 
         String coordinatingOnlyNode = internalCluster().startCoordinatingOnlyNode(Settings.EMPTY);
 
-        AbstractRefCounted refCounted = AbstractRefCounted.of(() -> {});
+        final ArrayList<IndexSource> contextsToRelease = new ArrayList<>();
         IncrementalBulkService incrementalBulkService = internalCluster().getInstance(IncrementalBulkService.class, coordinatingOnlyNode);
         try (IncrementalBulkService.Handler handler = incrementalBulkService.newBulkRequest()) {
 
             AtomicBoolean nextRequested = new AtomicBoolean(true);
             int successfullyStored = 0;
             while (nextRequested.get()) {
                 nextRequested.set(false);
-                refCounted.incRef();
-                handler.addItems(List.of(indexRequest(DATA_STREAM_NAME)), refCounted::decRef, () -> nextRequested.set(true));
+                IndexRequest indexRequest = indexRequest(DATA_STREAM_NAME);
+                contextsToRelease.add(indexRequest.indexSource());
+                handler.addItems(List.of(indexRequest), () -> nextRequested.set(true));
                 successfullyStored++;
             }
             assertBusy(() -> assertTrue(nextRequested.get()));
@@ -116,29 +117,34 @@ public void testShortCircuitFailure() throws Exception {
                 while (primaryPressure.stats().getPrimaryRejections() == primaryRejections) {
                     while (nextRequested.get()) {
                         nextRequested.set(false);
-                        refCounted.incRef();
                         List<DocWriteRequest<?>> requests = new ArrayList<>();
                         for (int i = 0; i < 20; ++i) {
-                            requests.add(indexRequest(DATA_STREAM_NAME));
+                            IndexRequest indexRequest = indexRequest(DATA_STREAM_NAME);
+                            contextsToRelease.add(indexRequest.indexSource());
+                            requests.add(indexRequest);
                         }
-                        handler.addItems(requests, refCounted::decRef, () -> nextRequested.set(true));
+                        handler.addItems(requests, () -> nextRequested.set(true));
                     }
                     assertBusy(() -> assertTrue(nextRequested.get()));
                 }
             }
 
             while (nextRequested.get()) {
                 nextRequested.set(false);
-                refCounted.incRef();
-                handler.addItems(List.of(indexRequest(DATA_STREAM_NAME)), refCounted::decRef, () -> nextRequested.set(true));
+                IndexRequest indexRequest = indexRequest(DATA_STREAM_NAME);
+                contextsToRelease.add(indexRequest.indexSource());
+                handler.addItems(List.of(indexRequest), () -> nextRequested.set(true));
             }
 
             assertBusy(() -> assertTrue(nextRequested.get()));
 
             PlainActionFuture<BulkResponse> future = new PlainActionFuture<>();
-            handler.lastItems(List.of(indexRequest(DATA_STREAM_NAME)), refCounted::decRef, future);
+            IndexRequest lastRequest = indexRequest(DATA_STREAM_NAME);
+            contextsToRelease.add(lastRequest.indexSource());
+            handler.lastItems(List.of(lastRequest), future);
 
             BulkResponse bulkResponse = safeGet(future);
+            assertThat(contextsToRelease.stream().filter(c -> c.isClosed() == false).count(), equalTo(0L));
 
             for (int i = 0; i < bulkResponse.getItems().length; ++i) {
                 // the first requests were successful

qa/smoke-test-http/src/internalClusterTest/java/org/elasticsearch/http/BulkRestIT.java

@@ -52,11 +52,7 @@ public void testBulkUriMatchingDoesNotMatchBulkCapabilitiesApi() throws IOExcept
     }
 
     public void testBulkMissingBody() throws IOException {
-        Request request = new Request(randomBoolean() ? "POST" : "PUT", "/_bulk");
-        request.setJsonEntity("");
-        ResponseException responseException = expectThrows(ResponseException.class, () -> getRestClient().performRequest(request));
-        assertEquals(400, responseException.getResponse().getStatusLine().getStatusCode());
-        assertThat(responseException.getMessage(), containsString("request body is required"));
+        sendMissingBody();
     }
 
     public void testBulkInvalidIndexNameString() throws IOException {
@@ -79,16 +75,7 @@ public void testBulkInvalidIndexNameString() throws IOException {
     }
 
     public void testBulkRequestBodyImproperlyTerminated() throws IOException {
-        Request request = new Request(randomBoolean() ? "POST" : "PUT", "/_bulk");
-        // missing final line of the bulk body. cannot process
-        request.setJsonEntity(
-            "{\"index\":{\"_index\":\"index_name\",\"_id\":\"1\"}}\n"
-                + "{\"field\":1}\n"
-                + "{\"index\":{\"_index\":\"index_name\",\"_id\":\"2\"}"
-        );
-        ResponseException responseException = expectThrows(ResponseException.class, () -> getRestClient().performRequest(request));
-        assertEquals(400, responseException.getResponse().getStatusLine().getStatusCode());
-        assertThat(responseException.getMessage(), containsString("The bulk request must be terminated by a newline"));
+        sendImproperlyTerminated();
     }
 
     public void testBulkRequest() throws IOException {
@@ -156,6 +143,9 @@ public void testBulkWithIncrementalDisabled() throws IOException {
 
         try {
             sendLargeBulk();
+            sendMalFormedActionLine();
+            sendImproperlyTerminated();
+            sendMissingBody();
         } finally {
             internalCluster().getInstances(IncrementalBulkService.class).forEach(i -> i.setForTests(true));
             updateClusterSettings(Settings.builder().put(IncrementalBulkService.INCREMENTAL_BULK.getKey(), (String) null));
@@ -177,6 +167,31 @@ public void testMalformedActionLineBulk() throws IOException {
         final Response indexCreatedResponse = getRestClient().performRequest(createRequest);
         assertThat(indexCreatedResponse.getStatusLine().getStatusCode(), equalTo(OK.getStatus()));
 
+        sendMalFormedActionLine();
+    }
+
+    private static void sendMissingBody() {
+        Request request = new Request(randomBoolean() ? "POST" : "PUT", "/_bulk");
+        request.setJsonEntity("");
+        ResponseException responseException = expectThrows(ResponseException.class, () -> getRestClient().performRequest(request));
+        assertEquals(400, responseException.getResponse().getStatusLine().getStatusCode());
+        assertThat(responseException.getMessage(), containsString("request body is required"));
+    }
+
+    private static void sendImproperlyTerminated() {
+        Request request = new Request(randomBoolean() ? "POST" : "PUT", "/_bulk");
+        // missing final line of the bulk body. cannot process
+        request.setJsonEntity(
+            "{\"index\":{\"_index\":\"index_name\",\"_id\":\"1\"}}\n"
+                + "{\"field\":1}\n"
+                + "{\"index\":{\"_index\":\"index_name\",\"_id\":\"2\"}"
+        );
+        ResponseException responseException = expectThrows(ResponseException.class, () -> getRestClient().performRequest(request));
+        assertEquals(400, responseException.getResponse().getStatusLine().getStatusCode());
+        assertThat(responseException.getMessage(), containsString("The bulk request must be terminated by a newline"));
+    }
+
+    private static void sendMalFormedActionLine() throws IOException {
         Request bulkRequest = new Request("POST", "/index_name/_bulk");
 
         final StringBuilder bulk = new StringBuilder();

qa/system-indices/src/main/java/org/elasticsearch/system/indices/SystemIndicesQA.java

@@ -180,10 +180,10 @@ public List<Route> routes() {
         protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
             var content = request.requiredContent();
             IndexRequest indexRequest = new IndexRequest(".net-new-system-index-primary");
-            indexRequest.source(content, request.getXContentType());
+            indexRequest.indexSource().source(content.retain(), request.getXContentType());
             indexRequest.id(request.param("id"));
             indexRequest.setRefreshPolicy(request.param("refresh"));
-            return channel -> client.index(indexRequest, ActionListener.withRef(new RestToXContentListener<>(channel), content));
+            return channel -> client.index(indexRequest, ActionListener.releaseAfter(new RestToXContentListener<>(channel), indexRequest));
         }
 
         @Override