fix: Reduce number of round trips for redis (#172)

Co-authored-by: C5371248 <ivan.kalachyev@sap.com>
Co-authored-by: C5371248 <baton4ik@gmail.com>

Author: sasjo

.github/workflows/pre-commit.yaml

@@ -18,6 +18,6 @@ jobs:
         uses: actions/setup-python@v6
 
       - name: Run pre-commit
-        uses: pre-commit/actions@v3.0.1
+        uses: pre-commit/action@v3.0.1
         with:
           extra_args: --from-ref=${{ github.event.pull_request.base.sha }} --to-ref=${{ github.sha }}

.github/workflows/pull_request.yaml

@@ -4,6 +4,9 @@ on:
     branches: [ master ]
     types: [ opened, synchronize, reopened ]
 
+env:
+  MAVEN_INIT: 'false'
+
 jobs:
   test:
     runs-on: ubuntu-latest
@@ -22,7 +25,7 @@ jobs:
         run: ./mvnw verify
 
       - name: Scan with SonarCloud
-        if: ${{ env.SECRET_AUTH != '' }}
+        if: github.repository == 'extenda/vertx-redis-clustermanager'
         uses: extenda/actions/sonar-scanner@v0
         with:
           sonar-host: https://sonarcloud.io

src/main/java/com/retailsvc/vertx/spi/cluster/redis/RedisClusterManager.java

@@ -125,12 +125,12 @@ public void nodeListener(NodeListener listener) {
 
   @Override
   public void setNodeInfo(NodeInfo nodeInfo, Promise<Void> promise) {
-    try (var ignored = CloseableLock.lock(lock)) {
-      this.nodeInfo = nodeInfo;
-    }
     vertx
         .<Void>executeBlocking(
             () -> {
+              try (var ignored = CloseableLock.lock(lock)) {
+                this.nodeInfo = nodeInfo;
+              }
               nodeInfoCatalog.setNodeInfo(nodeInfo);
               return null;
             },

src/main/java/com/retailsvc/vertx/spi/cluster/redis/impl/RedissonContext.java

@@ -35,7 +35,7 @@ public final class RedissonContext {
 
   private RedissonClient client;
   private ExecutorService lockReleaseExec;
-  private final Lock lock = new ReentrantLock();
+  private final Lock lock = new ReentrantLock(true);
 
   /**
    * Create a new Redisson context with specified configuration.
@@ -58,9 +58,8 @@ public RedissonContext(RedisConfig config, ClassLoader dataClassLoader) {
 
     BaseConfig<?> serverConfig =
         switch (config.getClientType()) {
-          case STANDALONE -> redisConfig
-              .useSingleServer()
-              .setAddress(config.getEndpoints().getFirst());
+          case STANDALONE ->
+              redisConfig.useSingleServer().setAddress(config.getEndpoints().getFirst());
           case CLUSTER -> {
             ClusterServersConfig clusterConfig = redisConfig.useClusterServers();
             clusterConfig.setNodeAddresses(config.getEndpoints());

src/main/java/com/retailsvc/vertx/spi/cluster/redis/impl/SubscriptionCatalog.java

@@ -8,14 +8,18 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
+import org.redisson.api.RSet;
 import org.redisson.api.RSetMultimap;
 import org.redisson.api.RTopic;
 import org.redisson.api.RedissonClient;
@@ -37,10 +41,9 @@ public class SubscriptionCatalog {
   private final NodeSelector nodeSelector;
   private final int listenerId;
   private final RTopic topic;
-
   private final ConcurrentMap<String, Set<RegistrationInfo>> localSubs = new ConcurrentHashMap<>();
   private final ConcurrentMap<String, Set<RegistrationInfo>> ownSubs = new ConcurrentHashMap<>();
-  private final ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
+  private final ReadWriteLock readWriteLock = new ReentrantReadWriteLock(true);
   private final Throttling throttling;
 
   /**
@@ -219,31 +222,103 @@ public void removeAllForNodes(Set<String> nodeIds) {
    * Remove subscriptions for nodes that are not part of the <code>availableNodeIds</code>
    * collection. Own subscriptions are never removed.
    *
-   * <p>Unknown nodes with lingering state can observed in clusters that has scaled down to one and
-   * then crashes. If a new node is not available to receive events when node entries expire in
+   * <p>Unknown nodes with lingering state can be observed in clusters that has scaled down to one
+   * and then crashes. If a new node is not available to receive events when node entries expire in
    * Redis, the subscriptions will remain registered in Redis.
    *
    * @param self the node ID of this process
    * @param availableNodeIds a set of available nodes
    */
   public void removeUnknownSubs(String self, Collection<String> availableNodeIds) {
-    Set<String> known = new HashSet<>(availableNodeIds);
-    known.add(self);
 
-    Set<String> updated = new HashSet<>();
-    subsMap
-        .entries()
-        .forEach(
-            entry -> {
-              if (!known.contains(entry.getValue().nodeId())) {
-                log.warn(
-                    "Remove lingering subscriptions from unknown node [{}]",
-                    entry.getValue().nodeId());
-                subsMap.remove(entry.getKey(), entry.getValue());
-                updated.add(entry.getKey());
-              }
-            });
-    updated.forEach(topic::publish);
+    // Build set of known nodes (cluster members + self)
+    Set<String> knownNodes = new HashSet<>(availableNodeIds);
+    knownNodes.add(self);
+
+    // Group stale subs by subscription key (address)
+    Map<String, List<RegistrationInfo>> cleanupByKey = new HashMap<>();
+
+    for (Map.Entry<String, RegistrationInfo> entry : subsMap.entries()) {
+      RegistrationInfo info = entry.getValue();
+      if (!knownNodes.contains(info.nodeId())) {
+        cleanupByKey.computeIfAbsent(entry.getKey(), k -> new ArrayList<>()).add(info);
+      }
+    }
+
+    if (cleanupByKey.isEmpty()) {
+      return;
+    }
+
+    // Bulk remove all unknown subscriptions
+    List<String> updatedKeys = bulkRemoveUnknownSubsByKey(cleanupByKey, subsMap);
+
+    // Notify all updated keys across the cluster nodes
+    updatedKeys.forEach(topic::publish);
+  }
+
+  /**
+   * Bulk remove unknown subscriptions grouped by key (address) to reduce round trips to Redis. If a
+   * bulk deletion fails, this method falls back to individual deletion per entry.
+   *
+   * @param cleanupByKey the stale subscriptions to remove, grouped by key (address)
+   * @param subscriptions the subscriptions map in Redis
+   * @return the updated keys (addresses)
+   */
+  static List<String> bulkRemoveUnknownSubsByKey(
+      Map<String, List<RegistrationInfo>> cleanupByKey,
+      RSetMultimap<String, RegistrationInfo> subscriptions) {
+    // Stats
+    AtomicInteger totalRemoved = new AtomicInteger();
+    AtomicInteger totalFailed = new AtomicInteger();
+    Map<String, Integer> removedPerNode = new HashMap<>();
+
+    List<String> updatedKeys = new ArrayList<>();
+    for (Map.Entry<String, List<RegistrationInfo>> entry : cleanupByKey.entrySet()) {
+      String key = entry.getKey();
+      List<RegistrationInfo> staleValues = entry.getValue();
+      try {
+        RSet<RegistrationInfo> set = subscriptions.get(key);
+
+        // Bulk remove all (single round trip to Redis)
+        set.removeAll(staleValues);
+
+        // Mark as updated because stale entries existed
+        updatedKeys.add(key);
+
+        // Stats
+        for (RegistrationInfo info : staleValues) {
+          removedPerNode.merge(info.nodeId(), 1, Integer::sum);
+          totalRemoved.incrementAndGet();
+        }
+      } catch (Exception e) {
+        // Fallback to safe per-value removal
+        log.warn("Bulk removal failed for key [{}], retrying individually", key, e);
+
+        for (RegistrationInfo info : staleValues) {
+          boolean ok = subscriptions.remove(key, info);
+          if (ok) {
+            removedPerNode.merge(info.nodeId(), 1, Integer::sum);
+            totalRemoved.incrementAndGet();
+          } else {
+            totalFailed.incrementAndGet();
+          }
+        }
+
+        updatedKeys.add(key);
+      }
+    }
+
+    // Logging summary
+    log.warn(
+        "Removed {} lingering subscriptions from {} unknown node(s). Breakdown: {}",
+        totalRemoved.get(),
+        removedPerNode.size(),
+        removedPerNode);
+
+    if (totalFailed.get() > 0) {
+      log.warn("Failed to remove {} subscriptions", totalFailed.get());
+    }
+    return updatedKeys;
   }
 
   /** Republish subscriptions that belongs to the current node (in which this is executed). */

src/test/java/com/retailsvc/vertx/spi/cluster/redis/impl/ITSubscriptionCatalog.java

@@ -7,6 +7,7 @@
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.timeout;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
@@ -15,9 +16,13 @@
 import io.vertx.core.spi.cluster.NodeSelector;
 import io.vertx.core.spi.cluster.RegistrationInfo;
 import io.vertx.core.spi.cluster.RegistrationUpdateEvent;
+import java.util.List;
+import java.util.Map;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.redisson.Redisson;
+import org.redisson.api.RSet;
+import org.redisson.api.RSetMultimap;
 import org.redisson.api.RedissonClient;
 import org.redisson.config.Config;
 import org.testcontainers.containers.GenericContainer;
@@ -32,6 +37,7 @@ class ITSubscriptionCatalog {
   private final RedisKeyFactory keyFactory = new RedisKeyFactory("test");
   private NodeSelector nodeSelector;
   private SubscriptionCatalog subsCatalog;
+  private RedissonClient redisson;
 
   @BeforeEach
   void beforeEach() {
@@ -40,7 +46,7 @@ void beforeEach() {
     String redisUrl = "redis://" + redis.getHost() + ":" + redis.getFirstMappedPort();
     Config config = new Config();
     config.useSingleServer().setAddress(redisUrl);
-    RedissonClient redisson = Redisson.create(config);
+    redisson = Redisson.create(config);
     subsCatalog = new SubscriptionCatalog(redisson, keyFactory, nodeSelector);
     when(nodeSelector.wantsUpdatesFor(anyString())).thenReturn(true);
   }
@@ -110,6 +116,35 @@ void removeUnknownSubsNoOp() {
     assertThat(subsCatalog.get("sub-2")).hasSize(1);
   }
 
+  @Test
+  void removeUnknownSubsBulkException() {
+    putSubs();
+    subsCatalog.put("sub-1", new RegistrationInfo("node3", 4, false));
+    subsCatalog.put("sub-1", new RegistrationInfo("node4", 5, false));
+
+    List<RegistrationInfo> subs1 = List.of(new RegistrationInfo("node1", 1, false));
+    List<RegistrationInfo> subs2 = List.of(new RegistrationInfo("node1", 2, false));
+
+    // Spy and mock the subs map to throw exception
+    RSetMultimap<String, RegistrationInfo> subsMap =
+        spy(redisson.getSetMultimap(keyFactory.vertx("subs")));
+    @SuppressWarnings("unchecked")
+    RSet<RegistrationInfo> setMock = mock(RSet.class);
+    when(subsMap.get("sub-1")).thenReturn(setMock);
+    when(setMock.removeAll(subs1)).thenThrow(new RuntimeException("TEST"));
+
+    SubscriptionCatalog.bulkRemoveUnknownSubsByKey(Map.of("sub-1", subs1, "sub-2", subs2), subsMap);
+
+    assertThat(subsCatalog.get("sub-1"))
+        .containsOnly(
+            new RegistrationInfo("node2", 3, false),
+            new RegistrationInfo("node3", 4, false),
+            new RegistrationInfo("node4", 5, false));
+    assertThat(subsCatalog.get("sub-2")).isEmpty();
+    verify(nodeSelector, timeout(100).atLeast(1))
+        .registrationsUpdated(any(RegistrationUpdateEvent.class));
+  }
+
   @Test
   void removeForAllNodes() {
     putSubs();