Release top-level response in TransportNodesAction (#103845)

Similarly to #103254, we should `decRef` the top-level response after
completing the listener too.

Author: DaveCTurner

server/src/main/java/org/elasticsearch/action/support/nodes/TransportNodesAction.java

@@ -209,7 +209,7 @@ protected void newResponseAsync(
         List<FailedNodeException> failures,
         ActionListener<NodesResponse> listener
     ) {
-        ActionListener.completeWith(listener, () -> newResponse(request, responses, failures));
+        ActionListener.run(listener, l -> ActionListener.respondAndRelease(l, newResponse(request, responses, failures)));
     }
 
     protected abstract NodeRequest newNodeRequest(NodesRequest request);

server/src/test/java/org/elasticsearch/action/support/nodes/TransportNodesActionTests.java

@@ -37,6 +37,7 @@
 import org.elasticsearch.test.transport.CapturingTransport;
 import org.elasticsearch.threadpool.TestThreadPool;
 import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.LeakTracker;
 import org.elasticsearch.transport.TransportRequest;
 import org.elasticsearch.transport.TransportService;
 import org.hamcrest.Matchers;
@@ -125,6 +126,7 @@ public void testResponseAggregation() {
         final PlainActionFuture<TestNodesResponse> listener = new PlainActionFuture<>();
         action.execute(null, new TestNodesRequest(), listener.delegateFailure((l, response) -> {
             assertTrue(response.getNodes().stream().allMatch(TestNodeResponse::hasReferences));
+            assertTrue(response.hasReferences());
             l.onResponse(response);
         }));
         assertFalse(listener.isDone());
@@ -152,12 +154,14 @@ public void testResponseAggregation() {
 
         final var allResponsesReleasedListener = new SubscribableListener<Void>();
         try (var listeners = new RefCountingListener(allResponsesReleasedListener)) {
+            response.addCloseListener(listeners.acquire());
             for (final var nodeResponse : response.getNodes()) {
                 nodeResponse.addCloseListener(listeners.acquire());
             }
         }
         safeAwait(allResponsesReleasedListener);
         assertTrue(response.getNodes().stream().noneMatch(TestNodeResponse::hasReferences));
+        assertFalse(response.hasReferences());
 
         for (TestNodeResponse nodeResponse : response.getNodes()) {
             assertThat(successfulNodes, Matchers.hasItem(nodeResponse.getNode()));
@@ -407,6 +411,9 @@ private static class TestNodesRequest extends BaseNodesRequest<TestNodesRequest>
 
     private static class TestNodesResponse extends BaseNodesResponse<TestNodeResponse> {
 
+        private final SubscribableListener<Void> onClose = new SubscribableListener<>();
+        private final RefCounted refCounted = LeakTracker.wrap(AbstractRefCounted.of(() -> onClose.onResponse(null)));
+
         TestNodesResponse(ClusterName clusterName, List<TestNodeResponse> nodeResponses, List<FailedNodeException> failures) {
             super(clusterName, nodeResponses, failures);
         }
@@ -420,6 +427,30 @@ protected List<TestNodeResponse> readNodesFrom(StreamInput in) throws IOExceptio
         protected void writeNodesTo(StreamOutput out, List<TestNodeResponse> nodes) throws IOException {
             out.writeCollection(nodes);
         }
+
+        @Override
+        public void incRef() {
+            refCounted.incRef();
+        }
+
+        @Override
+        public boolean tryIncRef() {
+            return refCounted.tryIncRef();
+        }
+
+        @Override
+        public boolean decRef() {
+            return refCounted.decRef();
+        }
+
+        @Override
+        public boolean hasReferences() {
+            return refCounted.hasReferences();
+        }
+
+        void addCloseListener(ActionListener<Void> listener) {
+            onClose.addListener(listener);
+        }
     }
 
     private static class TestNodeRequest extends TransportRequest {
@@ -455,7 +486,7 @@ public boolean hasReferences() {
     private static class TestNodeResponse extends BaseNodeResponse {
 
         private final SubscribableListener<Void> onClose = new SubscribableListener<>();
-        private final RefCounted refCounted = AbstractRefCounted.of(() -> onClose.onResponse(null));
+        private final RefCounted refCounted = LeakTracker.wrap(AbstractRefCounted.of(() -> onClose.onResponse(null)));
 
         TestNodeResponse() {
             this(mock(DiscoveryNode.class));