diff --git a/pom.xml b/pom.xml
index e21226c..e635be9 100644
--- a/pom.xml
+++ b/pom.xml
@@ -104,7 +104,7 @@
     <dependency>
       <groupId>ch.qos.logback</groupId>
       <artifactId>logback-classic</artifactId>
-      <version>1.2.11</version>
+      <version>1.4.12</version>
       <scope>test</scope>
     </dependency>
     <dependency>
@@ -162,14 +162,44 @@
             </argLine>
             <forkCount>1</forkCount>
             <reuseForks>true</reuseForks>
-            <excludes>
-              <exclude>**/it/**/*Test.java</exclude>
-            </excludes>
+          </configuration>
+        </plugin>
+        <plugin>
+          <groupId>org.apache.maven.plugins</groupId>
+          <artifactId>maven-failsafe-plugin</artifactId>
+          <version>2.19.1</version>
+          <configuration>
+            <failIfNoSpecifiedTests>false</failIfNoSpecifiedTests>
+            <systemPropertyVariables>
+              <java.net.preferIPv4Stack>true</java.net.preferIPv4Stack>
+              <IGNITE_PERFORMANCE_SUGGESTIONS_DISABLED>true</IGNITE_PERFORMANCE_SUGGESTIONS_DISABLED>
+              <IGNITE_NO_ASCII>true</IGNITE_NO_ASCII>
+              <IGNITE_UPDATE_NOTIFIER>false</IGNITE_UPDATE_NOTIFIER>
+              <IGNITE_ATOMIC_CACHE_DELETE_HISTORY_SIZE>1000</IGNITE_ATOMIC_CACHE_DELETE_HISTORY_SIZE>
+              <vertx.logger-delegate-factory-class-name>io.vertx.core.logging.SLF4JLogDelegateFactory</vertx.logger-delegate-factory-class-name>
+              <io.netty.leakDetectionLevel>PARANOID</io.netty.leakDetectionLevel>
+              <buildDirectory>${project.build.directory}</buildDirectory>
+              <vertxVersion>${project.version}</vertxVersion>
+            </systemPropertyVariables>
+            <argLine>
+              <!-- Needs to be small enough to run in a EC2 1.7GB small instance -->
+              -Xms512M -Xmx1200M
+            </argLine>
+            <forkCount>1</forkCount>
+            <reuseForks>true</reuseForks>
           </configuration>
         </plugin>
       </plugins>
     </pluginManagement>
     <plugins>
+      <plugin>
+        <artifactId>maven-surefire-plugin</artifactId>
+        <configuration>
+          <excludes>
+            <exclude>**/it/**/*Test.java</exclude>
+          </excludes>
+        </configuration>
+      </plugin>
       <plugin>
         <artifactId>maven-failsafe-plugin</artifactId>
         <executions>
@@ -216,6 +246,21 @@
               </argLine>
             </configuration>
           </plugin>
+          <plugin>
+            <groupId>org.apache.maven.plugins</groupId>
+            <artifactId>maven-failsafe-plugin</artifactId>
+            <configuration>
+              <argLine>
+                <!-- Needs to be small enough to run in a EC2 1.7GB small instance -->
+                -Xms512M -Xmx1200M
+                <!-- setup for JDK 17+ to replace illegal-access -->
+                --add-opens=java.base/java.nio=ALL-UNNAMED
+                --add-opens=java.base/java.util=ALL-UNNAMED
+                --add-opens=java.base/java.lang.invoke=ALL-UNNAMED
+                --add-opens=java.base/java.lang=ALL-UNNAMED
+              </argLine>
+            </configuration>
+          </plugin>
         </plugins>
       </build>
     </profile>
diff --git a/src/main/java/io/vertx/spi/cluster/ignite/IgniteClusterManager.java b/src/main/java/io/vertx/spi/cluster/ignite/IgniteClusterManager.java
index c57a04e..183fb1e 100644
--- a/src/main/java/io/vertx/spi/cluster/ignite/IgniteClusterManager.java
+++ b/src/main/java/io/vertx/spi/cluster/ignite/IgniteClusterManager.java
@@ -50,7 +50,6 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Consumer;
-import java.util.stream.Collectors;
 
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static java.util.concurrent.TimeUnit.NANOSECONDS;
@@ -197,7 +196,7 @@ public void nodeListener(NodeListener nodeListener) {
 
   @Override
   public <K, V> void getAsyncMap(String name, Promise<AsyncMap<K, V>> promise) {
-    vertx.<AsyncMap<K, V>>executeBlocking(prom -> prom.complete(new AsyncMapImpl<>(getCache(name), vertx))).onComplete(promise);
+    vertx.<AsyncMap<K, V>>executeBlocking(() -> new AsyncMapImpl<>(getCache(name), vertx)).onComplete(promise);
   }
 
   @Override
@@ -207,7 +206,7 @@ public <K, V> Map<K, V> getSyncMap(String name) {
 
   @Override
   public void getLockWithTimeout(String name, long timeout, Promise<Lock> promise) {
-    vertx.<Lock>executeBlocking(prom -> {
+    vertx.<Lock>executeBlocking(() -> {
       IgniteSemaphore semaphore = ignite.semaphore(LOCK_SEMAPHORE_PREFIX + name, 1, true, true);
       boolean locked;
       long remaining = timeout;
@@ -217,7 +216,7 @@ public void getLockWithTimeout(String name, long timeout, Promise<Lock> promise)
         remaining = remaining - TimeUnit.MILLISECONDS.convert(System.nanoTime() - start, NANOSECONDS);
       } while (!locked && remaining > 0);
       if (locked) {
-        prom.complete(new LockImpl(semaphore, lockReleaseExec));
+        return new LockImpl(semaphore, lockReleaseExec);
       } else {
         throw new VertxException("Timed out waiting to get lock " + name);
       }
@@ -226,7 +225,7 @@ public void getLockWithTimeout(String name, long timeout, Promise<Lock> promise)
 
   @Override
   public void getCounter(String name, Promise<Counter> promise) {
-    vertx.<Counter>executeBlocking(prom -> prom.complete(new CounterImpl(ignite.atomicLong(name, 0, true)))).onComplete(promise);
+    vertx.<Counter>executeBlocking(() -> new CounterImpl(ignite.atomicLong(name, 0, true))).onComplete(promise);
   }
 
   @Override
@@ -240,9 +239,9 @@ public void setNodeInfo(NodeInfo nodeInfo, Promise<Void> promise) {
       this.nodeInfo = nodeInfo;
     }
     IgniteNodeInfo value = new IgniteNodeInfo(nodeInfo);
-    vertx.<Void>executeBlocking(prom -> {
+    vertx.<Void>executeBlocking(() -> {
       nodeInfoMap.put(nodeId, value);
-      prom.complete();
+      return null;
     }, false).onComplete(promise);
   }
 
@@ -270,8 +269,15 @@ public void getNodeInfo(String id, Promise<NodeInfo> promise) {
   @Override
   public List<String> getNodes() {
     try {
-      return ignite.cluster().nodes().stream()
-        .map(IgniteClusterManager::nodeId).collect(Collectors.toList());
+      Collection<ClusterNode> nodes = ignite.cluster().nodes();
+      List<String> nodeIds;
+      synchronized (nodes) {
+        nodeIds = new ArrayList<>(nodes.size());
+        for (ClusterNode node : nodes) {
+          nodeIds.add(nodeId(node));
+        }
+      }
+      return nodeIds;
     } catch (IllegalStateException e) {
       log.debug(e.getMessage());
       return Collections.emptyList();
@@ -280,7 +286,7 @@ public List<String> getNodes() {
 
   @Override
   public void join(Promise<Void> promise) {
-    vertx.<Void>executeBlocking(prom -> {
+    vertx.<Void>executeBlocking(() -> {
       synchronized (monitor) {
         if (!active) {
           active = true;
@@ -303,14 +309,13 @@ public void join(Promise<Void> promise) {
               return false;
             }
 
-            vertx.<Void>executeBlocking(f -> {
+            vertx.<Void>executeBlocking(() -> {
               String id = nodeId(((DiscoveryEvent) event).eventNode());
               switch (event.type()) {
                 case EVT_NODE_JOINED:
                   notifyNodeListener(listener -> listener.nodeAdded(id));
                   log.debug("node " + id + " joined the cluster");
-                  f.complete();
-                  break;
+                  return null;
                 case EVT_NODE_LEFT:
                 case EVT_NODE_FAILED:
                   if (cleanNodeInfos(id)) {
@@ -318,8 +323,7 @@ public void join(Promise<Void> promise) {
                   }
                   notifyNodeListener(listener -> listener.nodeLeft(id));
                   log.debug("node " + id + " left the cluster");
-                  f.complete();
-                  break;
+                  return null;
                 case EVT_NODE_SEGMENTED:
                   if (customIgnite || !shutdownOnSegmentation) {
                     log.warn("node got segmented");
@@ -327,10 +331,9 @@ public void join(Promise<Void> promise) {
                     log.warn("node got segmented and will be shut down");
                     vertx.close();
                   }
-                  f.fail(new IllegalStateException("node is stopped"));
-                  break;
+                  throw new IllegalStateException("node is stopped");
                 default:
-                  f.fail("event not known");
+                  throw new IllegalStateException("event not known");
               }
             });
 
@@ -343,18 +346,18 @@ public void join(Promise<Void> promise) {
 
           try {
             MILLISECONDS.sleep(delayAfterStart);
-            prom.complete();
           } catch (InterruptedException e) {
-            prom.fail(e);
+            throw new IllegalStateException(e);
           }
         }
+        return null;
       }
     }).onComplete(promise);
   }
 
   @Override
   public void leave(Promise<Void> promise) {
-    vertx.<Void>executeBlocking(prom -> {
+    vertx.<Void>executeBlocking(() -> {
       synchronized (monitor) {
         if (active) {
           active = false;
@@ -370,12 +373,10 @@ public void leave(Promise<Void> promise) {
           } catch (Exception e) {
             log.error(e);
           }
-          subsMapHelper = null;
-          nodeInfoMap = null;
         }
       }
 
-      prom.complete();
+      return null;
     }).onComplete(promise);
   }
 
@@ -386,39 +387,29 @@ public boolean isActive() {
 
   @Override
   public void addRegistration(String address, RegistrationInfo registrationInfo, Promise<Void> promise) {
-    vertx.<Void>executeBlocking(prom -> {
-      subsMapHelper.put(address, registrationInfo)
-        .onComplete(prom);
-    }, false).onComplete(promise);
+    subsMapHelper.put(address, registrationInfo)
+        .onComplete(promise);
   }
 
   @Override
   public void removeRegistration(String address, RegistrationInfo registrationInfo, Promise<Void> promise) {
-    vertx.<Void>executeBlocking(prom -> {
-      subsMapHelper.remove(address, registrationInfo, prom);
-    }, false).onComplete(promise);
+    subsMapHelper.remove(address, registrationInfo, promise);
   }
 
   @Override
   public void getRegistrations(String address, Promise<List<RegistrationInfo>> promise) {
-    vertx.<List<RegistrationInfo>>executeBlocking(prom -> {
-      subsMapHelper.get(address, prom);
-    }, false).onComplete(promise);
+    subsMapHelper.get(address, promise);
   }
 
   private void cleanSubs(String id) {
-    try {
-      subsMapHelper.removeAllForNode(id);
-    } catch (IllegalStateException | CacheException e) {
-      //ignore
-    }
+    subsMapHelper.removeAllForNode(id);
   }
 
   private boolean cleanNodeInfos(String nid) {
     try {
       return nodeInfoMap.remove(nid);
     } catch (IllegalStateException | CacheException e) {
-      //ignore
+      log.warn("Could not remove nodeInfo (" + nid + "): " + e.getMessage());
     }
     return false;
   }
@@ -519,7 +510,7 @@ private CounterImpl(IgniteAtomicLong cnt) {
 
     @Override
     public Future<Long> get() {
-      return vertx.executeBlocking(fut -> fut.complete(cnt.get()));
+      return vertx.executeBlocking(cnt::get);
     }
 
     @Override
@@ -530,7 +521,7 @@ public void get(Handler<AsyncResult<Long>> handler) {
 
     @Override
     public Future<Long> incrementAndGet() {
-      return vertx.executeBlocking(fut -> fut.complete(cnt.incrementAndGet()));
+      return vertx.executeBlocking(cnt::incrementAndGet);
     }
 
     @Override
@@ -541,7 +532,7 @@ public void incrementAndGet(Handler<AsyncResult<Long>> handler) {
 
     @Override
     public Future<Long> getAndIncrement() {
-      return vertx.executeBlocking(fut -> fut.complete(cnt.getAndIncrement()));
+      return vertx.executeBlocking(cnt::getAndIncrement);
     }
 
     @Override
@@ -552,7 +543,7 @@ public void getAndIncrement(Handler<AsyncResult<Long>> handler) {
 
     @Override
     public Future<Long> decrementAndGet() {
-      return vertx.executeBlocking(fut -> fut.complete(cnt.decrementAndGet()));
+      return vertx.executeBlocking(cnt::decrementAndGet);
     }
 
     @Override
@@ -563,7 +554,7 @@ public void decrementAndGet(Handler<AsyncResult<Long>> handler) {
 
     @Override
     public Future<Long> addAndGet(long value) {
-      return vertx.executeBlocking(fut -> fut.complete(cnt.addAndGet(value)));
+      return vertx.executeBlocking(() -> cnt.addAndGet(value));
     }
 
     @Override
@@ -574,7 +565,7 @@ public void addAndGet(long value, Handler<AsyncResult<Long>> handler) {
 
     @Override
     public Future<Long> getAndAdd(long value) {
-      return vertx.executeBlocking(fut -> fut.complete(cnt.getAndAdd(value)));
+      return vertx.executeBlocking(() -> cnt.getAndAdd(value));
     }
 
     @Override
@@ -585,7 +576,7 @@ public void getAndAdd(long value, Handler<AsyncResult<Long>> handler) {
 
     @Override
     public Future<Boolean> compareAndSet(long expected, long value) {
-      return vertx.executeBlocking(fut -> fut.complete(cnt.compareAndSet(expected, value)));
+      return vertx.executeBlocking(() -> cnt.compareAndSet(expected, value));
     }
 
     @Override
diff --git a/src/main/java/io/vertx/spi/cluster/ignite/impl/AsyncMapImpl.java b/src/main/java/io/vertx/spi/cluster/ignite/impl/AsyncMapImpl.java
index cd7b7d7..3198ec6 100644
--- a/src/main/java/io/vertx/spi/cluster/ignite/impl/AsyncMapImpl.java
+++ b/src/main/java/io/vertx/spi/cluster/ignite/impl/AsyncMapImpl.java
@@ -135,16 +135,16 @@ public Future<List<V>> values() {
   @Override
   public Future<Map<K, V>> entries() {
     return vertx.executeBlocking(
-      promise -> {
+      () -> {
         try {
           List<Cache.Entry<K, V>> all = cache.query(new ScanQuery<K, V>()).getAll();
           Map<K, V> map = new HashMap<>(all.size());
           for (Cache.Entry<K, V> entry : all) {
             map.put(unmarshal(entry.getKey()), unmarshal(entry.getValue()));
           }
-          promise.complete(map);
+          return map;
         } catch (final RuntimeException cause) {
-          promise.fail(new VertxException(cause));
+          throw new VertxException(cause);
         }
       }
     );
@@ -163,17 +163,12 @@ private <T> Future<T> executeWithTtl(Function<IgniteCache<K, V>, IgniteFuture<T>
       : cache;
 
     return vertx.executeBlocking(
-      promise -> {
-        IgniteFuture<T> future = cacheOp.apply(cache0);
-        future.listen(
-          fut -> {
-            try {
-              promise.complete(unmarshal(future.get()));
-            } catch (final RuntimeException e) {
-              promise.fail(new VertxException(e));
-            }
-          }
-        );
+      () -> {
+        try {
+          return unmarshal(cacheOp.apply(cache0).get());
+        } catch (final RuntimeException e) {
+          throw new VertxException(e);
+        }
       }
     );
   }
diff --git a/src/main/java/io/vertx/spi/cluster/ignite/impl/SubsMapHelper.java b/src/main/java/io/vertx/spi/cluster/ignite/impl/SubsMapHelper.java
index 9b7c577..6c531d2 100644
--- a/src/main/java/io/vertx/spi/cluster/ignite/impl/SubsMapHelper.java
+++ b/src/main/java/io/vertx/spi/cluster/ignite/impl/SubsMapHelper.java
@@ -19,6 +19,8 @@
 import io.vertx.core.Promise;
 import io.vertx.core.VertxException;
 import io.vertx.core.impl.VertxInternal;
+import io.vertx.core.impl.logging.Logger;
+import io.vertx.core.impl.logging.LoggerFactory;
 import io.vertx.core.spi.cluster.NodeSelector;
 import io.vertx.core.spi.cluster.RegistrationInfo;
 import io.vertx.core.spi.cluster.RegistrationUpdateEvent;
@@ -33,16 +35,15 @@
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
-import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
 
-import static java.util.stream.Collectors.toList;
-
 /**
  * @author Thomas Segismont
  * @author Lukas Prettenthaler
  */
 public class SubsMapHelper {
+  private static final Logger log = LoggerFactory.getLogger(SubsMapHelper.class);
+
   private final IgniteCache<IgniteRegistrationInfo, Boolean> map;
   private final NodeSelector nodeSelector;
   private final ConcurrentMap<String, Set<RegistrationInfo>> localSubs = new ConcurrentHashMap<>();
@@ -67,12 +68,11 @@ public void get(String address, Promise<List<RegistrationInfo>> promise) {
       return;
     }
     try {
-      List<RegistrationInfo> infos = map.query(new ScanQuery<IgniteRegistrationInfo, Boolean>((k, v) -> k.address().equals(address)))
-        .getAll().stream()
-        .map(Cache.Entry::getKey)
-        .map(IgniteRegistrationInfo::registrationInfo)
-        .collect(toList());
-      int size = infos.size();
+      List<Cache.Entry<IgniteRegistrationInfo, Boolean>> remote = map
+        .query(new ScanQuery<IgniteRegistrationInfo, Boolean>((k, v) -> k.address().equals(address)))
+        .getAll();
+      List<RegistrationInfo> infos;
+      int size = remote.size();
       Set<RegistrationInfo> local = localSubs.get(address);
       if (local != null) {
         synchronized (local) {
@@ -81,13 +81,18 @@ public void get(String address, Promise<List<RegistrationInfo>> promise) {
             promise.complete(Collections.emptyList());
             return;
           }
+          infos = new ArrayList<>(size);
           infos.addAll(local);
         }
       } else if (size == 0) {
         promise.complete(Collections.emptyList());
         return;
+      } else {
+        infos = new ArrayList<>(size);
+      }
+      for (Cache.Entry<IgniteRegistrationInfo, Boolean> info : remote) {
+        infos.add(info.getKey().registrationInfo());
       }
-
       promise.complete(infos);
     } catch (IllegalStateException | CacheException e) {
       promise.fail(new VertxException(e));
@@ -127,7 +132,7 @@ public void remove(String address, RegistrationInfo registrationInfo, Promise<Vo
         localSubs.computeIfPresent(address, (add, curr) -> removeFromSet(registrationInfo, curr));
         fireRegistrationUpdateEvent(address);
       } else {
-        map.remove(new IgniteRegistrationInfo(address, registrationInfo));
+        map.remove(new IgniteRegistrationInfo(address, registrationInfo), Boolean.TRUE);
       }
       promise.complete();
     } catch (IllegalStateException | CacheException e) {
@@ -141,14 +146,15 @@ private Set<RegistrationInfo> removeFromSet(RegistrationInfo registrationInfo, S
   }
 
   public void removeAllForNode(String nodeId) {
-    TreeSet<IgniteRegistrationInfo> toRemove = map.query(new ScanQuery<IgniteRegistrationInfo, Boolean>((k, v) -> k.registrationInfo().nodeId().equals(nodeId)))
-      .getAll().stream()
-      .map(Cache.Entry::getKey)
-      .collect(Collectors.toCollection(TreeSet::new));
-    try {
-      map.removeAll(toRemove);
-    } catch (IllegalStateException | CacheException t) {
-      //ignore
+    List<Cache.Entry<IgniteRegistrationInfo, Boolean>> toRemove = map
+      .query(new ScanQuery<IgniteRegistrationInfo, Boolean>((k, v) -> k.registrationInfo().nodeId().equals(nodeId)))
+      .getAll();
+    for (Cache.Entry<IgniteRegistrationInfo, Boolean> info : toRemove) {
+      try {
+        map.remove(info.getKey(), Boolean.TRUE);
+      } catch (IllegalStateException | CacheException t) {
+        log.warn("Could not remove subscriber: " + t.getMessage());
+      }
     }
   }
 
@@ -174,12 +180,12 @@ private Future<List<RegistrationInfo>> getAndUpdate(String address) {
   }
 
   private void listen(final Iterable<CacheEntryEvent<? extends IgniteRegistrationInfo, ? extends Boolean>> events, final VertxInternal vertxInternal) {
-    vertxInternal.<List<RegistrationInfo>>executeBlocking(promise -> {
+    vertxInternal.<List<RegistrationInfo>>executeBlocking(() -> {
       StreamSupport.stream(events.spliterator(), false)
         .map(e -> e.getKey().address())
         .distinct()
         .forEach(this::fireRegistrationUpdateEvent);
-      promise.complete();
+      return null;
     });
   }
 }
diff --git a/src/test/java/io/vertx/spi/cluster/ignite/impl/ThrottlingTest.java b/src/test/java/io/vertx/spi/cluster/ignite/impl/ThrottlingTest.java
index ee44ea3..1227dbb 100644
--- a/src/test/java/io/vertx/spi/cluster/ignite/impl/ThrottlingTest.java
+++ b/src/test/java/io/vertx/spi/cluster/ignite/impl/ThrottlingTest.java
@@ -48,7 +48,7 @@ public void testInterval() throws Exception {
     Throttling throttling = new Throttling((VertxInternal) vertx, address -> {
       events.compute(address, (k, v) -> {
         if (v == null) {
-          v = Collections.synchronizedList(new LinkedList<>());
+          v = Collections.synchronizedList(Collections.synchronizedList(new LinkedList<>()));
         }
         v.add(System.nanoTime());
         return v;
diff --git a/src/test/resources/logback-test.xml b/src/test/resources/logback-test.xml
index ad977b6..d21466a 100644
--- a/src/test/resources/logback-test.xml
+++ b/src/test/resources/logback-test.xml
@@ -22,6 +22,7 @@
     </encoder>
   </appender>
   <logger name="org.apache.ignite.IgniteLogger" level="ERROR"/>
+  <logger name="io.vertx.core.eventbus.impl.clustered.ConnectionHolder" level="ERROR"/>
   <root level="${testLogLevel:-INFO}">
     <appender-ref ref="STDOUT"/>
   </root>