Repository: eclipse-vertx/vert.x

Problem statement:
HAManager couldn't see quorum if during its initialization cluster manager has all nodes joined already
### Version

4.5.7

### Context

My app is based on Vert.x  with embedded Hazelcast cluster manager. When I run several instances of the app with HA mode enabled some instances cannot deploy HA verticles, and logs show that they have not attain quorum:  `Quorum not attained. Deployment of verticle will be delayed until there's a quorum.` 

But actually they have.

The problem is that Hazelcast cluster could be in a state when all nodes have already joined at the moment when `HAManager` initialization starts. So `HAManager`s method  `nodeAdded` will never be called. Also `checkQuorum` method called on `init` see all nodes from `clusterManager`, but `clusterMap` do not contain information from all nodes. They will put it in several milliseconds later of course, but `checkQuorum` method will be completed at that time and never be called again.

### Steps to reproduce

It just occurs in some cases.





Hints:

Patch:
diff --git a/vertx-core/src/main/java/io/vertx/core/impl/HAManager.java b/vertx-core/src/main/java/io/vertx/core/impl/HAManager.java
--- a/vertx-core/src/main/java/io/vertx/core/impl/HAManager.java
+++ b/vertx-core/src/main/java/io/vertx/core/impl/HAManager.java
@@ -373,6 +373,8 @@ private void checkQuorum() {
           if (group.equals(this.group)) {
             count++;
           }
+        } else if (!attainedQuorum) {
+          checkQuorumWhenAdded(node, System.currentTimeMillis());
         }
       }
       boolean attained = count >= quorumSize;


Test patch:
diff --git a/vertx-core/src/test/java/io/vertx/tests/ha/HAQuorumTest.java b/vertx-core/src/test/java/io/vertx/tests/ha/HAQuorumTest.java
new file mode 100644
--- /dev/null
+++ b/vertx-core/src/test/java/io/vertx/tests/ha/HAQuorumTest.java
@@ -0,0 +1,90 @@
+package io.vertx.tests.ha;
+
+import io.vertx.core.DeploymentOptions;
+import io.vertx.core.Vertx;
+import io.vertx.core.VertxOptions;
+import io.vertx.core.json.JsonObject;
+import io.vertx.core.spi.cluster.ClusterManager;
+import io.vertx.core.spi.cluster.NodeListener;
+import io.vertx.test.core.VertxTestBase;
+import io.vertx.test.fakecluster.FakeClusterManager;
+import org.junit.Test;
+
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+public final class HAQuorumTest extends VertxTestBase {
+  protected ClusterManager getClusterManager() {
+    return new FakeClusterManager() {
+      @Override
+      public void nodeListener(NodeListener listener) {
+        //do nothing
+      }
+    };
+  }
+
+  @Test
+  public void quorumIsObtainedOnNodeInfoPutThatDoneLaterThanClusterWasCreated() throws Exception {
+    //given
+    final Vertx vertx1 = startVertx(2);
+    final DeploymentOptions options = new DeploymentOptions().setHa(true);
+    final JsonObject config = new JsonObject().put("foo", "bar");
+    options.setConfig(config);
+
+    final Map<String, String> haInfoMap = getClusterManager().getSyncMap("__vertx.haInfo");
+    assertEquals(1, haInfoMap.size());
+    final Map.Entry<String, String> vertx1HaInfo = haInfoMap.entrySet().iterator().next();
+    haInfoMap.remove(vertx1HaInfo.getKey());
+
+    final Vertx vertx2 = startVertx(2);
+
+    vertx2.deployVerticle("java:" + HAVerticle2.class.getName(), options).onComplete(onSuccess(id -> {
+      assertTrue(vertx2.deploymentIDs().contains(id));
+      testComplete();
+    }));
+
+    assertWaitUntil(() -> vertx2.deploymentIDs().isEmpty());
+
+    //when
+    haInfoMap.put(vertx1HaInfo.getKey(), vertx1HaInfo.getValue());
+
+    // then
+    await();
+
+    closeVertices(vertx1, vertx2);
+  }
+
+  private Vertx startVertx(int quorumSize) throws Exception {
+    final VertxOptions options = new VertxOptions().setHAEnabled(true);
+    options.getEventBusOptions().setHost("localhost");
+    options.setQuorumSize(quorumSize);
+    final CountDownLatch latch = new CountDownLatch(1);
+    final AtomicReference<Vertx> vertxRef = new AtomicReference<>();
+    Vertx.builder()
+      .with(options)
+      .withClusterManager(getClusterManager())
+      .buildClustered()
+      .onComplete(onSuccess(vertx -> {
+        vertxRef.set(vertx);
+        latch.countDown();
+      }));
+    latch.await(2, TimeUnit.MINUTES);
+    return vertxRef.get();
+  }
+
+  private void closeVertices(Vertx... vertices) throws Exception {
+    CountDownLatch latch = new CountDownLatch(vertices.length);
+    for (Vertx vertex : vertices) {
+      if (vertex != null) {
+        vertex.close().onComplete(onSuccess(res -> {
+          latch.countDown();
+        }));
+      } else {
+        latch.countDown();
+      }
+    }
+    awaitLatch(latch, 2, TimeUnit.MINUTES);
+  }
+}