Move event bus metrics unregistration to HandlerRegistration instead of HandlerHolder - fixes #2900

src/main/java/io/vertx/core/eventbus/impl/EventBusImpl.java

@@ -276,7 +276,7 @@ public class EventBusImpl implements EventBus, MetricsProvider {
     }
     registration.setHandlerContext(context);
 
-    HandlerHolder<T> holder = new HandlerHolder<>(metrics, registration, replyHandler, localOnly, context);
+    HandlerHolder<T> holder = new HandlerHolder<>(registration, replyHandler, localOnly, context);
 
     ConcurrentCyclicSequence<HandlerHolder> handlers = new ConcurrentCyclicSequence<HandlerHolder>().add(holder);
     ConcurrentCyclicSequence<HandlerHolder> actualHandlers = handlerMap.merge(

src/main/java/io/vertx/core/eventbus/impl/HandlerHolder.java

@@ -19,16 +19,14 @@ import io.vertx.core.spi.metrics.EventBusMetrics;
  */
 public class HandlerHolder<T> {
 
-  private final EventBusMetrics metrics;
   private final Context context;
   private final HandlerRegistration<T> handler;
   private final boolean replyHandler;
   private final boolean localOnly;
   private boolean removed;
 
-  public HandlerHolder(EventBusMetrics metrics, HandlerRegistration<T> handler, boolean replyHandler, boolean localOnly,
+  public HandlerHolder(HandlerRegistration<T> handler, boolean replyHandler, boolean localOnly,
                        Context context) {
-    this.metrics = metrics;
     this.context = context;
     this.handler = handler;
     this.replyHandler = replyHandler;
@@ -36,7 +34,7 @@ public class HandlerHolder<T> {
   }
 
   // We use a synchronized block to protect removed as it can be unregistered from a different thread
-  public boolean setRemoved() {
+  boolean setRemoved() {
     boolean unregistered = false;
     synchronized (this) {
       if (!removed) {
@@ -44,9 +42,6 @@ public class HandlerHolder<T> {
         unregistered = true;
       }
     }
-    if (metrics != null && unregistered) {
-      metrics.handlerUnregistered(handler.getMetric());
-    }
     return unregistered;
   }
 

src/main/java/io/vertx/core/eventbus/impl/HandlerRegistration.java

@@ -163,6 +163,11 @@ public class HandlerRegistration<T> implements MessageConsumer<T>, Handler<Messa
       if (result == null) {
         result = Future.failedFuture("Consumer unregistered before registration completed");
         callHandlerAsync(result, completionHandler);
+      } else {
+        EventBusMetrics metrics = eventBus.metrics;
+        if (metrics != null) {
+          metrics.handlerUnregistered(metric);
+        }
       }
     }
     if (discardHandler != null && discarded != null) {

src/test/java/io/vertx/core/spi/metrics/MetricsTest.java

@@ -254,6 +254,22 @@ public class MetricsTest extends VertxTestBase {
     await();
   }
 
+  @Test
+  public void testClusterUnregistration() {
+    startNodes(1);
+    FakeEventBusMetrics metrics = FakeMetricsBase.getMetrics(vertices[0].eventBus());
+    MessageConsumer<Object> consumer = vertices[0].eventBus().consumer(ADDRESS1, ar -> {
+      fail("Should not receive message");
+    });
+    consumer.completionHandler(ar -> {
+      assertTrue(ar.failed());
+      assertEquals(Collections.emptyList(), metrics.getRegistrations());
+      testComplete();
+    });
+    consumer.unregister();
+    await();
+  }
+
   @Test
   public void testHandlerProcessMessage() {
     testHandlerProcessMessage(vertx, vertx, 1);

src/test/java/io/vertx/test/fakemetrics/FakeEventBusMetrics.java

@@ -80,6 +80,9 @@ public class FakeEventBusMetrics extends FakeMetricsBase implements EventBusMetr
   }
 
   public void handlerUnregistered(HandlerMetric handler) {
+    if (handler == null) {
+      throw new NullPointerException("Must not be null");
+    }
     registrations.remove(handler);
   }