Optimize node and subscription cleanup on node exit

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>

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,34 +309,31 @@ 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)) {
                     cleanSubs(id);
                   }
                   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");
                   } else {
                     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

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);
+        }
       }
     );
   }