Unset discovery nodes for every transport node actions request (#17682)

* Removed includeDiscoveryNodes from BaseNodesRequest

Signed-off-by: Manik Garg <[email protected]>

Author: gargmanik13

CHANGELOG.md

@@ -6,6 +6,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 ## [Unreleased 3.x]
 ### Added
 - Add multi-threaded writer support in pull-based ingestion ([#17912](https://github.com/opensearch-project/OpenSearch/pull/17912))
+- Unset discovery nodes for every transport node actions request ([#17682](https://github.com/opensearch-project/OpenSearch/pull/17682))
 - Implement parallel shard refresh behind cluster settings ([#17782](https://github.com/opensearch-project/OpenSearch/pull/17782))
 
 ### Changed

server/src/main/java/org/opensearch/action/admin/cluster/node/hotthreads/NodesHotThreadsRequest.java

@@ -70,7 +70,7 @@ public NodesHotThreadsRequest(StreamInput in) throws IOException {
      * threads for all nodes is used.
      */
     public NodesHotThreadsRequest(String... nodesIds) {
-        super(false, nodesIds);
+        super(nodesIds);
     }
 
     public int threads() {

server/src/main/java/org/opensearch/action/admin/cluster/node/info/NodesInfoRequest.java

@@ -72,7 +72,7 @@ public NodesInfoRequest(StreamInput in) throws IOException {
      * for all nodes will be returned.
      */
     public NodesInfoRequest(String... nodesIds) {
-        super(false, nodesIds);
+        super(nodesIds);
         defaultMetrics();
     }
 

server/src/main/java/org/opensearch/action/admin/cluster/node/reload/NodesReloadSecureSettingsRequest.java

@@ -60,7 +60,7 @@ public class NodesReloadSecureSettingsRequest extends BaseNodesRequest<NodesRelo
     private SecureString secureSettingsPassword;
 
     public NodesReloadSecureSettingsRequest() {
-        super(true, (String[]) null);
+        super((String[]) null);
     }
 
     public NodesReloadSecureSettingsRequest(StreamInput in) throws IOException {
@@ -84,7 +84,7 @@ public NodesReloadSecureSettingsRequest(StreamInput in) throws IOException {
      * nodes.
      */
     public NodesReloadSecureSettingsRequest(String... nodesIds) {
-        super(true, nodesIds);
+        super(nodesIds);
     }
 
     @Nullable

server/src/main/java/org/opensearch/action/admin/cluster/node/stats/NodesStatsRequest.java

@@ -58,7 +58,7 @@ public class NodesStatsRequest extends BaseNodesRequest<NodesStatsRequest> {
     private final Set<String> requestedMetrics = new HashSet<>();
 
     public NodesStatsRequest() {
-        super(false, (String[]) null);
+        super((String[]) null);
     }
 
     public NodesStatsRequest(StreamInput in) throws IOException {
@@ -74,7 +74,7 @@ public NodesStatsRequest(StreamInput in) throws IOException {
      * for all nodes will be returned.
      */
     public NodesStatsRequest(String... nodesIds) {
-        super(false, nodesIds);
+        super(nodesIds);
     }
 
     /**

server/src/main/java/org/opensearch/action/admin/cluster/node/usage/NodesUsageRequest.java

@@ -61,7 +61,7 @@ public NodesUsageRequest(StreamInput in) throws IOException {
      * passed, usage for all nodes will be returned.
      */
     public NodesUsageRequest(String... nodesIds) {
-        super(false, nodesIds);
+        super(nodesIds);
     }
 
     /**

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

@@ -161,7 +161,7 @@ public Request(StreamInput in) throws IOException {
         }
 
         public Request(String[] nodesIds) {
-            super(false, nodesIds);
+            super(nodesIds);
         }
 
         public Request snapshots(Snapshot[] snapshots) {

server/src/main/java/org/opensearch/action/admin/cluster/stats/ClusterStatsRequest.java

@@ -83,7 +83,7 @@ public ClusterStatsRequest(StreamInput in) throws IOException {
      * based on all nodes will be returned.
      */
     public ClusterStatsRequest(String... nodesIds) {
-        super(false, nodesIds);
+        super(nodesIds);
     }
 
     public boolean useAggregatedNodeLevelResponses() {

server/src/main/java/org/opensearch/action/admin/cluster/wlm/WlmStatsRequest.java

@@ -37,13 +37,13 @@ public WlmStatsRequest(StreamInput in) throws IOException {
      * for all nodes will be returned.
      */
     public WlmStatsRequest(String[] nodesIds, Set<String> workloadGroupIds, Boolean breach) {
-        super(false, nodesIds);
+        super(nodesIds);
         this.workloadGroupIds = workloadGroupIds;
         this.breach = breach;
     }
 
     public WlmStatsRequest() {
-        super(false, (String[]) null);
+        super((String[]) null);
         workloadGroupIds = new HashSet<>();
         this.breach = false;
     }

server/src/main/java/org/opensearch/action/admin/indices/dangling/find/FindDanglingIndexRequest.java

@@ -53,7 +53,7 @@ public FindDanglingIndexRequest(StreamInput in) throws IOException {
     }
 
     public FindDanglingIndexRequest(String indexUUID) {
-        super(false, Strings.EMPTY_ARRAY);
+        super(Strings.EMPTY_ARRAY);
         this.indexUUID = indexUUID;
     }
 

server/src/main/java/org/opensearch/action/admin/indices/dangling/list/ListDanglingIndicesRequest.java

@@ -58,12 +58,12 @@ public ListDanglingIndicesRequest(StreamInput in) throws IOException {
     }
 
     public ListDanglingIndicesRequest() {
-        super(false, Strings.EMPTY_ARRAY);
+        super(Strings.EMPTY_ARRAY);
         this.indexUUID = null;
     }
 
     public ListDanglingIndicesRequest(String indexUUID) {
-        super(false, Strings.EMPTY_ARRAY);
+        super(Strings.EMPTY_ARRAY);
         this.indexUUID = indexUUID;
     }
 

server/src/main/java/org/opensearch/action/search/GetAllPitNodesRequest.java

@@ -27,7 +27,7 @@ public class GetAllPitNodesRequest extends BaseNodesRequest<GetAllPitNodesReques
 
     @Inject
     public GetAllPitNodesRequest(DiscoveryNode... concreteNodes) {
-        super(false, concreteNodes);
+        super(concreteNodes);
     }
 
     public GetAllPitNodesRequest(StreamInput in) throws IOException {

server/src/main/java/org/opensearch/action/support/nodes/BaseNodesRequest.java

@@ -63,17 +63,12 @@ public abstract class BaseNodesRequest<Request extends BaseNodesRequest<Request>
     /**
      * once {@link #nodesIds} are resolved this will contain the concrete nodes that are part of this request. If set, {@link #nodesIds}
      * will be ignored and this will be used.
-     * */
+     * <p>
+     * Note: concreteNodes when accessed by transport actions will be null.
+     * See {@link TransportNodesAction.AsyncAction#AsyncAction}
+     **/
     private DiscoveryNode[] concreteNodes;
 
-    /**
-     * Since do not use the discovery nodes coming from the request in all code paths following a request extended off from
-     * BaseNodeRequest, we do not require it to sent around across all nodes.
-     *
-     * Setting default behavior as `true` but can be explicitly changed in requests that do not require.
-     */
-    private boolean includeDiscoveryNodes = true;
-
     private final TimeValue DEFAULT_TIMEOUT_SECS = TimeValue.timeValueSeconds(30);
 
     private TimeValue timeout;
@@ -89,22 +84,11 @@ protected BaseNodesRequest(String... nodesIds) {
         this.nodesIds = nodesIds;
     }
 
-    protected BaseNodesRequest(boolean includeDiscoveryNodes, String... nodesIds) {
-        this.nodesIds = nodesIds;
-        this.includeDiscoveryNodes = includeDiscoveryNodes;
-    }
-
     protected BaseNodesRequest(DiscoveryNode... concreteNodes) {
         this.nodesIds = null;
         this.concreteNodes = concreteNodes;
     }
 
-    protected BaseNodesRequest(boolean includeDiscoveryNodes, DiscoveryNode... concreteNodes) {
-        this.nodesIds = null;
-        this.concreteNodes = concreteNodes;
-        this.includeDiscoveryNodes = includeDiscoveryNodes;
-    }
-
     public final String[] nodesIds() {
         return nodesIds;
     }
@@ -139,10 +123,6 @@ public void setConcreteNodes(DiscoveryNode[] concreteNodes) {
         this.concreteNodes = concreteNodes;
     }
 
-    public boolean getIncludeDiscoveryNodes() {
-        return includeDiscoveryNodes;
-    }
-
     @Override
     public ActionRequestValidationException validate() {
         return null;

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

@@ -240,12 +240,10 @@ class AsyncAction {
             }
             this.responses = new AtomicReferenceArray<>(request.concreteNodes().length);
             this.concreteNodes = request.concreteNodes();
-            if (request.getIncludeDiscoveryNodes() == false) {
-                // As we transfer the ownership of discovery nodes to route the request to into the AsyncAction class,
-                // we remove the list of DiscoveryNodes from the request. This reduces the payload of the request and improves
-                // the number of concrete nodes in the memory.
-                request.setConcreteNodes(null);
-            }
+            // As we transfer the ownership of discovery nodes to route the request to into the AsyncAction class,
+            // we remove the list of DiscoveryNodes from the request. This reduces the payload of the request and improves
+            // the number of concrete nodes in the memory.
+            request.setConcreteNodes(null);
         }
 
         void start() {

server/src/main/java/org/opensearch/gateway/TransportNodesListGatewayMetaState.java

@@ -133,7 +133,7 @@ public Request(StreamInput in) throws IOException {
         }
 
         public Request(String... nodesIds) {
-            super(false, nodesIds);
+            super(nodesIds);
         }
     }
 

server/src/main/java/org/opensearch/gateway/TransportNodesListGatewayStartedShards.java

@@ -197,7 +197,7 @@ public Request(StreamInput in) throws IOException {
         }
 
         public Request(ShardId shardId, String customDataPath, DiscoveryNode[] nodes) {
-            super(false, nodes);
+            super(nodes);
             this.shardId = Objects.requireNonNull(shardId);
             this.customDataPath = Objects.requireNonNull(customDataPath);
         }

server/src/main/java/org/opensearch/gateway/TransportNodesListGatewayStartedShardsBatch.java

@@ -182,7 +182,7 @@ public Request(StreamInput in) throws IOException {
         }
 
         public Request(DiscoveryNode[] nodes, Map<ShardId, ShardAttributes> shardAttributes) {
-            super(false, nodes);
+            super(nodes);
             this.shardAttributes = Objects.requireNonNull(shardAttributes);
         }
 

server/src/main/java/org/opensearch/indices/store/TransportNodesListShardStoreMetadata.java

@@ -176,7 +176,7 @@ public Request(StreamInput in) throws IOException {
         }
 
         public Request(ShardId shardId, String customDataPath, DiscoveryNode[] nodes) {
-            super(false, nodes);
+            super(nodes);
             this.shardId = Objects.requireNonNull(shardId);
             this.customDataPath = Objects.requireNonNull(customDataPath);
         }

server/src/main/java/org/opensearch/indices/store/TransportNodesListShardStoreMetadataBatch.java

@@ -188,7 +188,7 @@ public Request(StreamInput in) throws IOException {
         }
 
         public Request(Map<ShardId, ShardAttributes> shardAttributes, DiscoveryNode[] nodes) {
-            super(false, nodes);
+            super(nodes);
             this.shardAttributes = Objects.requireNonNull(shardAttributes);
         }
 

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

@@ -168,25 +168,9 @@ public void testCustomResolving() throws Exception {
         assertEquals(clusterService.state().nodes().getDataNodes().size(), capturedRequests.size());
     }
 
-    public void testTransportNodesActionWithDiscoveryNodesIncluded() {
-        String[] nodeIds = clusterService.state().nodes().getNodes().keySet().toArray(new String[0]);
-        TestNodesRequest request = new TestNodesRequest(true, nodeIds);
-        getTestTransportNodesAction().new AsyncAction(null, request, new PlainActionFuture<>()).start();
-        Map<String, List<CapturingTransport.CapturedRequest>> capturedRequests = transport.getCapturedRequestsByTargetNodeAndClear();
-        List<TestNodeRequest> capturedTransportNodeRequestList = capturedRequests.values()
-            .stream()
-            .flatMap(Collection::stream)
-            .map(capturedRequest -> (TestNodeRequest) capturedRequest.request)
-            .collect(Collectors.toList());
-        assertEquals(nodeIds.length, capturedTransportNodeRequestList.size());
-        capturedTransportNodeRequestList.forEach(
-            capturedRequest -> assertEquals(nodeIds.length, capturedRequest.testNodesRequest.concreteNodes().length)
-        );
-    }
-
     public void testTransportNodesActionWithDiscoveryNodesReset() {
         String[] nodeIds = clusterService.state().nodes().getNodes().keySet().toArray(new String[0]);
-        TestNodesRequest request = new TestNodesRequest(false, nodeIds);
+        TestNodesRequest request = new TestNodesRequest(nodeIds);
         getTestTransportNodesAction().new AsyncAction(null, request, new PlainActionFuture<>()).start();
         Map<String, List<CapturingTransport.CapturedRequest>> capturedRequests = transport.getCapturedRequestsByTargetNodeAndClear();
         List<TestNodeRequest> capturedTransportNodeRequestList = capturedRequests.values()
@@ -389,10 +373,6 @@ private static class TestNodesRequest extends BaseNodesRequest<TestNodesRequest>
         TestNodesRequest(String... nodesIds) {
             super(nodesIds);
         }
-
-        TestNodesRequest(boolean includeDiscoveryNodes, String... nodesIds) {
-            super(includeDiscoveryNodes, nodesIds);
-        }
     }
 
     private static class TestNodesResponse extends BaseNodesResponse<TestNodeResponse> {