Repository: eclipse-vertx/vert.x

Problem statement:
Clustered consumer keeps local registration when unregistered
A method name was changed in `EventBusImpl` but not updated in `ClusteredEventBus` while re-working the clustering SPI.
This might have happened while rebasing #3132 on `master`.

The bug was not caught because we didn't have a test for unregistration.




Hints:

Patch:
diff --git a/src/main/java/io/vertx/core/eventbus/impl/EventBusImpl.java b/src/main/java/io/vertx/core/eventbus/impl/EventBusImpl.java
--- a/src/main/java/io/vertx/core/eventbus/impl/EventBusImpl.java
+++ b/src/main/java/io/vertx/core/eventbus/impl/EventBusImpl.java
@@ -236,7 +236,6 @@ public MessageImpl createMessage(boolean send, String address, MultiMap headers,
   }
 
   protected <T> HandlerHolder<T> addRegistration(String address, HandlerRegistration<T> registration, boolean replyHandler, boolean localOnly, Promise<Void> promise) {
-//    Objects.requireNonNull(registration.getHandler(), "handler");
     HandlerHolder<T> holder = addLocalRegistration(address, registration, replyHandler, localOnly);
     onLocalRegistration(holder, promise);
     return holder;
diff --git a/src/main/java/io/vertx/core/eventbus/impl/clustered/ClusteredEventBus.java b/src/main/java/io/vertx/core/eventbus/impl/clustered/ClusteredEventBus.java
--- a/src/main/java/io/vertx/core/eventbus/impl/clustered/ClusteredEventBus.java
+++ b/src/main/java/io/vertx/core/eventbus/impl/clustered/ClusteredEventBus.java
@@ -11,7 +11,6 @@
 
 package io.vertx.core.eventbus.impl.clustered;
 
-import io.vertx.core.Future;
 import io.vertx.core.Handler;
 import io.vertx.core.MultiMap;
 import io.vertx.core.Promise;
@@ -145,7 +144,7 @@ protected <T> HandlerHolder<T> createHandlerHolder(HandlerRegistration<T> regist
   }
 
   @Override
-  protected <T> void removeRegistration(HandlerHolder<T> handlerHolder, Promise<Void> completionHandler) {
+  protected <T> void onLocalUnregistration(HandlerHolder<T> handlerHolder, Promise<Void> completionHandler) {
     if (!handlerHolder.isReplyHandler()) {
       RegistrationInfo registrationInfo = new RegistrationInfo(
         nodeId,
@@ -154,11 +153,7 @@ protected <T> void removeRegistration(HandlerHolder<T> handlerHolder, Promise<Vo
       );
       Promise<Void> promise = Promise.promise();
       clusterManager.removeRegistration(handlerHolder.getHandler().address, registrationInfo, promise);
-      if (completionHandler != null) {
-        promise.future().onComplete(completionHandler);
-      } else {
-        promise.future().onFailure(t -> log.error("Failed to remove sub", t));
-      }
+      promise.future().onComplete(completionHandler);
     } else {
       completionHandler.complete();
     }


Test patch:
diff --git a/src/test/java/io/vertx/core/eventbus/ClusteredEventBusTestBase.java b/src/test/java/io/vertx/core/eventbus/ClusteredEventBusTestBase.java
--- a/src/test/java/io/vertx/core/eventbus/ClusteredEventBusTestBase.java
+++ b/src/test/java/io/vertx/core/eventbus/ClusteredEventBusTestBase.java
@@ -23,6 +23,9 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Consumer;
 
+import static org.hamcrest.CoreMatchers.instanceOf;
+import static org.hamcrest.CoreMatchers.is;
+
 
 /**
  * @author <a href="http://tfox.org">Tim Fox</a>
@@ -204,4 +207,30 @@ public void testMessageBodyInterceptor() throws Exception {
     }).send("whatever", content, new DeliveryOptions().setCodecName(codec.name()));
     await();
   }
+
+  @Test
+  public void testClusteredUnregistration() {
+    startNodes(2);
+    MessageConsumer<Object> consumer = vertices[0].eventBus().consumer("foo", msg -> msg.reply(msg.body()));
+    consumer.completionHandler(onSuccess(reg -> {
+      vertices[0].eventBus().request("foo", "echo", onSuccess(reply1 -> {
+        assertEquals("echo", reply1.body());
+        vertices[1].eventBus().request("foo", "echo", onSuccess(reply2 -> {
+          assertEquals("echo", reply1.body());
+          consumer.unregister(onSuccess(unreg -> {
+            vertices[1].eventBus().request("foo", "echo", onFailure(fail1 -> {
+              assertThat(fail1, is(instanceOf(ReplyException.class)));
+              assertEquals(ReplyFailure.NO_HANDLERS, ((ReplyException) fail1).failureType());
+              vertices[0].eventBus().request("foo", "echo", onFailure(fail2 -> {
+                assertThat(fail2, is(instanceOf(ReplyException.class)));
+                assertEquals(ReplyFailure.NO_HANDLERS, ((ReplyException) fail2).failureType());
+                testComplete();
+              }));
+            }));
+          }));
+        }));
+      }));
+    }));
+    await();
+  }
 }