Repository: eclipse-vertx/vert.x

Problem statement:
NullPointer in sendToSubs
Cloned from vert-x3/vertx-hazelcast#52


Hints:
@inno-steffg _commented on 7 Dec 2016_

We're using vertx 3.3.3 in a productive environment clustered with hazelcast.
We are observing the following NullPointerException from time to time:
```
2016-12-07 09:57:08,063 ERROR [vert.x-eventloop-thread-4] [] io.vertx.core.impl.ContextImpl - Unhandled exception
java.lang.NullPointerException
	at io.vertx.core.eventbus.impl.clustered.ClusteredEventBus.sendToSubs(ClusteredEventBus.java:279)
	at io.vertx.core.eventbus.impl.clustered.ClusteredEventBus.lambda$sendOrPub$133(ClusteredEventBus.java:182)
	at io.vertx.spi.cluster.hazelcast.impl.HazelcastAsyncMultiMap.lambda$get$30(HazelcastAsyncMultiMap.java:116)
	at io.vertx.core.impl.FutureImpl.checkCallHandler(FutureImpl.java:158)
	at io.vertx.core.impl.FutureImpl.setHandler(FutureImpl.java:100)
	at io.vertx.core.impl.ContextImpl.lambda$null$16(ContextImpl.java:305)
	at io.vertx.core.impl.ContextImpl.lambda$wrapTask$18(ContextImpl.java:335)
	at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:358)
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:357)
	at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:112)
	at java.lang.Thread.run(Thread.java:745)
```

I guess, this happens when we're unregistering an eventbus consumer at the same time the sendToSubs is trying to submit a message to it.

@vietj _commented on 7 Dec 2016_

it seems to be a race condition, what is your version ?
@inno-steffg _commented on 8 Dec 2016_

Vertx 3.2.1 and vertx 3.3.3 show this problem.
The stack trace above is from Vertx 3.2.1
Version 3.3.3 throws NullPointer at the same place (but another line no):
```
2016-12-08 01:46:18,435 ERROR [vert.x-eventloop-thread-2]  io.vertx.core.impl.ContextImpl - Unhandled exception
java.lang.NullPointerException
	at io.vertx.core.eventbus.impl.clustered.ClusteredEventBus.sendToSubs(ClusteredEventBus.java:315)
	at io.vertx.core.eventbus.impl.clustered.ClusteredEventBus.lambda$sendOrPub$4(ClusteredEventBus.java:218)
	at io.vertx.spi.cluster.hazelcast.impl.HazelcastAsyncMultiMap.lambda$get$3(HazelcastAsyncMultiMap.java:116)
	at io.vertx.core.impl.FutureImpl.checkCallHandler(FutureImpl.java:158)
	at io.vertx.core.impl.FutureImpl.setHandler(FutureImpl.java:100)
	at io.vertx.core.impl.ContextImpl.lambda$null$0(ContextImpl.java:271)
	at io.vertx.core.impl.ContextImpl.lambda$wrapTask$2(ContextImpl.java:316)
	at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:163)
	at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:418)
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:440)
	at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:873)
	at java.lang.Thread.run(Thread.java:745)
```
@tsegismont _commented on 9 Dec 2016_

Could you put together a small reproducer?

Patch:
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
@@ -332,7 +332,7 @@ private <T> void sendToSubs(ChoosableIterable<ServerID> subs, SendContextImpl<T>
     if (sendContext.message.isSend()) {
       // Choose one
       ServerID sid = subs.choose();
-      if (!sid.equals(serverID)) {  //We don't send to this node
+      if (sid != null && !sid.equals(serverID)) {  //We don't send to this node
         metrics.messageSent(address, false, false, true);
         sendRemote(sid, sendContext.message);
       } else {


Test patch:
diff --git a/src/test/java/io/vertx/test/core/EventBusTestBase.java b/src/test/java/io/vertx/test/core/EventBusTestBase.java
--- a/src/test/java/io/vertx/test/core/EventBusTestBase.java
+++ b/src/test/java/io/vertx/test/core/EventBusTestBase.java
@@ -19,9 +19,13 @@
 import io.netty.util.CharsetUtil;
 import io.vertx.core.AbstractVerticle;
 import io.vertx.core.DeploymentOptions;
+import io.vertx.core.Handler;
+import io.vertx.core.Vertx;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.eventbus.DeliveryOptions;
+import io.vertx.core.eventbus.Message;
 import io.vertx.core.eventbus.MessageCodec;
+import io.vertx.core.eventbus.MessageConsumer;
 import io.vertx.core.eventbus.ReplyException;
 import io.vertx.core.eventbus.ReplyFailure;
 import io.vertx.core.json.JsonArray;
@@ -29,8 +33,11 @@
 import org.junit.Test;
 
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Consumer;
 
+import static org.hamcrest.CoreMatchers.*;
+
 /**
  * @author <a href="http://tfox.org">Tim Fox</a>
  */
@@ -413,6 +420,82 @@ public void testSendFromExecuteBlocking() throws Exception {
     await();
   }
 
+  @Test
+  public void testSendWhileUnsubscribing() throws Exception {
+    startNodes(2);
+    Vertx vertx0 = vertices[0];
+    CountDownLatch registrationLatch = new CountDownLatch(1);
+    MessageConsumer<String> consumer = vertx0.eventBus().consumer("whatever");
+    consumer.handler(new Handler<Message<String>>() {
+      int received = 0;
+
+      @Override
+      public void handle(Message<String> msg) {
+        if (++received == 1) {
+          consumer.unregister();
+        }
+      }
+    }).completionHandler(ar -> {
+      if (ar.succeeded()) {
+        registrationLatch.countDown();
+      } else {
+        fail(ar.cause());
+      }
+    });
+    awaitLatch(registrationLatch);
+    AtomicBoolean send = new AtomicBoolean(true);
+    vertx0.exceptionHandler(throwable -> {
+      send.set(false);
+      fail(throwable);
+    });
+    while (send.get()) {
+      vertx0.eventBus().send("whatever", "marseille", ar -> {
+        if (send.get() && ar.failed()) {
+          Throwable cause = ar.cause();
+          assertThat(cause, instanceOf(ReplyException.class));
+          ReplyException replyException = (ReplyException) cause;
+          assertEquals(ReplyFailure.NO_HANDLERS, replyException.failureType());
+          send.set(false);
+        }
+      });
+    }
+  }
+
+  @Test
+  public void testPublishWhileUnsubscribing() throws Exception {
+    startNodes(2);
+    Vertx vertx0 = vertices[0];
+    CountDownLatch registrationLatch = new CountDownLatch(1);
+    AtomicBoolean send = new AtomicBoolean(true);
+    MessageConsumer<String> consumer = vertx0.eventBus().consumer("whatever");
+    consumer.handler(new Handler<Message<String>>() {
+      int received = 0;
+
+      @Override
+      public void handle(Message<String> msg) {
+        if (++received == 1) {
+          consumer.unregister(ar -> {
+            vertx0.setTimer(10, l -> send.set(false));
+          });
+        }
+      }
+    }).completionHandler(ar -> {
+      if (ar.succeeded()) {
+        registrationLatch.countDown();
+      } else {
+        fail(ar.cause());
+      }
+    });
+    awaitLatch(registrationLatch);
+    vertx0.exceptionHandler(throwable -> {
+      send.set(false);
+      fail(throwable);
+    });
+    while (send.get()) {
+      vertx0.eventBus().publish("whatever", "marseille");
+    }
+  }
+
   protected <T> void testSend(T val) {
     testSend(val, null);
   }