Feature - shardedSubscriptionMode setting added in Cluster configuration. #5690

redisson/src/main/java/org/redisson/cluster/ClusterConnectionManager.java

@@ -169,7 +169,11 @@ public class ClusterConnectionManager extends MasterSlaveConnectionManager {
             }
         }
 
+        if (cfg.getShardedSubscriptionMode() == ShardedSubscriptionMode.AUTO) {
             subscribeService.detectSharding();
+        } else if (cfg.getShardedSubscriptionMode() == ShardedSubscriptionMode.ON) {
+            subscribeService.setShardingSupported(true);
+        }
         scheduleClusterChangeCheck(cfg);
     }
 

redisson/src/main/java/org/redisson/config/ClusterServersConfig.java

@@ -45,6 +45,8 @@ public class ClusterServersConfig extends BaseMasterSlaveServersConfig<ClusterSe
 
     private boolean checkSlotsCoverage = true;
 
+    private ShardedSubscriptionMode shardedSubscriptionMode = ShardedSubscriptionMode.AUTO;
+
     public ClusterServersConfig() {
     }
 
@@ -54,6 +56,7 @@ public class ClusterServersConfig extends BaseMasterSlaveServersConfig<ClusterSe
         setScanInterval(config.getScanInterval());
         setNatMapper(config.getNatMapper());
         setCheckSlotsCoverage(config.isCheckSlotsCoverage());
+        setShardedSubscriptionMode(config.getShardedSubscriptionMode());
     }
 
     /**
@@ -98,7 +101,7 @@ public class ClusterServersConfig extends BaseMasterSlaveServersConfig<ClusterSe
      * <p>
      * Default is <code>true</code>
      *
-     * @param checkSlotsCoverage - boolean value
+     * @param checkSlotsCoverage boolean value
      * @return config
      */
     public ClusterServersConfig setCheckSlotsCoverage(boolean checkSlotsCoverage) {
@@ -128,7 +131,7 @@ public class ClusterServersConfig extends BaseMasterSlaveServersConfig<ClusterSe
      * @see HostNatMapper
      * @see HostPortNatMapper
      *
-     * @param natMapper - nat mapper object
+     * @param natMapper nat mapper object
      * @return config
      */
     public ClusterServersConfig setNatMapper(NatMapper natMapper) {
@@ -136,5 +139,20 @@ public class ClusterServersConfig extends BaseMasterSlaveServersConfig<ClusterSe
         return this;
     }
 
+    public ShardedSubscriptionMode getShardedSubscriptionMode() {
+        return shardedSubscriptionMode;
+    }
 
+    /**
+     * Defines detection of sharded subscription feature available in Redis 7.0+
+     * <p>
+     * Default is <code>AUTO</code>
+     *
+     * @param shardedSubscriptionMode param
+     * @return config
+     */
+    public ClusterServersConfig setShardedSubscriptionMode(ShardedSubscriptionMode shardedSubscriptionMode) {
+        this.shardedSubscriptionMode = shardedSubscriptionMode;
+        return this;
+    }
 }

redisson/src/main/java/org/redisson/config/ShardedSubscriptionMode.java

@@ -0,0 +1,37 @@
+/**
+ * Copyright (c) 2013-2024 Nikita Koksharov
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.redisson.config;
+
+/**
+ *
+ * @author Nikita Koksharov
+ *
+ */
+public enum ShardedSubscriptionMode {
+
+    /**
+     * Subscribe to slave nodes
+     */
+    AUTO,
+
+    /**
+     * Subscribe to master node
+     */
+    ON,
+
+    OFF
+
+}

redisson/src/main/java/org/redisson/pubsub/PublishSubscribeService.java

@@ -1015,7 +1015,7 @@ public class PublishSubscribeService {
         RedisConnection c = entry.connectionWriteOp(null).join();
         try {
             c.sync(RedisCommands.PUBSUB_SHARDNUMSUB);
-            shardingSupported = true;
+            setShardingSupported(true);
         } catch (Exception e) {
             // skip
         } finally {
@@ -1023,6 +1023,10 @@ public class PublishSubscribeService {
         }
     }
 
+    public void setShardingSupported(boolean shardingSupported) {
+        this.shardingSupported = shardingSupported;
+    }
+
     public boolean isShardingSupported() {
         return shardingSupported;
     }