Use ref-count aware chunked-xcontent listener in most spots

client/client-benchmark-noop-api-plugin/src/main/java/org/elasticsearch/plugin/noop/action/search/RestNoopSearchAction.java

@@ -12,7 +12,7 @@
 import org.elasticsearch.plugin.noop.NoopPlugin;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.RestRequest;
-import org.elasticsearch.rest.action.RestChunkedToXContentListener;
+import org.elasticsearch.rest.action.RestRefCountedChunkedToXContentListener;
 
 import java.util.List;
 
@@ -39,6 +39,10 @@ public String getName() {
     @Override
     public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) {
         SearchRequest searchRequest = new SearchRequest();
-        return channel -> client.execute(NoopPlugin.NOOP_SEARCH_ACTION, searchRequest, new RestChunkedToXContentListener<>(channel));
+        return channel -> client.execute(
+            NoopPlugin.NOOP_SEARCH_ACTION,
+            searchRequest,
+            new RestRefCountedChunkedToXContentListener<>(channel)
+        );
     }
 }

modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/rest/RestDataStreamLifecycleStatsAction.java

@@ -14,7 +14,7 @@
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.Scope;
 import org.elasticsearch.rest.ServerlessScope;
-import org.elasticsearch.rest.action.RestChunkedToXContentListener;
+import org.elasticsearch.rest.action.RestRefCountedChunkedToXContentListener;
 
 import java.util.List;
 
@@ -40,6 +40,10 @@ protected RestChannelConsumer prepareRequest(RestRequest restRequest, NodeClient
         if (masterNodeTimeout != null) {
             request.masterNodeTimeout(masterNodeTimeout);
         }
-        return channel -> client.execute(GetDataStreamLifecycleStatsAction.INSTANCE, request, new RestChunkedToXContentListener<>(channel));
+        return channel -> client.execute(
+            GetDataStreamLifecycleStatsAction.INSTANCE,
+            request,
+            new RestRefCountedChunkedToXContentListener<>(channel)
+        );
     }
 }

modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/rest/RestExplainDataStreamLifecycleAction.java

@@ -16,7 +16,7 @@
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.Scope;
 import org.elasticsearch.rest.ServerlessScope;
-import org.elasticsearch.rest.action.RestChunkedToXContentListener;
+import org.elasticsearch.rest.action.RestRefCountedChunkedToXContentListener;
 
 import java.util.List;
 
@@ -48,7 +48,7 @@ protected RestChannelConsumer prepareRequest(RestRequest restRequest, NodeClient
         return channel -> client.execute(
             ExplainDataStreamLifecycleAction.INSTANCE,
             explainRequest,
-            new RestChunkedToXContentListener<>(channel)
+            new RestRefCountedChunkedToXContentListener<>(channel)
         );
     }
 

modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/rest/RestGetDataStreamLifecycleAction.java

@@ -15,7 +15,7 @@
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.Scope;
 import org.elasticsearch.rest.ServerlessScope;
-import org.elasticsearch.rest.action.RestChunkedToXContentListener;
+import org.elasticsearch.rest.action.RestRefCountedChunkedToXContentListener;
 
 import java.util.List;
 
@@ -44,7 +44,7 @@ protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient cli
         return channel -> client.execute(
             GetDataStreamLifecycleAction.INSTANCE,
             getDataLifecycleRequest,
-            new RestChunkedToXContentListener<>(channel)
+            new RestRefCountedChunkedToXContentListener<>(channel)
         );
     }
 

server/src/main/java/org/elasticsearch/health/RestGetHealthAction.java

@@ -14,7 +14,7 @@
 import org.elasticsearch.rest.Scope;
 import org.elasticsearch.rest.ServerlessScope;
 import org.elasticsearch.rest.action.RestCancellableNodeClient;
-import org.elasticsearch.rest.action.RestChunkedToXContentListener;
+import org.elasticsearch.rest.action.RestRefCountedChunkedToXContentListener;
 
 import java.io.IOException;
 import java.util.List;
@@ -48,7 +48,7 @@ protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient cli
         return channel -> new RestCancellableNodeClient(client, request.getHttpChannel()).execute(
             GetHealthAction.INSTANCE,
             getHealthRequest,
-            new RestChunkedToXContentListener<>(channel)
+            new RestRefCountedChunkedToXContentListener<>(channel)
         );
     }
 

server/src/main/java/org/elasticsearch/rest/action/RestFieldCapabilitiesAction.java

@@ -72,7 +72,7 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
         }
         return channel -> {
             RestCancellableNodeClient cancelClient = new RestCancellableNodeClient(client, request.getHttpChannel());
-            cancelClient.fieldCaps(fieldRequest, new RestChunkedToXContentListener<>(channel));
+            cancelClient.fieldCaps(fieldRequest, new RestRefCountedChunkedToXContentListener<>(channel));
         };
     }
 

server/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestClusterAllocationExplainAction.java

@@ -14,7 +14,7 @@
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.Scope;
 import org.elasticsearch.rest.ServerlessScope;
-import org.elasticsearch.rest.action.RestChunkedToXContentListener;
+import org.elasticsearch.rest.action.RestRefCountedChunkedToXContentListener;
 import org.elasticsearch.xcontent.XContentParser;
 
 import java.io.IOException;
@@ -58,6 +58,6 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
 
         req.includeYesDecisions(request.paramAsBoolean("include_yes_decisions", false));
         req.includeDiskInfo(request.paramAsBoolean("include_disk_info", false));
-        return channel -> client.admin().cluster().allocationExplain(req, new RestChunkedToXContentListener<>(channel));
+        return channel -> client.admin().cluster().allocationExplain(req, new RestRefCountedChunkedToXContentListener<>(channel));
     }
 }

server/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestClusterRerouteAction.java

@@ -19,7 +19,7 @@
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.Scope;
 import org.elasticsearch.rest.ServerlessScope;
-import org.elasticsearch.rest.action.RestChunkedToXContentListener;
+import org.elasticsearch.rest.action.RestRefCountedChunkedToXContentListener;
 import org.elasticsearch.xcontent.ObjectParser;
 import org.elasticsearch.xcontent.ObjectParser.ValueType;
 import org.elasticsearch.xcontent.ParseField;
@@ -84,7 +84,7 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
         if (metric == null) {
             request.params().put("metric", DEFAULT_METRICS);
         }
-        return channel -> client.admin().cluster().reroute(clusterRerouteRequest, new RestChunkedToXContentListener<>(channel));
+        return channel -> client.admin().cluster().reroute(clusterRerouteRequest, new RestRefCountedChunkedToXContentListener<>(channel));
     }
 
     @Override

server/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestGetDesiredBalanceAction.java

@@ -13,7 +13,7 @@
 import org.elasticsearch.client.internal.node.NodeClient;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.RestRequest;
-import org.elasticsearch.rest.action.RestChunkedToXContentListener;
+import org.elasticsearch.rest.action.RestRefCountedChunkedToXContentListener;
 
 import java.io.IOException;
 import java.util.List;
@@ -35,7 +35,7 @@ protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient cli
         return restChannel -> client.execute(
             TransportGetDesiredBalanceAction.TYPE,
             new DesiredBalanceRequest(),
-            new RestChunkedToXContentListener<>(restChannel)
+            new RestRefCountedChunkedToXContentListener<>(restChannel)
         );
     }
 }

server/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestGetSnapshotsAction.java

@@ -16,7 +16,7 @@
 import org.elasticsearch.rest.Scope;
 import org.elasticsearch.rest.ServerlessScope;
 import org.elasticsearch.rest.action.RestCancellableNodeClient;
-import org.elasticsearch.rest.action.RestChunkedToXContentListener;
+import org.elasticsearch.rest.action.RestRefCountedChunkedToXContentListener;
 import org.elasticsearch.search.sort.SortOrder;
 
 import java.io.IOException;
@@ -82,6 +82,6 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
         getSnapshotsRequest.masterNodeTimeout(request.paramAsTime("master_timeout", getSnapshotsRequest.masterNodeTimeout()));
         return channel -> new RestCancellableNodeClient(client, request.getHttpChannel()).admin()
             .cluster()
-            .getSnapshots(getSnapshotsRequest, new RestChunkedToXContentListener<>(channel));
+            .getSnapshots(getSnapshotsRequest, new RestRefCountedChunkedToXContentListener<>(channel));
     }
 }

server/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestNodesStatsAction.java

@@ -22,7 +22,7 @@
 import org.elasticsearch.rest.Scope;
 import org.elasticsearch.rest.ServerlessScope;
 import org.elasticsearch.rest.action.RestCancellableNodeClient;
-import org.elasticsearch.rest.action.RestChunkedToXContentListener;
+import org.elasticsearch.rest.action.RestRefCountedChunkedToXContentListener;
 
 import java.io.IOException;
 import java.util.Collections;
@@ -189,7 +189,7 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
 
         return channel -> new RestCancellableNodeClient(client, request.getHttpChannel()).admin()
             .cluster()
-            .nodesStats(nodesStatsRequest, new RestChunkedToXContentListener<>(channel));
+            .nodesStats(nodesStatsRequest, new RestRefCountedChunkedToXContentListener<>(channel));
     }
 
     private final Set<String> RESPONSE_PARAMS = Collections.singleton("level");

server/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestPendingClusterTasksAction.java

@@ -15,7 +15,7 @@
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.Scope;
 import org.elasticsearch.rest.ServerlessScope;
-import org.elasticsearch.rest.action.RestChunkedToXContentListener;
+import org.elasticsearch.rest.action.RestRefCountedChunkedToXContentListener;
 
 import java.io.IOException;
 import java.util.List;
@@ -43,7 +43,7 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
         return channel -> client.execute(
             TransportPendingClusterTasksAction.TYPE,
             pendingClusterTasksRequest,
-            new RestChunkedToXContentListener<>(channel)
+            new RestRefCountedChunkedToXContentListener<>(channel)
         );
     }
 }

server/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestSnapshotsStatusAction.java

@@ -16,7 +16,7 @@
 import org.elasticsearch.rest.Scope;
 import org.elasticsearch.rest.ServerlessScope;
 import org.elasticsearch.rest.action.RestCancellableNodeClient;
-import org.elasticsearch.rest.action.RestChunkedToXContentListener;
+import org.elasticsearch.rest.action.RestRefCountedChunkedToXContentListener;
 
 import java.io.IOException;
 import java.util.List;
@@ -56,6 +56,6 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
         snapshotsStatusRequest.masterNodeTimeout(request.paramAsTime("master_timeout", snapshotsStatusRequest.masterNodeTimeout()));
         return channel -> new RestCancellableNodeClient(client, request.getHttpChannel()).admin()
             .cluster()
-            .snapshotsStatus(snapshotsStatusRequest, new RestChunkedToXContentListener<>(channel));
+            .snapshotsStatus(snapshotsStatusRequest, new RestRefCountedChunkedToXContentListener<>(channel));
     }
 }

server/src/main/java/org/elasticsearch/rest/action/admin/indices/RestFieldUsageStatsAction.java

@@ -19,7 +19,7 @@
 import org.elasticsearch.rest.Scope;
 import org.elasticsearch.rest.ServerlessScope;
 import org.elasticsearch.rest.action.RestCancellableNodeClient;
-import org.elasticsearch.rest.action.RestChunkedToXContentListener;
+import org.elasticsearch.rest.action.RestRefCountedChunkedToXContentListener;
 
 import java.io.IOException;
 import java.util.List;
@@ -45,7 +45,7 @@ public BaseRestHandler.RestChannelConsumer prepareRequest(final RestRequest requ
         fusRequest.fields(request.paramAsStringArray("fields", fusRequest.fields()));
         return channel -> {
             final RestCancellableNodeClient cancelClient = new RestCancellableNodeClient(client, request.getHttpChannel());
-            cancelClient.execute(FieldUsageStatsAction.INSTANCE, fusRequest, new RestChunkedToXContentListener<>(channel));
+            cancelClient.execute(FieldUsageStatsAction.INSTANCE, fusRequest, new RestRefCountedChunkedToXContentListener<>(channel));
         };
     }
 }

server/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetIndicesAction.java

@@ -20,7 +20,7 @@
 import org.elasticsearch.rest.Scope;
 import org.elasticsearch.rest.ServerlessScope;
 import org.elasticsearch.rest.action.RestCancellableNodeClient;
-import org.elasticsearch.rest.action.RestChunkedToXContentListener;
+import org.elasticsearch.rest.action.RestRefCountedChunkedToXContentListener;
 
 import java.io.IOException;
 import java.util.List;
@@ -72,7 +72,7 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
         final var httpChannel = request.getHttpChannel();
         return channel -> new RestCancellableNodeClient(client, httpChannel).admin()
             .indices()
-            .getIndex(getIndexRequest, new RestChunkedToXContentListener<>(channel));
+            .getIndex(getIndexRequest, new RestRefCountedChunkedToXContentListener<>(channel));
     }
 
     /**

server/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetMappingAction.java

@@ -21,7 +21,7 @@
 import org.elasticsearch.rest.Scope;
 import org.elasticsearch.rest.ServerlessScope;
 import org.elasticsearch.rest.action.RestCancellableNodeClient;
-import org.elasticsearch.rest.action.RestChunkedToXContentListener;
+import org.elasticsearch.rest.action.RestRefCountedChunkedToXContentListener;
 
 import java.io.IOException;
 import java.util.List;
@@ -90,6 +90,6 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
         final HttpChannel httpChannel = request.getHttpChannel();
         return channel -> new RestCancellableNodeClient(client, httpChannel).admin()
             .indices()
-            .getMappings(getMappingsRequest, new RestChunkedToXContentListener<>(channel));
+            .getMappings(getMappingsRequest, new RestRefCountedChunkedToXContentListener<>(channel));
     }
 }

server/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetSettingsAction.java

@@ -17,7 +17,7 @@
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.Scope;
 import org.elasticsearch.rest.ServerlessScope;
-import org.elasticsearch.rest.action.RestChunkedToXContentListener;
+import org.elasticsearch.rest.action.RestRefCountedChunkedToXContentListener;
 
 import java.io.IOException;
 import java.util.List;
@@ -56,6 +56,6 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
             .names(names);
         getSettingsRequest.local(request.paramAsBoolean("local", getSettingsRequest.local()));
         getSettingsRequest.masterNodeTimeout(request.paramAsTime("master_timeout", getSettingsRequest.masterNodeTimeout()));
-        return channel -> client.admin().indices().getSettings(getSettingsRequest, new RestChunkedToXContentListener<>(channel));
+        return channel -> client.admin().indices().getSettings(getSettingsRequest, new RestRefCountedChunkedToXContentListener<>(channel));
     }
 }

server/src/main/java/org/elasticsearch/rest/action/admin/indices/RestIndicesSegmentsAction.java

@@ -19,7 +19,7 @@
 import org.elasticsearch.rest.Scope;
 import org.elasticsearch.rest.ServerlessScope;
 import org.elasticsearch.rest.action.RestCancellableNodeClient;
-import org.elasticsearch.rest.action.RestChunkedToXContentListener;
+import org.elasticsearch.rest.action.RestRefCountedChunkedToXContentListener;
 
 import java.io.IOException;
 import java.util.List;
@@ -58,6 +58,6 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
         indicesSegmentsRequest.indicesOptions(IndicesOptions.fromRequest(request, indicesSegmentsRequest.indicesOptions()));
         return channel -> new RestCancellableNodeClient(client, request.getHttpChannel()).admin()
             .indices()
-            .segments(indicesSegmentsRequest, new RestChunkedToXContentListener<>(channel));
+            .segments(indicesSegmentsRequest, new RestRefCountedChunkedToXContentListener<>(channel));
     }
 }

server/src/main/java/org/elasticsearch/rest/action/admin/indices/RestIndicesShardStoresAction.java

@@ -16,7 +16,7 @@
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.RestCancellableNodeClient;
-import org.elasticsearch.rest.action.RestChunkedToXContentListener;
+import org.elasticsearch.rest.action.RestRefCountedChunkedToXContentListener;
 
 import java.io.IOException;
 import java.util.List;
@@ -58,7 +58,7 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
         return channel -> new RestCancellableNodeClient(client, request.getHttpChannel()).execute(
             TransportIndicesShardStoresAction.TYPE,
             indicesShardStoresRequest,
-            new RestChunkedToXContentListener<>(channel)
+            new RestRefCountedChunkedToXContentListener<>(channel)
         );
     }
 }

server/src/main/java/org/elasticsearch/rest/action/admin/indices/RestIndicesStatsAction.java

@@ -22,7 +22,7 @@
 import org.elasticsearch.rest.Scope;
 import org.elasticsearch.rest.ServerlessScope;
 import org.elasticsearch.rest.action.RestCancellableNodeClient;
-import org.elasticsearch.rest.action.RestChunkedToXContentListener;
+import org.elasticsearch.rest.action.RestRefCountedChunkedToXContentListener;
 import org.elasticsearch.rest.action.document.RestMultiTermVectorsAction;
 
 import java.io.IOException;
@@ -146,7 +146,7 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
 
         return channel -> new RestCancellableNodeClient(client, request.getHttpChannel()).admin()
             .indices()
-            .stats(indicesStatsRequest, new RestChunkedToXContentListener<>(channel));
+            .stats(indicesStatsRequest, new RestRefCountedChunkedToXContentListener<>(channel));
     }
 
     @Override

server/src/main/java/org/elasticsearch/rest/action/admin/indices/RestRecoveryAction.java

@@ -17,7 +17,7 @@
 import org.elasticsearch.rest.Scope;
 import org.elasticsearch.rest.ServerlessScope;
 import org.elasticsearch.rest.action.RestCancellableNodeClient;
-import org.elasticsearch.rest.action.RestChunkedToXContentListener;
+import org.elasticsearch.rest.action.RestRefCountedChunkedToXContentListener;
 
 import java.io.IOException;
 import java.util.List;
@@ -54,6 +54,6 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
         recoveryRequest.indicesOptions(IndicesOptions.fromRequest(request, recoveryRequest.indicesOptions()));
         return channel -> new RestCancellableNodeClient(client, request.getHttpChannel()).admin()
             .indices()
-            .recoveries(recoveryRequest, new RestChunkedToXContentListener<>(channel));
+            .recoveries(recoveryRequest, new RestRefCountedChunkedToXContentListener<>(channel));
     }
 }

server/src/main/java/org/elasticsearch/rest/action/search/RestKnnSearchAction.java

@@ -14,7 +14,7 @@
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.RestCancellableNodeClient;
-import org.elasticsearch.rest.action.RestChunkedToXContentListener;
+import org.elasticsearch.rest.action.RestRefCountedChunkedToXContentListener;
 import org.elasticsearch.search.vectors.KnnSearchRequestParser;
 
 import java.io.IOException;
@@ -55,6 +55,6 @@ protected RestChannelConsumer prepareRequest(RestRequest restRequest, NodeClient
         SearchRequestBuilder searchRequestBuilder = cancellableNodeClient.prepareSearch();
         parser.toSearchRequest(searchRequestBuilder);
 
-        return channel -> searchRequestBuilder.execute(new RestChunkedToXContentListener<>(channel));
+        return channel -> searchRequestBuilder.execute(new RestRefCountedChunkedToXContentListener<>(channel));
     }
 }