Avoid redundant connections in ClusterConnectionManager (#77196)

Today it's possible to open two connections to a node, and then we
notice when registering the second connection and close it instead.
Fixing #67873 will require us to keep tighter control over the identity
and lifecycle of each connection, and opening redundant connections gets
in the way of this. This commit adds a check for an existing connection
_after_ marking the connection as pending, which guarantees that we
don't open those redundant connections.

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

@@ -110,13 +110,25 @@ public void connectToNode(DiscoveryNode node, ConnectionProfile connectionProfil
 
         currentListener.addListener(listener);
 
+        // It's possible that a connection completed, and the pendingConnections entry was removed, between the calls to
+        // connectedNodes.containsKey and pendingConnections.putIfAbsent above, so we check again to make sure we don't open a redundant
+        // extra connection to the node. We could _just_ check here, but checking up front skips the work to mark the connection as pending.
+        if (connectedNodes.containsKey(node)) {
+            ListenableFuture<Void> future = pendingConnections.remove(node);
+            assert future == currentListener : "Listener in pending map is different than the expected listener";
+            connectingRefCounter.decRef();
+            future.onResponse(null);
+            return;
+        }
+
         final RunOnce releaseOnce = new RunOnce(connectingRefCounter::decRef);
         internalOpenConnection(node, resolvedProfile, ActionListener.wrap(conn -> {
             connectionValidator.validate(conn, resolvedProfile, ActionListener.wrap(
                 ignored -> {
                     assert Transports.assertNotTransportThread("connection validator success");
                     try {
                         if (connectedNodes.putIfAbsent(node, conn) != null) {
+                            assert false : "redundant conection to " + node;
                             logger.debug("existing connection to node [{}], closing new redundant connection", node);
                             IOUtils.closeWhileHandlingException(conn);
                         } else {