Avoid retaining ClusterState in lambda closures

Harish Narasimhan · Harish Narasimhan · commit 64a4b05b7aff · 2026-03-13T02:58:14.000Z
Extract version and cluster-manager node ID before creating lambda
closures to prevent retaining full ClusterState objects in memory.

- Add build(long, String) overload and private buildPredicate helper
  to ClusterManagerNodeChangePredicate
- Add StoredState(String, long) constructor and matching
  ClusterStateObserver constructors accepting pre-extracted values
- Update TransportClusterManagerNodeAction methods to capture only
  primitives and Strings instead of full ClusterState in closures

Signed-off-by: Harish Narasimhan &lt;hxarishk@amazon.com&gt;
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -53,6 +53,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Delegate getMin/getMax methods for ExitableTerms ([#20775](https://github.com/opensearch-project/OpenSearch/pull/20775))
 - Fix terms lookup subquery fetch limit reading from non-existent index setting instead of cluster `max_clause_count` ([#20823](https://github.com/opensearch-project/OpenSearch/pull/20823))
 - Fix array_index_out_of_bounds_exception with wildcard and aggregations ([#20842](https://github.com/opensearch-project/OpenSearch/pull/20842))
+- Reduce ClusterState retention in retry closures of TransportClusterManagerNodeAction ([#20858](https://github.com/opensearch-project/OpenSearch/pull/20858))
 
 ### Dependencies
 - Bump shadow-gradle-plugin from 8.3.9 to 9.3.1 ([#20569](https://github.com/opensearch-project/OpenSearch/pull/20569))
diff --git a/server/src/main/java/org/opensearch/action/support/clustermanager/TransportClusterManagerNodeAction.java b/server/src/main/java/org/opensearch/action/support/clustermanager/TransportClusterManagerNodeAction.java
@@ -266,7 +266,9 @@ protected void doStart(ClusterState clusterState) {
                 } else {
                     if (nodes.getClusterManagerNode() == null) {
                         logger.debug("no known cluster-manager node, scheduling a retry");
-                        retryOnMasterChange(clusterState, null);
+                        // Extract version before creating closure to avoid retaining full ClusterState
+                        final long stateVersion = clusterState.version();
+                        retryOnMasterChange(stateVersion, null, null, null);
                     } else {
                         DiscoveryNode clusterManagerNode = nodes.getClusterManagerNode();
                         if (clusterManagerNode.getVersion().onOrAfter(V_2_13_0) && localExecuteSupportedByAction()) {
@@ -285,11 +287,23 @@ protected void doStart(ClusterState clusterState) {
             }
         }
 
-        private void retryOnMasterChange(ClusterState state, Throwable failure) {
-            retry(state, failure, ClusterManagerNodeChangePredicate.build(state));
+        private void retryOnMasterChange(long stateVersion, String ephemeralNodeId, String persistentNodeId, Throwable failure) {
+            retry(
+                stateVersion,
+                ephemeralNodeId,
+                persistentNodeId,
+                failure,
+                ClusterManagerNodeChangePredicate.build(stateVersion, ephemeralNodeId)
+            );
         }
 
-        private void retry(ClusterState state, final Throwable failure, final Predicate<ClusterState> statePredicate) {
+        private void retry(
+            final long stateVersion,
+            final String ephemeralNodeId,
+            final String persistentNodeId,
+            final Throwable failure,
+            final Predicate<ClusterState> statePredicate
+        ) {
             if (observer == null) {
                 final long remainingTimeoutMS = request.clusterManagerNodeTimeout().millis() - (threadPool.relativeTimeInMillis()
                     - startTime);
@@ -299,7 +313,8 @@ private void retry(ClusterState state, final Throwable failure, final Predicate<
                     return;
                 }
                 this.observer = new ClusterStateObserver(
-                    state,
+                    persistentNodeId,
+                    stateVersion,
                     clusterService,
                     TimeValue.timeValueMillis(remainingTimeoutMS),
                     logger,
@@ -329,6 +344,12 @@ public void onTimeout(TimeValue timeout) {
         }
 
         private ActionListener<Response> getDelegateForLocalExecute(ClusterState clusterState) {
+            // Extract version and cluster manager node IDs before creating closure to avoid retaining full ClusterState
+            final long stateVersion = clusterState.version();
+            final DiscoveryNode clusterManagerNode = clusterState.nodes().getClusterManagerNode();
+            final String ephemeralNodeId = clusterManagerNode != null ? clusterManagerNode.getEphemeralId() : null;
+            final String persistentNodeId = clusterManagerNode != null ? clusterManagerNode.getId() : null;
+
             return ActionListener.delegateResponse(listener, (delegatedListener, t) -> {
                 if (t instanceof FailedToCommitClusterStateException || t instanceof NotClusterManagerException) {
                     logger.debug(
@@ -340,7 +361,7 @@ private ActionListener<Response> getDelegateForLocalExecute(ClusterState cluster
                         t
                     );
 
-                    retryOnMasterChange(clusterState, t);
+                    retryOnMasterChange(stateVersion, ephemeralNodeId, persistentNodeId, t);
                 } else {
                     delegatedListener.onFailure(t);
                 }
@@ -352,6 +373,12 @@ protected BiConsumer<DiscoveryNode, ClusterState> clusterStateLatestChecker(
             BiConsumer<DiscoveryNode, ClusterState> onStaleLocalState
         ) {
             return (clusterManagerNode, clusterState) -> {
+                // Extract version and cluster manager node IDs before creating closure to avoid retaining full ClusterState
+                final long stateVersion = clusterState.version();
+                final DiscoveryNode currentClusterManagerNode = clusterState.nodes().getClusterManagerNode();
+                final String currentEphemeralNodeId = currentClusterManagerNode != null ? currentClusterManagerNode.getEphemeralId() : null;
+                final String currentPersistentNodeId = currentClusterManagerNode != null ? currentClusterManagerNode.getId() : null;
+
                 transportService.sendRequest(
                     clusterManagerNode,
                     GetTermVersionAction.NAME,
@@ -377,7 +404,13 @@ public void handleResponse(GetTermVersionResponse response) {
 
                         @Override
                         public void handleException(TransportException exp) {
-                            handleTransportException(clusterManagerNode, clusterState, exp);
+                            handleTransportException(
+                                clusterManagerNode,
+                                stateVersion,
+                                currentEphemeralNodeId,
+                                currentPersistentNodeId,
+                                exp
+                            );
                         }
 
                         @Override
@@ -455,7 +488,12 @@ private boolean checkForBlock(Request request, ClusterState localClusterState) {
                     listener.onFailure(blockException);
                 } else {
                     logger.debug("can't execute due to a cluster block, retrying", blockException);
-                    retry(localClusterState, blockException, newState -> {
+                    // Avoid capturing full ClusterState in the lambda
+                    final long blockStateVersion = localClusterState.version();
+                    final DiscoveryNode blockClusterManagerNode = localClusterState.nodes().getClusterManagerNode();
+                    final String blockEphemeralNodeId = blockClusterManagerNode != null ? blockClusterManagerNode.getEphemeralId() : null;
+                    final String blockPersistentNodeId = blockClusterManagerNode != null ? blockClusterManagerNode.getId() : null;
+                    retry(blockStateVersion, blockEphemeralNodeId, blockPersistentNodeId, blockException, newState -> {
                         try {
                             ClusterBlockException newException = checkBlock(request, newState);
                             return (newException == null || !newException.retryable());
@@ -490,20 +528,32 @@ private void executeOnLocalNode(ClusterState localClusterState) {
         private void executeOnClusterManager(DiscoveryNode clusterManagerNode, ClusterState clusterState) {
             final String actionName = getClusterManagerActionName(clusterManagerNode);
 
+            // Extract version and cluster manager node IDs before creating closure to avoid retaining full ClusterState
+            final long stateVersion = clusterState.version();
+            final DiscoveryNode currentClusterManagerNode = clusterState.nodes().getClusterManagerNode();
+            final String currentEphemeralNodeId = currentClusterManagerNode != null ? currentClusterManagerNode.getEphemeralId() : null;
+            final String currentPersistentNodeId = currentClusterManagerNode != null ? currentClusterManagerNode.getId() : null;
+
             transportService.sendRequest(
                 clusterManagerNode,
                 actionName,
                 request,
                 new ActionListenerResponseHandler<Response>(listener, TransportClusterManagerNodeAction.this::read) {
                     @Override
                     public void handleException(final TransportException exp) {
-                        handleTransportException(clusterManagerNode, clusterState, exp);
+                        handleTransportException(clusterManagerNode, stateVersion, currentEphemeralNodeId, currentPersistentNodeId, exp);
                     }
                 }
             );
         }
 
-        private void handleTransportException(DiscoveryNode clusterManagerNode, ClusterState clusterState, final TransportException exp) {
+        private void handleTransportException(
+            DiscoveryNode clusterManagerNode,
+            long stateVersion,
+            String ephemeralNodeId,
+            String persistentNodeId,
+            final TransportException exp
+        ) {
             Throwable cause = exp.unwrapCause();
             if (cause instanceof ConnectTransportException
                 || (exp instanceof RemoteTransportException && cause instanceof NodeClosedException)) {
@@ -517,7 +567,7 @@ private void handleTransportException(DiscoveryNode clusterManagerNode, ClusterS
                     exp.getDetailedMessage()
                 );
 
-                retryOnMasterChange(clusterState, cause);
+                retryOnMasterChange(stateVersion, ephemeralNodeId, persistentNodeId, cause);
             } else {
                 listener.onFailure(exp);
             }
diff --git a/server/src/main/java/org/opensearch/cluster/ClusterManagerNodeChangePredicate.java b/server/src/main/java/org/opensearch/cluster/ClusterManagerNodeChangePredicate.java
@@ -48,13 +48,27 @@ private ClusterManagerNodeChangePredicate() {
     }
 
     /**
-     * builds a predicate that will accept a cluster state only if it was generated after the current has
-     * (re-)joined the master
+     * Builds a predicate that accepts a cluster state only if it was generated after the current node
+     * (re-)joined the cluster manager. Extracts version and cluster-manager ID from the given state
+     * and delegates to {@link #build(long, String)}.
      */
     public static Predicate<ClusterState> build(ClusterState currentState) {
         final long currentVersion = currentState.version();
         final DiscoveryNode clusterManagerNode = currentState.nodes().getClusterManagerNode();
         final String currentMasterId = clusterManagerNode == null ? null : clusterManagerNode.getEphemeralId();
+        return build(currentVersion, currentMasterId);
+    }
+
+    /**
+     * Builds a predicate that accepts a cluster state only if the cluster manager has changed
+     * or the state version has increased beyond the provided version. Accepts pre-extracted
+     * values to avoid retaining the full {@link ClusterState} in lambda closures.
+     *
+     * @param currentVersion  the cluster state version to compare against
+     * @param currentMasterId the ephemeral ID of the current cluster manager node, or null if none
+     * @return predicate that returns true when cluster manager changes or version increases
+     */
+    public static Predicate<ClusterState> build(long currentVersion, String currentMasterId) {
         return newState -> {
             final DiscoveryNode newClusterManager = newState.nodes().getClusterManagerNode();
             final boolean accept;
diff --git a/server/src/main/java/org/opensearch/cluster/ClusterStateObserver.java b/server/src/main/java/org/opensearch/cluster/ClusterStateObserver.java
@@ -117,6 +117,36 @@ public ClusterStateObserver(
         this.contextHolder = contextHolder;
     }
 
+    public ClusterStateObserver(
+        String clusterManagerNodeId,
+        long version,
+        ClusterService clusterService,
+        @Nullable TimeValue timeout,
+        Logger logger,
+        ThreadContext contextHolder
+    ) {
+        this(clusterManagerNodeId, version, clusterService.getClusterApplierService(), timeout, logger, contextHolder);
+    }
+
+    public ClusterStateObserver(
+        String clusterManagerNodeId,
+        long version,
+        ClusterApplierService clusterApplierService,
+        @Nullable TimeValue timeout,
+        Logger logger,
+        ThreadContext contextHolder
+    ) {
+        this.clusterApplierService = clusterApplierService;
+        this.threadPool = clusterApplierService.threadPool();
+        this.lastObservedState = new AtomicReference<>(new StoredState(clusterManagerNodeId, version));
+        this.timeOutValue = timeout;
+        if (timeOutValue != null) {
+            this.startTimeMS = threadPool.relativeTimeInMillis();
+        }
+        this.logger = logger;
+        this.contextHolder = contextHolder;
+    }
+
     /** sets the last observed state to the currently applied cluster state and returns it */
     public ClusterState setAndGetObservedState() {
         if (observingContext.get() != null) {
@@ -311,8 +341,12 @@ private static class StoredState {
         private final long version;
 
         StoredState(ClusterState clusterState) {
-            this.clusterManagerNodeId = clusterState.nodes().getClusterManagerNodeId();
-            this.version = clusterState.version();
+            this(clusterState.nodes().getClusterManagerNodeId(), clusterState.version());
+        }
+
+        StoredState(String clusterManagerNodeId, long version) {
+            this.clusterManagerNodeId = clusterManagerNodeId;
+            this.version = version;
         }
 
         /**
diff --git a/server/src/test/java/org/opensearch/action/support/clustermanager/TransportClusterManagerNodeActionTests.java b/server/src/test/java/org/opensearch/action/support/clustermanager/TransportClusterManagerNodeActionTests.java
@@ -610,6 +610,94 @@ protected void clusterManagerOperation(Request request, ClusterState state, Acti
         assertThat(listener.get(), equalTo(response));
     }
 
+    /**
+     * Tests that when the local master steps down and no new master is elected yet,
+     * the retry mechanism waits via the observer (early predicate check returns false
+     * because there's no master), and then correctly forwards the request once a new
+     * master is elected.
+     */
+    public void testStepDownToNoMasterThenNewMasterElected() throws ExecutionException, InterruptedException {
+        Request request = new Request().clusterManagerNodeTimeout(TimeValue.timeValueHours(1));
+        PlainActionFuture<Response> listener = new PlainActionFuture<>();
+
+        final Response response = new Response();
+
+        setState(clusterService, ClusterStateCreationUtils.state(localNode, localNode, allNodes));
+
+        new Action("internal:testAction", transportService, clusterService, threadPool) {
+            @Override
+            protected void clusterManagerOperation(Request request, ClusterState state, ActionListener<Response> listener)
+                throws Exception {
+                // Master steps down but no new master elected yet (master = null)
+                setState(clusterService, ClusterStateCreationUtils.state(localNode, null, allNodes));
+                listener.onFailure(new NotClusterManagerException("Fake error"));
+            }
+        }.execute(request, listener);
+
+        // No transport request yet, waiting for a master to appear
+        assertThat(transport.capturedRequests().length, equalTo(0));
+        assertFalse(listener.isDone());
+
+        // Now a new master is elected (remoteNode)
+        setState(clusterService, ClusterStateCreationUtils.state(localNode, remoteNode, allNodes));
+
+        // The request should now be forwarded to the new master
+        assertThat(transport.capturedRequests().length, equalTo(1));
+        CapturingTransport.CapturedRequest capturedRequest = transport.capturedRequests()[0];
+        assertTrue(capturedRequest.node.isClusterManagerNode());
+        assertThat(capturedRequest.request, equalTo(request));
+        assertThat(capturedRequest.action, equalTo("internal:testAction"));
+
+        transport.handleResponse(capturedRequest.requestId, response);
+        assertTrue(listener.isDone());
+        assertThat(listener.get(), equalTo(response));
+    }
+
+    /**
+     * Tests that when the local master steps down with no new master, and then the
+     * SAME node becomes master again (with a higher version), the retry correctly
+     * re-executes locally.
+     */
+    public void testStepDownToNoMasterThenSameNodeReelected() throws ExecutionException, InterruptedException {
+        Request request = new Request().clusterManagerNodeTimeout(TimeValue.timeValueHours(1));
+        PlainActionFuture<Response> listener = new PlainActionFuture<>();
+
+        final Response response = new Response();
+        final int[] callCount = { 0 };
+
+        // Set initial state with a higher base version so re-election produces a version
+        // that the predicate recognizes as newer than the original
+        setState(clusterService, ClusterState.builder(ClusterStateCreationUtils.state(localNode, localNode, allNodes)).version(10));
+
+        new Action("internal:testAction", transportService, clusterService, threadPool) {
+            @Override
+            protected void clusterManagerOperation(Request request, ClusterState state, ActionListener<Response> listener)
+                throws Exception {
+                callCount[0]++;
+                if (callCount[0] == 1) {
+                    // First call: master steps down, no new master yet
+                    setState(clusterService, ClusterStateCreationUtils.state(localNode, null, allNodes));
+                    listener.onFailure(new FailedToCommitClusterStateException("Fake error"));
+                } else {
+                    // Second call: we're master again, succeed
+                    listener.onResponse(response);
+                }
+            }
+        }.execute(request, listener);
+
+        // Waiting for a master, no transport request, not done yet
+        assertFalse(listener.isDone());
+        assertThat(transport.capturedRequests().length, equalTo(0));
+
+        // Same node becomes master again with higher version
+        setState(clusterService, ClusterState.builder(ClusterStateCreationUtils.state(localNode, localNode, allNodes)).version(11));
+
+        // Should have re-executed locally and succeeded
+        assertTrue(listener.isDone());
+        assertThat(listener.get(), equalTo(response));
+        assertThat(callCount[0], equalTo(2));
+    }
+
     // Validate TransportMasterNodeAction.testDelegateToClusterManager() works correctly on node with the deprecated MASTER_ROLE.
     public void testDelegateToClusterManagerOnNodeWithDeprecatedMasterRole() throws ExecutionException, InterruptedException {
         DiscoveryNode localNode = new DiscoveryNode(
diff --git a/server/src/test/java/org/opensearch/cluster/ClusterManagerNodeChangePredicateTests.java b/server/src/test/java/org/opensearch/cluster/ClusterManagerNodeChangePredicateTests.java
