[test] Improve failure mode of NodeConnectionsServiceTests (#110956) (#115489)

Currently, when NodeConnectionServiceTests
#testOnlyBlocksOnConnectionsToNewNodes fails due to an exception, it
resets a CyclicBarrier which causes an AssertionError to be thrown on a
connector thread.
The AssertionError ends up being uncaught, so when we try to stop the
TransportService in the tear-down it blocks indefinitely waiting for
that connector thread to complete. This causes the original exception
to be obscured.
This change makes the connector thread fail more gracefully so that the
test can complete and the root cause will be revealed.

(cherry picked from commit 61bf77e)

Closes #115381

Author: nicktindall

server/src/test/java/org/elasticsearch/cluster/NodeConnectionsServiceTests.java

@@ -615,29 +615,25 @@ public TransportAddress[] addressesFromString(String address) {
             return new TransportAddress[0];
         }
 
-        private void runConnectionBlock(CheckedRunnable<Exception> connectionBlock) {
+        private void runConnectionBlock(CheckedRunnable<Exception> connectionBlock) throws Exception {
             if (connectionBlock == null) {
                 return;
             }
-            try {
-                connectionBlock.run();
-            } catch (Exception e) {
-                throw new AssertionError(e);
-            }
+            connectionBlock.run();
         }
 
         @Override
         public void openConnection(DiscoveryNode node, ConnectionProfile profile, ActionListener<Connection> listener) {
             final CheckedRunnable<Exception> connectionBlock = nodeConnectionBlocks.get(node);
             if (profile == null && randomConnectionExceptions && randomBoolean()) {
-                threadPool.generic().execute(() -> {
+                threadPool.generic().execute(() -> ActionListener.completeWith(listener, () -> {
                     runConnectionBlock(connectionBlock);
-                    listener.onFailure(new ConnectTransportException(node, "simulated"));
-                });
+                    throw new ConnectTransportException(node, "simulated");
+                }));
             } else {
-                threadPool.generic().execute(() -> {
+                threadPool.generic().execute(() -> ActionListener.completeWith(listener, () -> {
                     runConnectionBlock(connectionBlock);
-                    listener.onResponse(new Connection() {
+                    return new Connection() {
                         private final ListenableActionFuture<Void> closeListener = new ListenableActionFuture<>();
                         private final ListenableActionFuture<Void> removedListener = new ListenableActionFuture<>();
 
@@ -696,8 +692,8 @@ public boolean decRef() {
                         public boolean hasReferences() {
                             return refCounted.hasReferences();
                         }
-                    });
-                });
+                    };
+                }));
             }
         }