Implement lifecycle on SimulatePipelineRequest (#117585)

Rather than releasing the REST request body after computing the
response, we can link the lifecycles of the REST and transport requests
and release the REST request body sooner. Not that we expect these
bodies to be particularly large in this case, but still it's a better
pattern to follow.

Author: DaveCTurner

modules/ingest-geoip/src/internalClusterTest/java/org/elasticsearch/ingest/geoip/GeoIpDownloaderIT.java

@@ -41,7 +41,6 @@
 import org.elasticsearch.search.sort.SortOrder;
 import org.elasticsearch.test.junit.annotations.TestLogging;
 import org.elasticsearch.xcontent.XContentBuilder;
-import org.elasticsearch.xcontent.XContentType;
 import org.elasticsearch.xcontent.json.JsonXContent;
 import org.junit.After;
 
@@ -67,6 +66,7 @@
 import java.util.zip.GZIPInputStream;
 
 import static org.elasticsearch.ingest.ConfigurationUtils.readStringProperty;
+import static org.elasticsearch.ingest.IngestPipelineTestUtils.jsonSimulatePipelineRequest;
 import static org.elasticsearch.ingest.geoip.GeoIpTestUtils.copyDefaultDatabases;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertResponse;
@@ -494,7 +494,7 @@ private SimulateDocumentBaseResult simulatePipeline() throws IOException {
             builder.endObject();
             bytes = BytesReference.bytes(builder);
         }
-        SimulatePipelineRequest simulateRequest = new SimulatePipelineRequest(bytes, XContentType.JSON);
+        SimulatePipelineRequest simulateRequest = jsonSimulatePipelineRequest(bytes);
         simulateRequest.setId("_id");
         // Avoid executing on a coordinating only node, because databases are not available there and geoip processor won't do any lookups.
         // (some test seeds repeatedly hit such nodes causing failures)

server/src/internalClusterTest/java/org/elasticsearch/ingest/IngestClientIT.java

@@ -37,6 +37,7 @@
 import java.util.List;
 import java.util.Map;
 
+import static org.elasticsearch.ingest.IngestPipelineTestUtils.jsonSimulatePipelineRequest;
 import static org.elasticsearch.ingest.IngestPipelineTestUtils.putJsonPipelineRequest;
 import static org.elasticsearch.test.NodeRoles.nonIngestNode;
 import static org.elasticsearch.xcontent.XContentFactory.jsonBuilder;
@@ -97,7 +98,7 @@ public void testSimulate() throws Exception {
         if (randomBoolean()) {
             response = clusterAdmin().prepareSimulatePipeline(bytes, XContentType.JSON).setId("_id").get();
         } else {
-            SimulatePipelineRequest request = new SimulatePipelineRequest(bytes, XContentType.JSON);
+            SimulatePipelineRequest request = jsonSimulatePipelineRequest(bytes);
             request.setId("_id");
             response = clusterAdmin().simulatePipeline(request).get();
         }

server/src/main/java/org/elasticsearch/action/ingest/SimulatePipelineRequest.java

@@ -12,6 +12,7 @@
 import org.elasticsearch.action.ActionRequest;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.bytes.ReleasableBytesReference;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.logging.DeprecationLogger;
@@ -41,19 +42,20 @@ public class SimulatePipelineRequest extends ActionRequest implements ToXContent
     private static final DeprecationLogger deprecationLogger = DeprecationLogger.getLogger(SimulatePipelineRequest.class);
     private String id;
     private boolean verbose;
-    private final BytesReference source;
+    private final ReleasableBytesReference source;
     private final XContentType xContentType;
     private RestApiVersion restApiVersion;
 
     /**
      * Creates a new request with the given source and its content type
      */
-    public SimulatePipelineRequest(BytesReference source, XContentType xContentType) {
+    public SimulatePipelineRequest(ReleasableBytesReference source, XContentType xContentType) {
         this(source, xContentType, RestApiVersion.current());
     }
 
-    public SimulatePipelineRequest(BytesReference source, XContentType xContentType, RestApiVersion restApiVersion) {
+    public SimulatePipelineRequest(ReleasableBytesReference source, XContentType xContentType, RestApiVersion restApiVersion) {
         this.source = Objects.requireNonNull(source);
+        assert source.hasReferences();
         this.xContentType = Objects.requireNonNull(xContentType);
         this.restApiVersion = restApiVersion;
     }
@@ -62,7 +64,7 @@ public SimulatePipelineRequest(BytesReference source, XContentType xContentType,
         super(in);
         id = in.readOptionalString();
         verbose = in.readBoolean();
-        source = in.readBytesReference();
+        source = in.readReleasableBytesReference();
         xContentType = in.readEnum(XContentType.class);
     }
 
@@ -88,6 +90,7 @@ public void setVerbose(boolean verbose) {
     }
 
     public BytesReference getSource() {
+        assert source.hasReferences();
         return source;
     }
 
@@ -250,4 +253,24 @@ private static List<IngestDocument> parseDocs(Map<String, Object> config, RestAp
     public RestApiVersion getRestApiVersion() {
         return restApiVersion;
     }
+
+    @Override
+    public final void incRef() {
+        source.incRef();
+    }
+
+    @Override
+    public final boolean tryIncRef() {
+        return source.tryIncRef();
+    }
+
+    @Override
+    public final boolean decRef() {
+        return source.decRef();
+    }
+
+    @Override
+    public final boolean hasReferences() {
+        return source.hasReferences();
+    }
 }

server/src/main/java/org/elasticsearch/action/ingest/SimulatePipelineRequestBuilder.java

@@ -12,6 +12,7 @@
 import org.elasticsearch.action.ActionRequestBuilder;
 import org.elasticsearch.client.internal.ElasticsearchClient;
 import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.bytes.ReleasableBytesReference;
 import org.elasticsearch.xcontent.XContentType;
 
 public class SimulatePipelineRequestBuilder extends ActionRequestBuilder<SimulatePipelineRequest, SimulatePipelineResponse> {
@@ -20,7 +21,7 @@ public class SimulatePipelineRequestBuilder extends ActionRequestBuilder<Simulat
      * Create a new builder for {@link SimulatePipelineRequest}s
      */
     public SimulatePipelineRequestBuilder(ElasticsearchClient client, BytesReference source, XContentType xContentType) {
-        super(client, SimulatePipelineAction.INSTANCE, new SimulatePipelineRequest(source, xContentType));
+        super(client, SimulatePipelineAction.INSTANCE, new SimulatePipelineRequest(ReleasableBytesReference.wrap(source), xContentType));
     }
 
     /**

server/src/main/java/org/elasticsearch/rest/action/ingest/RestSimulatePipelineAction.java

@@ -9,7 +9,6 @@
 
 package org.elasticsearch.rest.action.ingest;
 
-import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ingest.SimulatePipelineRequest;
 import org.elasticsearch.client.internal.node.NodeClient;
 import org.elasticsearch.common.bytes.ReleasableBytesReference;
@@ -48,12 +47,9 @@ public String getName() {
     @Override
     public RestChannelConsumer prepareRequest(RestRequest restRequest, NodeClient client) throws IOException {
         Tuple<XContentType, ReleasableBytesReference> sourceTuple = restRequest.contentOrSourceParam();
-        var content = sourceTuple.v2();
-        SimulatePipelineRequest request = new SimulatePipelineRequest(sourceTuple.v2(), sourceTuple.v1(), restRequest.getRestApiVersion());
+        final var request = new SimulatePipelineRequest(sourceTuple.v2(), sourceTuple.v1(), restRequest.getRestApiVersion());
         request.setId(restRequest.param("id"));
         request.setVerbose(restRequest.paramAsBoolean("verbose", false));
-        return channel -> client.admin()
-            .cluster()
-            .simulatePipeline(request, ActionListener.withRef(new RestToXContentListener<>(channel), content));
+        return channel -> client.admin().cluster().simulatePipeline(request, new RestToXContentListener<>(channel));
     }
 }

server/src/test/java/org/elasticsearch/action/ingest/SimulatePipelineRequestTests.java

@@ -16,14 +16,14 @@
 import org.elasticsearch.xcontent.XContentType;
 
 import java.io.IOException;
-import java.nio.charset.StandardCharsets;
 
+import static org.elasticsearch.ingest.IngestPipelineTestUtils.jsonSimulatePipelineRequest;
 import static org.hamcrest.CoreMatchers.equalTo;
 
 public class SimulatePipelineRequestTests extends ESTestCase {
 
     public void testSerialization() throws IOException {
-        SimulatePipelineRequest request = new SimulatePipelineRequest(new BytesArray(""), XContentType.JSON);
+        SimulatePipelineRequest request = jsonSimulatePipelineRequest(new BytesArray(""));
         // Sometimes we set an id
         if (randomBoolean()) {
             request.setId(randomAlphaOfLengthBetween(1, 10));
@@ -44,10 +44,7 @@ public void testSerialization() throws IOException {
     }
 
     public void testSerializationWithXContent() throws IOException {
-        SimulatePipelineRequest request = new SimulatePipelineRequest(
-            new BytesArray("{}".getBytes(StandardCharsets.UTF_8)),
-            XContentType.JSON
-        );
+        SimulatePipelineRequest request = jsonSimulatePipelineRequest("{}");
         assertEquals(XContentType.JSON, request.getXContentType());
 
         BytesStreamOutput output = new BytesStreamOutput();

test/framework/src/main/java/org/elasticsearch/ingest/IngestPipelineTestUtils.java

@@ -14,11 +14,13 @@
 import org.elasticsearch.action.ingest.DeletePipelineTransportAction;
 import org.elasticsearch.action.ingest.PutPipelineRequest;
 import org.elasticsearch.action.ingest.PutPipelineTransportAction;
+import org.elasticsearch.action.ingest.SimulatePipelineRequest;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.client.internal.ElasticsearchClient;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.bytes.ReleasableBytesReference;
 import org.elasticsearch.logging.LogManager;
 import org.elasticsearch.logging.Logger;
 import org.elasticsearch.test.ESTestCase;
@@ -124,4 +126,18 @@ public void onFailure(Exception e) {
             );
         }
     }
+
+    /**
+     * Construct a new {@link SimulatePipelineRequest} whose content is the given JSON document, represented as a {@link String}.
+     */
+    public static SimulatePipelineRequest jsonSimulatePipelineRequest(String jsonString) {
+        return jsonSimulatePipelineRequest(new BytesArray(jsonString));
+    }
+
+    /**
+     * Construct a new {@link SimulatePipelineRequest} whose content is the given JSON document, represented as a {@link BytesReference}.
+     */
+    public static SimulatePipelineRequest jsonSimulatePipelineRequest(BytesReference jsonBytes) {
+        return new SimulatePipelineRequest(ReleasableBytesReference.wrap(jsonBytes), XContentType.JSON);
+    }
 }

x-pack/plugin/enrich/src/internalClusterTest/java/org/elasticsearch/xpack/enrich/EnrichProcessorIT.java

@@ -9,9 +9,7 @@
 
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.ingest.SimulateDocumentBaseResult;
-import org.elasticsearch.action.ingest.SimulatePipelineRequest;
 import org.elasticsearch.action.support.WriteRequest;
-import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.ingest.common.IngestCommonPlugin;
 import org.elasticsearch.plugins.Plugin;
@@ -27,6 +25,7 @@
 import java.util.Collection;
 import java.util.List;
 
+import static org.elasticsearch.ingest.IngestPipelineTestUtils.jsonSimulatePipelineRequest;
 import static org.elasticsearch.xpack.enrich.AbstractEnrichTestCase.createSourceIndices;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.nullValue;
@@ -90,7 +89,7 @@ public void testEnrichCacheValuesCannotBeCorrupted() {
         var executePolicyRequest = new ExecuteEnrichPolicyAction.Request(TEST_REQUEST_TIMEOUT, policyName);
         client().execute(ExecuteEnrichPolicyAction.INSTANCE, executePolicyRequest).actionGet();
 
-        var simulatePipelineRequest = new SimulatePipelineRequest(new BytesArray("""
+        var simulatePipelineRequest = jsonSimulatePipelineRequest("""
             {
               "pipeline": {
                 "processors": [
@@ -119,7 +118,7 @@ public void testEnrichCacheValuesCannotBeCorrupted() {
                 }
               ]
             }
-            """), XContentType.JSON);
+            """);
         var response = clusterAdmin().simulatePipeline(simulatePipelineRequest).actionGet();
         var result = (SimulateDocumentBaseResult) response.getResults().get(0);
         assertThat(result.getFailure(), nullValue());
@@ -132,7 +131,7 @@ public void testEnrichCacheValuesCannotBeCorrupted() {
         assertThat(statsResponse.getCacheStats().get(0).misses(), equalTo(1L));
         assertThat(statsResponse.getCacheStats().get(0).hits(), equalTo(0L));
 
-        simulatePipelineRequest = new SimulatePipelineRequest(new BytesArray("""
+        simulatePipelineRequest = jsonSimulatePipelineRequest("""
             {
               "pipeline": {
                 "processors": [
@@ -155,7 +154,7 @@ public void testEnrichCacheValuesCannotBeCorrupted() {
                 }
               ]
             }
-            """), XContentType.JSON);
+            """);
         response = clusterAdmin().simulatePipeline(simulatePipelineRequest).actionGet();
         result = (SimulateDocumentBaseResult) response.getResults().get(0);
         assertThat(result.getFailure(), nullValue());

x-pack/plugin/ml/src/internalClusterTest/java/org/elasticsearch/license/MachineLearningLicensingIT.java

@@ -11,14 +11,12 @@
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.ingest.SimulateDocumentBaseResult;
 import org.elasticsearch.action.ingest.SimulatePipelineAction;
-import org.elasticsearch.action.ingest.SimulatePipelineRequest;
 import org.elasticsearch.action.ingest.SimulatePipelineResponse;
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.action.support.WriteRequest;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.cluster.ClusterState;
-import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.core.Strings;
 import org.elasticsearch.core.TimeValue;
@@ -61,13 +59,13 @@
 import org.elasticsearch.xpack.ml.support.BaseMlIntegTestCase;
 import org.junit.Before;
 
-import java.nio.charset.StandardCharsets;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 
+import static org.elasticsearch.ingest.IngestPipelineTestUtils.jsonSimulatePipelineRequest;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.empty;
 import static org.hamcrest.Matchers.hasItem;
@@ -541,11 +539,7 @@ public void testMachineLearningCreateInferenceProcessorRestricted() {
                 }}]
             }""", pipeline);
         PlainActionFuture<SimulatePipelineResponse> simulatePipelineListener = new PlainActionFuture<>();
-        client().execute(
-            SimulatePipelineAction.INSTANCE,
-            new SimulatePipelineRequest(new BytesArray(simulateSource.getBytes(StandardCharsets.UTF_8)), XContentType.JSON),
-            simulatePipelineListener
-        );
+        client().execute(SimulatePipelineAction.INSTANCE, jsonSimulatePipelineRequest(simulateSource), simulatePipelineListener);
 
         assertThat(simulatePipelineListener.actionGet().getResults(), is(not(empty())));
 
@@ -575,7 +569,7 @@ public void testMachineLearningCreateInferenceProcessorRestricted() {
         // Simulating the pipeline should fail
         SimulateDocumentBaseResult simulateResponse = (SimulateDocumentBaseResult) client().execute(
             SimulatePipelineAction.INSTANCE,
-            new SimulatePipelineRequest(new BytesArray(simulateSource.getBytes(StandardCharsets.UTF_8)), XContentType.JSON)
+            jsonSimulatePipelineRequest(simulateSource)
         ).actionGet().getResults().get(0);
         assertThat(simulateResponse.getFailure(), is(not(nullValue())));
         assertThat((simulateResponse.getFailure()).getCause(), is(instanceOf(ElasticsearchSecurityException.class)));
@@ -588,11 +582,7 @@ public void testMachineLearningCreateInferenceProcessorRestricted() {
         putJsonPipeline("test_infer_license_pipeline", pipeline);
 
         PlainActionFuture<SimulatePipelineResponse> simulatePipelineListenerNewLicense = new PlainActionFuture<>();
-        client().execute(
-            SimulatePipelineAction.INSTANCE,
-            new SimulatePipelineRequest(new BytesArray(simulateSource.getBytes(StandardCharsets.UTF_8)), XContentType.JSON),
-            simulatePipelineListenerNewLicense
-        );
+        client().execute(SimulatePipelineAction.INSTANCE, jsonSimulatePipelineRequest(simulateSource), simulatePipelineListenerNewLicense);
 
         assertThat(simulatePipelineListenerNewLicense.actionGet().getResults(), is(not(empty())));
 

x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/action/TransportPreviewTransformAction.java

@@ -21,6 +21,7 @@
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.bytes.ReleasableBytesReference;
 import org.elasticsearch.common.logging.HeaderWarning;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
@@ -282,7 +283,10 @@ private void getPreview(
                     builder.startObject();
                     builder.field("docs", results);
                     builder.endObject();
-                    var pipelineRequest = new SimulatePipelineRequest(BytesReference.bytes(builder), XContentType.JSON);
+                    var pipelineRequest = new SimulatePipelineRequest(
+                        ReleasableBytesReference.wrap(BytesReference.bytes(builder)),
+                        XContentType.JSON
+                    );
                     pipelineRequest.setId(pipeline);
                     parentTaskClient.execute(SimulatePipelineAction.INSTANCE, pipelineRequest, pipelineResponseActionListener);
                 }