Fix EventBus cluster manager node lookup failure.

Motivation:

When the EventBus allocated a connection for a given node, it reserves a slot in a map to cumulate messages until the connection is established. When connection fails, the allocated connection slot should be removed to let further reconnect attempts work.

Author: vietj

vertx-core/src/main/java/io/vertx/core/eventbus/impl/clustered/ClusteredEventBus.java

@@ -385,7 +385,10 @@ private void connect(OutboundConnection conn) {
           });
           conn.connected(connection);
         } else {
-          log.warn("Connecting to server " + conn.remoteNodeId() + " failed", ar.cause());
+          if (log.isWarnEnabled()) {
+            log.warn("Connecting to server " + conn.remoteNodeId() + " failed", ar.cause());
+          }
+          outboundConnections.remove(conn.remoteNodeId(), conn);
           conn.handleClose(ar.cause());
         }
       });

vertx-core/src/main/java/io/vertx/core/eventbus/impl/clustered/OutboundConnection.java

@@ -21,6 +21,7 @@
 import io.vertx.core.internal.VertxInternal;
 import io.vertx.core.internal.logging.Logger;
 import io.vertx.core.internal.logging.LoggerFactory;
+import io.vertx.core.internal.net.NetSocketInternal;
 import io.vertx.core.net.NetSocket;
 import io.vertx.core.spi.metrics.EventBusMetrics;
 
@@ -49,6 +50,7 @@ final class OutboundConnection implements Handler<Buffer> {
   private boolean connected;
   private long pingReplyTimeoutID = -1;
   private long pingTimeoutID = -1;
+  private boolean closed;
 
   OutboundConnection(ClusteredEventBus eventBus, String remoteNodeId) {
     this.eventBus = eventBus;
@@ -62,17 +64,28 @@ String remoteNodeId() {
   }
 
   synchronized void writeMessage(MessageImpl<?, ?> message, Promise<Void> writePromise) {
-    if (connected) {
+    Throwable failure;
+    synchronized (this) {
+      if (closed) {
+        failure = NetSocketInternal.CLOSED_EXCEPTION;
+      } else if (connected) {
+        failure = null;
+      } else {
+        if (pendingWrites == null) {
+          if (log.isDebugEnabled()) {
+            log.debug("Not connected to server " + remoteNodeId + " - starting queuing");
+          }
+          pendingWrites = new ArrayDeque<>();
+        }
+        pendingWrites.add(new MessageWrite(message, writePromise));
+        return;
+      }
+    }
+    if (failure == null) {
       writeMessage(message)
         .onComplete(writePromise);
     } else {
-      if (pendingWrites == null) {
-        if (log.isDebugEnabled()) {
-          log.debug("Not connected to server " + remoteNodeId + " - starting queuing");
-        }
-        pendingWrites = new ArrayDeque<>();
-      }
-      pendingWrites.add(new MessageWrite(message, writePromise));
+      writePromise.tryFail(failure);
     }
   }
 
@@ -91,6 +104,7 @@ void handleClose(Throwable cause) {
       vertx.cancelTimer(pingTimeoutID);
     }
     synchronized (this) {
+      closed = true;
       MessageWrite msg;
       if (pendingWrites != null) {
         while ((msg = pendingWrites.poll()) != null) {

vertx-core/src/test/java/io/vertx/tests/eventbus/ClusteredEventBusTest.java

@@ -752,4 +752,32 @@ public void disconnected(Object socketMetric, SocketAddress remoteAddress) {
     Future.future(eventBus::close).await();
     assertWaitUntil(() -> numberOfOutboundConnections.get() == 0 && numberOfInboundConnections.get() == 0);
   }
+
+  @Test
+  public void testHandleCloseRemovesStaleOutboundConnectionOnConnectFailure() {
+    AtomicInteger idx = new AtomicInteger();
+    startNodes(2, () -> new WrappedClusterManager(getClusterManager()) {
+      @Override
+      public void getNodeInfo(String nodeId, Completable<io.vertx.core.spi.cluster.NodeInfo> promise) {
+        if (idx.getAndIncrement() == 0) {
+          promise.fail("induced failure");
+        } else {
+          super.getNodeInfo(nodeId, promise);
+        }
+      }
+    });
+
+    vertices[1].eventBus().consumer(ADDRESS1, msg -> {
+      testComplete();
+    }).completion().await();
+
+    try {
+      vertices[0].eventBus().sender(ADDRESS1).write("will fail").await();
+      fail("Should have failed");
+    } catch (Exception e) {
+      vertices[0].eventBus().request(ADDRESS1, "will succeed");
+    }
+
+    await();
+  }
 }