Move all logic to transportService/ClusterConnectionManager as per comments

Signed-off-by: Rahul Karajgikar <karajgik@amazon.com>

Author: Rahul Karajgikar

server/src/main/java/org/opensearch/cluster/NodeConnectionsService.java

@@ -172,7 +172,6 @@ public void disconnectFromNodesExcept(DiscoveryNodes discoveryNodes) {
             }
 
             for (final DiscoveryNode discoveryNode : nodesToDisconnect) {
-                logger.info("NodeConnectionsService - disconnecting from node [{}] in loop", discoveryNode);
                 runnables.add(targetsByNode.get(discoveryNode).disconnect());
             }
         }
@@ -390,13 +389,8 @@ public String toString() {
             protected void doRun() {
                 assert Thread.holdsLock(mutex) == false : "mutex unexpectedly held";
                 transportService.disconnectFromNode(discoveryNode);
-                transportService.markPendingLeftAsCompleted(discoveryNode);
                 consecutiveFailureCount.set(0);
-                logger.debug(
-                    "disconnected from {} and marked pending left as completed. " + "pending lefts: [{}]",
-                    discoveryNode,
-                    transportService.getNodesLeftInProgress()
-                );
+                logger.debug("disconnected from {}", discoveryNode);
                 onCompletion(ActivityType.DISCONNECTING, null, connectActivity);
             }
 

server/src/main/java/org/opensearch/cluster/coordination/Coordinator.java

@@ -630,16 +630,6 @@ private void handleJoinRequest(JoinRequest joinRequest, JoinHelper.JoinCallback
             );
             return;
         }
-        // if node-left is still in progress, we fail the joinRequest early
-        if (transportService.getNodesLeftInProgress().contains(joinRequest.getSourceNode())) {
-            joinCallback.onFailure(
-                new IllegalStateException(
-                    "cannot join node [" + joinRequest.getSourceNode() + "] because node-left is currently in progress for this node"
-
-                )
-            );
-            return;
-        }
 
         transportService.connectToNode(joinRequest.getSourceNode(), ActionListener.wrap(ignore -> {
             final ClusterState stateForJoinValidation = getStateForClusterManagerService();
@@ -1369,13 +1359,13 @@ assert getLocalNode().equals(clusterState.getNodes().get(getLocalNode().getId())
                 currentPublication = Optional.of(publication);
 
                 final DiscoveryNodes publishNodes = publishRequest.getAcceptedState().nodes();
+                // marking pending disconnects before publish
+                // if we try to joinRequest during pending disconnect, it should fail
+                transportService.markPendingDisconnects(clusterChangedEvent.nodesDelta());
                 leaderChecker.setCurrentNodes(publishNodes);
                 followersChecker.setCurrentNodes(publishNodes);
                 lagDetector.setTrackedNodes(publishNodes);
                 coordinationState.get().handlePrePublish(clusterState);
-                // trying to mark pending disconnects before publish
-                // if we try to joinRequest during pending disconnect, it should fail
-                transportService.markPendingConnections(clusterChangedEvent.nodesDelta());
                 publication.start(followersChecker.getFaultyNodes());
             }
         } catch (Exception e) {

server/src/main/java/org/opensearch/cluster/coordination/FollowersChecker.java

@@ -161,7 +161,6 @@ public FollowersChecker(
         transportService.addConnectionListener(new TransportConnectionListener() {
             @Override
             public void onNodeDisconnected(DiscoveryNode node, Transport.Connection connection) {
-                logger.info("in transport listener onNodeDisconnected");
                 handleDisconnectedNode(node);
             }
         });
@@ -391,7 +390,6 @@ public void handleException(TransportException exp) {
                         failureCountSinceLastSuccess++;
 
                         final String reason;
-
                         if (exp instanceof ConnectTransportException || exp.getCause() instanceof ConnectTransportException) {
                             logger.info(() -> new ParameterizedMessage("{} disconnected", FollowerChecker.this), exp);
                             reason = "disconnected";

server/src/main/java/org/opensearch/cluster/coordination/LeaderChecker.java

@@ -231,7 +231,7 @@ private void handleLeaderCheck(LeaderCheckRequest request) {
                 "rejecting leader check since [" + request.getSender() + "] has been removed from the cluster"
             );
         } else {
-            logger.debug("handling {}", request);
+            logger.trace("handling {}", request);
         }
     }
 

server/src/main/java/org/opensearch/cluster/service/ClusterApplierService.java

@@ -580,6 +580,7 @@ private void applyChanges(UpdateTask task, ClusterState previousClusterState, Cl
 
         logger.debug("apply cluster state with version {}", newClusterState.version());
         callClusterStateAppliers(clusterChangedEvent, stopWatch);
+        logger.debug("completed calling appliers of cluster state for version {}", newClusterState.version());
 
         nodeConnectionsService.disconnectFromNodesExcept(newClusterState.nodes());
 
@@ -592,11 +593,11 @@ private void applyChanges(UpdateTask task, ClusterState previousClusterState, Cl
                 + " on "
                 + newClusterState.nodes().getLocalNode();
 
-        logger.info("set locally applied cluster state to version {}", newClusterState.version());
+        logger.debug("set locally applied cluster state to version {}", newClusterState.version());
         state.set(newClusterState);
 
         callClusterStateListeners(clusterChangedEvent, stopWatch);
-        logger.info("completed appliers and listeners of cluster state for version {}", newClusterState.version());
+        logger.debug("completed calling listeners of cluster state for version {}", newClusterState.version());
     }
 
     protected void connectToNodesAndWait(ClusterState newClusterState) {

server/src/main/java/org/opensearch/transport/ClusterConnectionManager.java

@@ -65,7 +65,13 @@ public class ClusterConnectionManager implements ConnectionManager {
 
     private final ConcurrentMap<DiscoveryNode, Transport.Connection> connectedNodes = ConcurrentCollections.newConcurrentMap();
     private final ConcurrentMap<DiscoveryNode, ListenableFuture<Void>> pendingConnections = ConcurrentCollections.newConcurrentMap();
-    private final Set<DiscoveryNode> pendingLeft = ConcurrentCollections.newConcurrentSet();
+    /**
+     This set is used only by cluster-manager nodes.
+     Nodes are marked as pending disconnect right before cluster state publish phase.
+     They are cleared up as part of cluster state apply commit phase
+     This is to avoid connections from being made to nodes that are in the process of leaving the cluster
+     */
+    private final Set<DiscoveryNode> pendingDisconnections = ConcurrentCollections.newConcurrentSet();
     private final AbstractRefCounted connectingRefCounter = new AbstractRefCounted("connection manager") {
         @Override
         protected void closeInternal() {
@@ -113,19 +119,9 @@ public void openConnection(DiscoveryNode node, ConnectionProfile connectionProfi
     }
 
     @Override
-    public Set<DiscoveryNode> getNodesLeftInProgress() {
-        return this.pendingLeft;
-    }
-
-    @Override
-    public void markPendingLefts(List<DiscoveryNode> nodes) {
-        logger.info("marking pending left for nodes: [{}]", nodes);
-        pendingLeft.addAll(nodes);
-    }
-
-    @Override
-    public boolean markPendingLeftCompleted(DiscoveryNode discoveryNode) {
-        return pendingLeft.remove(discoveryNode);
+    public void markPendingDisconnects(List<DiscoveryNode> nodes) {
+        logger.info("marking pending disconnects for nodes: [{}]", nodes);
+        pendingDisconnections.addAll(nodes);
     }
 
     /**
@@ -147,6 +143,16 @@ public void connectToNode(
             return;
         }
 
+        // if node-left is still in progress, we fail the connect request early
+        if (pendingDisconnections.contains(node)) {
+            listener.onFailure(
+                new IllegalStateException(
+                    "blocked connection to node [" + node + "] because node-left is currently in progress for this node"
+                )
+            );
+            return;
+        }
+
         if (connectingRefCounter.tryIncRef() == false) {
             listener.onFailure(new IllegalStateException("connection manager is closed"));
             return;
@@ -246,6 +252,8 @@ public void disconnectFromNode(DiscoveryNode node) {
             // if we found it and removed it we close
             nodeChannels.close();
         }
+        pendingDisconnections.remove(node);
+        logger.info("Removed node {} from pending disconnects list", node);
     }
 
     /**

server/src/main/java/org/opensearch/transport/ConnectionManager.java

@@ -64,11 +64,7 @@ void connectToNode(
 
     boolean nodeConnected(DiscoveryNode node);
 
-    Set<DiscoveryNode> getNodesLeftInProgress();
-
-    boolean markPendingLeftCompleted(DiscoveryNode node);
-
-    void markPendingLefts(List<DiscoveryNode> nodes);
+    void markPendingDisconnects(List<DiscoveryNode> nodes);
 
     void disconnectFromNode(DiscoveryNode node);
 

server/src/main/java/org/opensearch/transport/RemoteConnectionManager.java

@@ -120,18 +120,8 @@ public ConnectionProfile getConnectionProfile() {
     }
 
     @Override
-    public Set<DiscoveryNode> getNodesLeftInProgress() {
-        throw new UnsupportedOperationException("not implemented");
-    }
-
-    @Override
-    public boolean markPendingLeftCompleted(DiscoveryNode node) {
-        throw new UnsupportedOperationException("not implemented");
-    }
-
-    @Override
-    public void markPendingLefts(List<DiscoveryNode> nodes) {
-        throw new UnsupportedOperationException("not implemented");
+    public void markPendingDisconnects(List<DiscoveryNode> nodes) {
+        delegate.markPendingDisconnects(nodes);
     }
 
     public Transport.Connection getAnyRemoteConnection() {

server/src/main/java/org/opensearch/transport/TransportService.java

@@ -483,18 +483,9 @@ public void connectToNode(DiscoveryNode node, ActionListener<Void> listener) thr
         connectToNode(node, null, listener);
     }
 
-    public Set<DiscoveryNode> getNodesLeftInProgress() {
-        return connectionManager.getNodesLeftInProgress();
-    }
-
-    // this
-    public void markPendingConnections(DiscoveryNodes.Delta nodesDelta) {
-        // connectionManager.markPendingJoins(nodesDelta.addedNodes());
-        connectionManager.markPendingLefts(nodesDelta.removedNodes());
-    }
 
-    public boolean markPendingLeftAsCompleted(DiscoveryNode node) {
-        return connectionManager.markPendingLeftCompleted(node);
+    public void markPendingDisconnects(DiscoveryNodes.Delta nodesDelta) {
+        connectionManager.markPendingDisconnects(nodesDelta.removedNodes());
     }
 
     public void connectToExtensionNode(DiscoveryNode node, ActionListener<Void> listener) throws ConnectTransportException {

test/framework/src/main/java/org/opensearch/test/transport/StubbableConnectionManager.java

@@ -120,18 +120,8 @@ public void connectToNode(
     }
 
     @Override
-    public Set<DiscoveryNode> getNodesLeftInProgress() {
-        return delegate.getNodesLeftInProgress();
-    }
-
-    @Override
-    public boolean markPendingLeftCompleted(DiscoveryNode node) {
-        return delegate.markPendingLeftCompleted(node);
-    }
-
-    @Override
-    public void markPendingLefts(List<DiscoveryNode> nodes) {
-        delegate.markPendingLefts(nodes);
+    public void markPendingDisconnects(List<DiscoveryNode> nodes) {
+        delegate.markPendingDisconnects(nodes);
     }
 
     @Override