Fixed - different topics subscribed to the same Redis node in Cluster.

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

@@ -16,6 +16,7 @@
 package org.redisson.pubsub;
 
 import java.util.Collection;
+import java.util.LinkedList;
 import java.util.Queue;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentLinkedQueue;
@@ -65,7 +66,9 @@ public class PublishSubscribeService {
 
     private final ConcurrentMap<ChannelName, PubSubConnectionEntry> name2PubSubConnection = new ConcurrentHashMap<>();
 
-    private final Queue<PubSubConnectionEntry> freePubSubConnections = new ConcurrentLinkedQueue<>();
+    private final ConcurrentMap<MasterSlaveEntry, Queue<PubSubConnectionEntry>> freePubSubMap = new ConcurrentHashMap<>();
+
+    private final Queue<PubSubConnectionEntry> emptyQueue = new LinkedList<>();
 
     private final SemaphorePubSub semaphorePubSub = new SemaphorePubSub(this);
 
@@ -171,6 +174,8 @@ public class PublishSubscribeService {
                     return;
                 }
 
+                Queue<PubSubConnectionEntry> freePubSubConnections = getConnectionsQueue(channelName);
+
                 PubSubConnectionEntry freeEntry = freePubSubConnections.peek();
                 if (freeEntry == null) {
                     connect(codec, channelName, promise, type, lock, listeners);
@@ -228,6 +233,12 @@ public class PublishSubscribeService {
         });
     }
 
+    private Queue<PubSubConnectionEntry> getConnectionsQueue(ChannelName channelName) {
+        int slot = connectionManager.calcSlot(channelName.getName());
+        MasterSlaveEntry entry = connectionManager.getEntry(slot);
+        return freePubSubMap.getOrDefault(entry, emptyQueue);
+    }
+
     private RFuture<Void> addListeners(ChannelName channelName, RPromise<PubSubConnectionEntry> promise,
             PubSubType type, AsyncSemaphore lock, PubSubConnectionEntry connEntry,
             RedisPubSubListener<?>... listeners) {
@@ -282,11 +293,11 @@ public class PublishSubscribeService {
                 ((RPromise<RedisPubSubConnection>) connFuture).tryFailure(e);
             }
         });
-        connFuture.onComplete((conn, e) -> {
-            if (e != null) {
+        connFuture.onComplete((conn, ex) -> {
+            if (ex != null) {
                 freePubSubLock.release();
                 lock.release();
-                promise.tryFailure(e);
+                promise.tryFailure(ex);
                 return;
             }
 
@@ -304,7 +315,7 @@ public class PublishSubscribeService {
             }
 
             if (remainFreeAmount > 0) {
-                freePubSubConnections.add(entry);
+                addFreeConnectionEntry(channelName, entry);
             }
             freePubSubLock.release();
 
@@ -355,7 +366,7 @@ public class PublishSubscribeService {
                     executed.set(true);
 
                     if (entry.release() == 1) {
-                        freePubSubConnections.add(entry);
+                        addFreeConnectionEntry(channelName, entry);
                     }
 
                     lock.release();
@@ -409,6 +420,7 @@ public class PublishSubscribeService {
                 freePubSubLock.acquire(new Runnable() {
                     @Override
                     public void run() {
+                        Queue<PubSubConnectionEntry> freePubSubConnections = getConnectionsQueue(channelName);
                         freePubSubConnections.remove(entry);
                         freePubSubLock.release();
 
@@ -483,7 +495,7 @@ public class PublishSubscribeService {
                 if (type == PubSubType.PUNSUBSCRIBE && channel.equals(channelName)) {
 
                     if (entry.release() == 1) {
-                        freePubSubConnections.add(entry);
+                        addFreeConnectionEntry(channelName, entry);
                     }
 
                     lock.release();
@@ -495,19 +507,28 @@ public class PublishSubscribeService {
         });
     }
 
+    private void addFreeConnectionEntry(ChannelName channelName, PubSubConnectionEntry entry) {
+        int slot = connectionManager.calcSlot(channelName.getName());
+        MasterSlaveEntry me = connectionManager.getEntry(slot);
+        Queue<PubSubConnectionEntry> freePubSubConnections = freePubSubMap.computeIfAbsent(me, e -> new ConcurrentLinkedQueue<>());
+        freePubSubConnections.add(entry);
+    }
+
     public void reattachPubSub(RedisPubSubConnection redisPubSubConnection) {
-        for (PubSubConnectionEntry entry : freePubSubConnections) {
+        for (Queue<PubSubConnectionEntry> queue : freePubSubMap.values()) {
+            for (PubSubConnectionEntry entry : queue) {
                 if (entry.getConnection().equals(redisPubSubConnection)) {
                     freePubSubLock.acquire(new Runnable() {
                         @Override
                         public void run() {
-                        freePubSubConnections.remove(entry);
+                            queue.remove(entry);
                             freePubSubLock.release();
                         }
                     });
                     break;
                 }
             }
+        }
 
         for (ChannelName channelName : redisPubSubConnection.getChannels().keySet()) {
             PubSubConnectionEntry pubSubEntry = getPubSubEntry(channelName);
@@ -569,12 +590,7 @@ public class PublishSubscribeService {
 
     @Override
     public String toString() {
-        return "PublishSubscribeService [name2PubSubConnection=" + name2PubSubConnection + ", freePubSubConnections="
-                + freePubSubConnections + "]";
+        return "PublishSubscribeService [name2PubSubConnection=" + name2PubSubConnection + ", freePubSubMap=" + freePubSubMap + "]";
     }
 
-    
-    
-    
-    
 }

redisson/src/test/java/org/redisson/RedissonTopicTest.java

@@ -22,6 +22,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 
+import org.awaitility.Awaitility;
 import org.awaitility.Duration;
 import org.junit.After;
 import org.junit.AfterClass;
@@ -1138,6 +1139,48 @@ public class RedissonTopicTest {
         return t;
     }
 
+    @Test
+    public void testClusterSharding() throws IOException, InterruptedException {
+        RedisRunner master1 = new RedisRunner().randomPort().randomDir().nosave();
+        RedisRunner master2 = new RedisRunner().randomPort().randomDir().nosave();
+        RedisRunner master3 = new RedisRunner().randomPort().randomDir().nosave();
+        RedisRunner slave1 = new RedisRunner().randomPort().randomDir().nosave();
+        RedisRunner slave2 = new RedisRunner().randomPort().randomDir().nosave();
+        RedisRunner slave3 = new RedisRunner().randomPort().randomDir().nosave();
+
+        ClusterRunner clusterRunner = new ClusterRunner()
+                .addNode(master1, slave1)
+                .addNode(master2, slave2)
+                .addNode(master3, slave3);
+        ClusterProcesses process = clusterRunner.run();
+
+        Config config = new Config();
+        config.useClusterServers()
+        .setLoadBalancer(new RandomLoadBalancer())
+        .addNodeAddress(process.getNodes().stream().findAny().get().getRedisServerAddressAndPort());
+        RedissonClient redisson = Redisson.create(config);
+
+        AtomicInteger counter = new AtomicInteger();
+        for (int i = 0; i < 10; i++) {
+            int j = i;
+            RTopic topic = redisson.getTopic("test" + i);
+            topic.addListener(Integer.class, (c, v) -> {
+                assertThat(v).isEqualTo(j);
+                counter.incrementAndGet();
+            });
+        }
+
+        for (int i = 0; i < 10; i++) {
+            RTopic topic = redisson.getTopic("test" + i);
+            topic.publish(i);
+        }
+
+        Awaitility.await().atMost(Duration.FIVE_SECONDS).until(() -> counter.get() == 10);
+
+        redisson.shutdown();
+        process.shutdown();
+    }
+
     @Test
     public void testReattachInClusterSlave() throws Exception {
         RedisRunner master1 = new RedisRunner().randomPort().randomDir().nosave();