Reuse local node in async shard fetch responses

We read various objects from the wire that already exist in the cluster
state. The most notable is `DiscoveryNode` which can consume ~2kB in
heap for each fresh object, but rarely changes, so it's pretty wasteful
to use fresh objects here. There could be thousands (millions?) of
`DiscoveryNode` objects in flight from various `TransportNodesAction`
responses.

This branch adds a `DiscoveryNode` parameter to the response
deserialisation method and makes sure that the worst offenders re-use
the local object rather than creating a fresh one:

- `TransportNodesListShardStoreMetadata`
- `TransportNodesListGatewayStartedShards`

Relates #77266

Author: DaveCTurner

modules/ingest-geoip/src/main/java/org/elasticsearch/ingest/geoip/stats/GeoIpDownloaderStatsTransportAction.java

@@ -11,6 +11,7 @@
 import org.elasticsearch.action.FailedNodeException;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.nodes.TransportNodesAction;
+import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.io.stream.StreamInput;
@@ -57,7 +58,7 @@ protected NodeRequest newNodeRequest(Request request) {
     }
 
     @Override
-    protected NodeResponse newNodeResponse(StreamInput in) throws IOException {
+    protected NodeResponse newNodeResponse(StreamInput in, DiscoveryNode node) throws IOException {
         return new NodeResponse(in);
     }
 

server/src/main/java/org/elasticsearch/action/admin/cluster/node/hotthreads/TransportNodesHotThreadsAction.java

@@ -12,6 +12,7 @@
 import org.elasticsearch.action.FailedNodeException;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.nodes.TransportNodesAction;
+import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.io.stream.StreamInput;
@@ -49,7 +50,7 @@ protected NodeRequest newNodeRequest(NodesHotThreadsRequest request) {
     }
 
     @Override
-    protected NodeHotThreads newNodeResponse(StreamInput in) throws IOException {
+    protected NodeHotThreads newNodeResponse(StreamInput in, DiscoveryNode node) throws IOException {
         return new NodeHotThreads(in);
     }
 

server/src/main/java/org/elasticsearch/action/admin/cluster/node/info/TransportNodesInfoAction.java

@@ -11,6 +11,7 @@
 import org.elasticsearch.action.FailedNodeException;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.nodes.TransportNodesAction;
+import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.io.stream.StreamInput;
@@ -52,7 +53,7 @@ protected NodeInfoRequest newNodeRequest(NodesInfoRequest request) {
     }
 
     @Override
-    protected NodeInfo newNodeResponse(StreamInput in) throws IOException {
+    protected NodeInfo newNodeResponse(StreamInput in, DiscoveryNode node) throws IOException {
         return new NodeInfo(in);
     }
 

server/src/main/java/org/elasticsearch/action/admin/cluster/node/reload/TransportNodesReloadSecureSettingsAction.java

@@ -67,7 +67,7 @@ protected NodeRequest newNodeRequest(NodesReloadSecureSettingsRequest request) {
     }
 
     @Override
-    protected NodesReloadSecureSettingsResponse.NodeResponse newNodeResponse(StreamInput in) throws IOException {
+    protected NodesReloadSecureSettingsResponse.NodeResponse newNodeResponse(StreamInput in, DiscoveryNode node) throws IOException {
         return new NodesReloadSecureSettingsResponse.NodeResponse(in);
     }
 

server/src/main/java/org/elasticsearch/action/admin/cluster/node/stats/TransportNodesStatsAction.java

@@ -11,6 +11,7 @@
 import org.elasticsearch.action.FailedNodeException;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.nodes.TransportNodesAction;
+import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.io.stream.StreamInput;
@@ -54,7 +55,7 @@ protected NodeStatsRequest newNodeRequest(NodesStatsRequest request) {
     }
 
     @Override
-    protected NodeStats newNodeResponse(StreamInput in) throws IOException {
+    protected NodeStats newNodeResponse(StreamInput in, DiscoveryNode node) throws IOException {
         return new NodeStats(in);
     }
 

server/src/main/java/org/elasticsearch/action/admin/cluster/node/usage/TransportNodesUsageAction.java

@@ -11,6 +11,7 @@
 import org.elasticsearch.action.FailedNodeException;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.nodes.TransportNodesAction;
+import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.io.stream.StreamInput;
@@ -55,7 +56,7 @@ protected NodeUsageRequest newNodeRequest(NodesUsageRequest request) {
     }
 
     @Override
-    protected NodeUsage newNodeResponse(StreamInput in) throws IOException {
+    protected NodeUsage newNodeResponse(StreamInput in, DiscoveryNode node) throws IOException {
         return new NodeUsage(in);
     }
 

server/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/status/TransportNodesSnapshotsStatus.java

@@ -81,7 +81,7 @@ protected NodeRequest newNodeRequest(Request request) {
     }
 
     @Override
-    protected NodeSnapshotStatus newNodeResponse(StreamInput in) throws IOException {
+    protected NodeSnapshotStatus newNodeResponse(StreamInput in, DiscoveryNode node) throws IOException {
         return new NodeSnapshotStatus(in);
     }
 

server/src/main/java/org/elasticsearch/action/admin/cluster/stats/TransportClusterStatsAction.java

@@ -23,6 +23,7 @@
 import org.elasticsearch.cluster.health.ClusterHealthStatus;
 import org.elasticsearch.cluster.health.ClusterStateHealth;
 import org.elasticsearch.cluster.metadata.Metadata;
+import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.io.stream.StreamInput;
@@ -119,7 +120,7 @@ protected ClusterStatsNodeRequest newNodeRequest(ClusterStatsRequest request) {
     }
 
     @Override
-    protected ClusterStatsNodeResponse newNodeResponse(StreamInput in) throws IOException {
+    protected ClusterStatsNodeResponse newNodeResponse(StreamInput in, DiscoveryNode node) throws IOException {
         return new ClusterStatsNodeResponse(in);
     }
 

server/src/main/java/org/elasticsearch/action/admin/indices/dangling/find/TransportFindDanglingIndexAction.java

@@ -75,7 +75,7 @@ protected NodeFindDanglingIndexRequest newNodeRequest(FindDanglingIndexRequest r
     }
 
     @Override
-    protected NodeFindDanglingIndexResponse newNodeResponse(StreamInput in) throws IOException {
+    protected NodeFindDanglingIndexResponse newNodeResponse(StreamInput in, DiscoveryNode node) throws IOException {
         return new NodeFindDanglingIndexResponse(in);
     }
 

server/src/main/java/org/elasticsearch/action/admin/indices/dangling/list/TransportListDanglingIndicesAction.java

@@ -76,7 +76,7 @@ protected NodeListDanglingIndicesRequest newNodeRequest(ListDanglingIndicesReque
     }
 
     @Override
-    protected NodeListDanglingIndicesResponse newNodeResponse(StreamInput in) throws IOException {
+    protected NodeListDanglingIndicesResponse newNodeResponse(StreamInput in, DiscoveryNode node) throws IOException {
         return new NodeListDanglingIndicesResponse(in);
     }