Consider ReadFrom.isOrderSensitive() in cluster scan command #1146

ReadFrom order sensitivity is now considered in Cluster SCAN for keys.
redis · Oct 16, 2019 · a8afab9 · a8afab9
1 parent 63f79fd
commit a8afab9
Show file tree

Hide file tree

Showing 2 changed files with 22 additions and 1 deletion.
diff --git a/src/main/java/io/lettuce/core/cluster/ClusterScanSupport.java b/src/main/java/io/lettuce/core/cluster/ClusterScanSupport.java
@@ -18,6 +18,7 @@
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
+import java.util.concurrent.ThreadLocalRandom;
 import java.util.function.Function;
 
 import reactor.core.publisher.Mono;
@@ -158,7 +159,13 @@ public Iterator<RedisNodeDescription> iterator() {
                 });
 
                 if (!selection.isEmpty()) {
-                    RedisClusterNode selectedNode = (RedisClusterNode) selection.get(0);
+
+                    int indexToUse = 0;
+                    if (!OrderingReadFromAccessor.isOrderSensitive(connection.getReadFrom())) {
+                        indexToUse = ThreadLocalRandom.current().nextInt(selection.size());
+                    }
+
+                    RedisClusterNode selectedNode = (RedisClusterNode) selection.get(indexToUse);
                     nodeIds.add(selectedNode.getNodeId());
                     continue;
                 }

diff --git a/src/test/java/io/lettuce/core/cluster/ScanIteratorIntegrationTests.java b/src/test/java/io/lettuce/core/cluster/ScanIteratorIntegrationTests.java
@@ -49,6 +49,7 @@ class ScanIteratorIntegrationTests extends TestSupport {
     @Inject
     ScanIteratorIntegrationTests(StatefulRedisClusterConnection<String, String> connection) {
         this.connection = connection;
+        this.connection.setReadFrom(ReadFrom.MASTER);
         this.redis = connection.sync();
     }
 
@@ -103,6 +104,19 @@ void keysMultiPass() {
         assertThat(keys).containsAll(KeysAndValues.KEYS);
     }
 
+    @Test
+    void keysMultiPassFromAnyNode() {
+
+        redis.mset(KeysAndValues.MAP);
+        this.connection.setReadFrom(ReadFrom.ANY);
+
+        ScanIterator<String> scan = ScanIterator.scan(redis);
+
+        List<String> keys = scan.stream().collect(Collectors.toList());
+
+        assertThat(keys).containsAll(KeysAndValues.KEYS);
+    }
+
     @Test
     void hscanShouldThrowNoSuchElementExceptionOnEmpty() {