Event bus consumer racy cluster registration - fixes #2899

eclipse-vertx · Apr 8, 2019 · 6d33b93 · 6d33b93
1 parent ed9db2e
commit 6d33b93
Show file tree

Hide file tree

Showing 3 changed files with 55 additions and 16 deletions.
diff --git a/src/main/java/io/vertx/core/eventbus/impl/HandlerRegistration.java b/src/main/java/io/vertx/core/eventbus/impl/HandlerRegistration.java
@@ -131,15 +131,14 @@ public void unregister(Handler<AsyncResult<Void>> completionHandler) {
     doUnregister(completionHandler);
   }
 
-  private void doUnregister(Handler<AsyncResult<Void>> completionHandler) {
+  private void doUnregister(Handler<AsyncResult<Void>> doneHandler) {
     Deque<Message<T>> discarded;
     Handler<Message<T>> discardHandler;
     synchronized (this) {
-      handler = null;
       if (endHandler != null) {
         Handler<Void> theEndHandler = endHandler;
-        Handler<AsyncResult<Void>> handler = completionHandler;
+        Handler<AsyncResult<Void>> handler = doneHandler;
-        completionHandler = ar -> {
+        doneHandler = ar -> {
           theEndHandler.handle(null);
           if (handler != null) {
             handler.handle(ar);
@@ -155,10 +154,15 @@ private void doUnregister(Handler<AsyncResult<Void>> completionHandler) {
       }
       discardHandler = this.discardHandler;
       if (holder != null) {
+        handler = null;
         registered = null;
-        eventBus.removeRegistration(holder, completionHandler);
+        eventBus.removeRegistration(holder, doneHandler);
       } else {
-        callCompletionHandlerAsync(completionHandler);
+        callHandlerAsync(Future.succeededFuture(), doneHandler);
+      }
+      if (result == null) {
+        result = Future.failedFuture("Consumer unregistered before registration completed");
+        callHandlerAsync(result, completionHandler);
       }
     }
     if (discardHandler != null && discarded != null) {
@@ -169,9 +173,9 @@ private void doUnregister(Handler<AsyncResult<Void>> completionHandler) {
     }
   }
 
-  private void callCompletionHandlerAsync(Handler<AsyncResult<Void>> completionHandler) {
+  private void callHandlerAsync(AsyncResult<Void> result, Handler<AsyncResult<Void>> completionHandler) {
     if (completionHandler != null) {
-      vertx.runOnContext(v -> completionHandler.handle(Future.succeededFuture()));
+      vertx.runOnContext(v -> completionHandler.handle(result));
     }
   }
 
@@ -180,17 +184,17 @@ synchronized void setHandlerContext(Context context) {
   }
 
   public synchronized void setResult(AsyncResult<Void> result) {
+    if (this.result != null) {
+      return;
+    }
     this.result = result;
-    if (completionHandler != null) {
+    if (result.failed()) {
-      if (metrics != null && result.succeeded()) {
+      log.error("Failed to propagate registration for handler " + handler + " and address " + address);
+    } else {
+      if (metrics != null) {
         metric = metrics.handlerRegistered(address, repliedAddress);
       }
-      Handler<AsyncResult<Void>> callback = completionHandler;
+      callHandlerAsync(result, completionHandler);
-      vertx.runOnContext(v -> callback.handle(result));
-    } else if (result.failed()) {
-      log.error("Failed to propagate registration for handler " + handler + " and address " + address);
-    } else if (metrics != null) {
-      metric = metrics.handlerRegistered(address, repliedAddress);
     }
   }
 

diff --git a/src/test/java/io/vertx/core/eventbus/ClusteredEventBusTest.java b/src/test/java/io/vertx/core/eventbus/ClusteredEventBusTest.java
@@ -332,4 +332,22 @@ public void testLocalOnlyDoesNotApplyToReplies() {
     }));
     await();
   }
+
+  @Test
+  public void testImmediateUnregistration() {
+    startNodes(1);
+    MessageConsumer<Object> consumer = vertices[0].eventBus().consumer(ADDRESS1);
+    AtomicInteger completionCount = new AtomicInteger();
+    consumer.completionHandler(ar -> {
+      int val = completionCount.getAndIncrement();
+      assertEquals(0, val);
+      assertTrue(ar.failed());
+      vertx.setTimer(10, id -> {
+        testComplete();
+      });
+    });
+    consumer.handler(msg -> {});
+    consumer.unregister();
+    await();
+  }
 }
diff --git a/src/test/java/io/vertx/core/eventbus/LocalEventBusTest.java b/src/test/java/io/vertx/core/eventbus/LocalEventBusTest.java
@@ -1399,5 +1399,22 @@ public void testUnregisterConsumerDiscardPendingMessages() {
     eb.send(ADDRESS1, "val0");
     await();
   }
+
+  @Test
+  public void testImmediateUnregistration() {
+    MessageConsumer<Object> consumer = vertx.eventBus().consumer(ADDRESS1);
+    AtomicInteger completionCount = new AtomicInteger();
+    consumer.completionHandler(ar -> {
+      int val = completionCount.getAndIncrement();
+      assertEquals(0, val);
+      assertTrue(ar.succeeded());
+      vertx.setTimer(10, id -> {
+        testComplete();
+      });
+    });
+    consumer.handler(msg -> {});
+    consumer.unregister();
+    await();
+  }
 }