RKeys.deleteByPattern throws an error in cluster mode. Fixed

src/main/java/org/redisson/RedissonKeys.java

@@ -152,103 +152,81 @@ public Future<String> randomKeyAsync() {
         return commandExecutor.readRandomAsync(RedisCommands.RANDOM_KEY);
     }
 
-    /**
-     * Find keys by key search pattern
-     *
-     *  Supported glob-style patterns:
-     *    h?llo subscribes to hello, hallo and hxllo
-     *    h*llo subscribes to hllo and heeeello
-     *    h[ae]llo subscribes to hello and hallo, but not hillo
-     *
-     * @param pattern
-     * @return
-     */
     @Override
     public Collection<String> findKeysByPattern(String pattern) {
         return commandExecutor.get(findKeysByPatternAsync(pattern));
     }
 
-    /**
-     * Find keys by key search pattern in async mode
-     *
-     *  Supported glob-style patterns:
-     *    h?llo subscribes to hello, hallo and hxllo
-     *    h*llo subscribes to hllo and heeeello
-     *    h[ae]llo subscribes to hello and hallo, but not hillo
-     *
-     * @param pattern
-     * @return
-     */
     @Override
     public Future<Collection<String>> findKeysByPatternAsync(String pattern) {
         return commandExecutor.readAllAsync(RedisCommands.KEYS, pattern);
     }
 
-    /**
-     * Delete multiple objects by a key pattern
-     *
-     *  Supported glob-style patterns:
-     *    h?llo subscribes to hello, hallo and hxllo
-     *    h*llo subscribes to hllo and heeeello
-     *    h[ae]llo subscribes to hello and hallo, but not hillo
-     *
-     * @param pattern
-     * @return
-     */
     @Override
     public long deleteByPattern(String pattern) {
         return commandExecutor.get(deleteByPatternAsync(pattern));
     }
 
-    /**
-     * Delete multiple objects by a key pattern in async mode
-     *
-     *  Supported glob-style patterns:
-     *    h?llo subscribes to hello, hallo and hxllo
-     *    h*llo subscribes to hllo and heeeello
-     *    h[ae]llo subscribes to hello and hallo, but not hillo
-     *
-     * @param pattern
-     * @return
-     */
     @Override
     public Future<Long> deleteByPatternAsync(String pattern) {
-        return commandExecutor.evalWriteAllAsync(RedisCommands.EVAL_LONG, new SlotCallback<Long, Long>() {
-            AtomicLong results = new AtomicLong();
-            @Override
-            public void onSlotResult(Long result) {
-                results.addAndGet(result);
-            }
+        if (!commandExecutor.getConnectionManager().isClusterMode()) {
+            return commandExecutor.evalWriteAsync((String)null, null, RedisCommands.EVAL_LONG, "local keys = redis.call('keys', ARGV[1]) "
+                              + "local n = 0 "
+                              + "redis.log(redis.LOG_WARNING, 'keys number ' .. #keys); "
+                              + "for i=1, #keys,5000 do "
+                                  + "n = n + redis.call('del', unpack(keys, i, math.min(i+4999, table.getn(keys)))) "
+                              + "end "
+                          + "return n;",Collections.emptyList(), pattern);
+        }
 
+        final Promise<Long> result = commandExecutor.getConnectionManager().newPromise();
+        final AtomicReference<Throwable> failed = new AtomicReference<Throwable>();
+        final AtomicLong count = new AtomicLong();
+        final AtomicLong executed = new AtomicLong(commandExecutor.getConnectionManager().getEntries().size());
+        final FutureListener<Long> listener = new FutureListener<Long>() {
             @Override
-            public Long onFinish() {
-                return results.get();
+            public void operationComplete(Future<Long> future) throws Exception {
+                if (future.isSuccess()) {
+                    count.addAndGet(future.getNow());
+                } else {
+                    failed.set(future.cause());
+                }
+
+                checkExecution(result, failed, count, executed);
             }
-        }, "local keys = redis.call('keys', ARGV[1]) "
-                + "local n = 0 "
-                + "for i=1, table.getn(keys),5000 do "
-                    + "n = n + redis.call('del', unpack(keys, i, math.min(i+4999, table.getn(keys)))) "
-                + "end "
-            + "return n;",Collections.emptyList(), pattern);
+        };
+
+        for (ClusterSlotRange slot : commandExecutor.getConnectionManager().getEntries().keySet()) {
+            Future<Collection<String>> findFuture = commandExecutor.readAsync(slot.getStartSlot(), null, RedisCommands.KEYS, pattern);
+            findFuture.addListener(new FutureListener<Collection<String>>() {
+                @Override
+                public void operationComplete(Future<Collection<String>> future) throws Exception {
+                    if (!future.isSuccess()) {
+                        failed.set(future.cause());
+                        checkExecution(result, failed, count, executed);
+                        return;
+                    }
+
+                    Collection<String> keys = future.getNow();
+                    if (keys.isEmpty()) {
+                        checkExecution(result, failed, count, executed);
+                        return;
+                    }
+
+                    Future<Long> deleteFuture = deleteAsync(keys.toArray(new String[keys.size()]));
+                    deleteFuture.addListener(listener);
+                }
+            });
+        }
+
+        return result;
     }
 
-    /**
-     * Delete multiple objects by name
-     *
-     * @param keys - object names
-     * @return number of removed keys
-     */
     @Override
     public long delete(String ... keys) {
         return commandExecutor.get(deleteAsync(keys));
     }
 
-    /**
-     * Delete multiple objects by name in async mode
-     *
-     * @param keys - object names
-     * @return number of removed keys
-     */
     @Override
     public Future<Long> deleteAsync(String ... keys) {
         if (!commandExecutor.getConnectionManager().isClusterMode()) {
@@ -286,18 +264,7 @@ public void operationComplete(Future<List<?>> future) throws Exception {
                     failed.set(future.cause());
                 }
 
-                if (executed.decrementAndGet() == 0) {
-                    if (failed.get() != null) {
-                        if (count.get() > 0) {
-                            RedisException ex = new RedisException("" + count.get() + " keys has been deleted. But one or more nodes has an error", failed.get());
-                            result.setFailure(ex);
-                        } else {
-                            result.setFailure(failed.get());
-                        }
-                    } else {
-                        result.setSuccess(count.get());
-                    }
-                }
+                checkExecution(result, failed, count, executed);
             }
         };
 
@@ -356,4 +323,20 @@ public Future<Void> flushallAsync() {
         return commandExecutor.writeAllAsync(RedisCommands.FLUSHALL);
     }
 
+    private void checkExecution(final Promise<Long> result, final AtomicReference<Throwable> failed,
+            final AtomicLong count, final AtomicLong executed) {
+        if (executed.decrementAndGet() == 0) {
+            if (failed.get() != null) {
+                if (count.get() > 0) {
+                    RedisException ex = new RedisException("" + count.get() + " keys has been deleted. But one or more nodes has an error", failed.get());
+                    result.setFailure(ex);
+                } else {
+                    result.setFailure(failed.get());
+                }
+            } else {
+                result.setSuccess(count.get());
+            }
+        }
+    }
+
 }

src/main/java/org/redisson/command/CommandAsyncExecutor.java

@@ -54,8 +54,12 @@ public interface CommandAsyncExecutor {
 
     <T, R> Future<R> evalReadAsync(String key, Codec codec, RedisCommand<T> evalCommandType, String script, List<Object> keys, Object ... params);
 
+    <T, R> Future<R> evalReadAsync(Integer slot, Codec codec, RedisCommand<T> evalCommandType, String script, List<Object> keys, Object ... params);
+
     <T, R> Future<R> evalWriteAsync(String key, Codec codec, RedisCommand<T> evalCommandType, String script, List<Object> keys, Object ... params);
 
+    <T, R> Future<R> evalWriteAsync(Integer slot, Codec codec, RedisCommand<T> evalCommandType, String script, List<Object> keys, Object ... params);
+
     <T, R> Future<R> readAsync(String key, Codec codec, RedisCommand<T> command, Object ... params);
 
     <T, R> Future<R> writeAsync(String key, Codec codec, RedisCommand<T> command, Object ... params);
@@ -68,6 +72,8 @@ public interface CommandAsyncExecutor {
 
     <T, R> Future<R> readAsync(String key, RedisCommand<T> command, Object ... params);
 
+    <T, R> Future<R> readAsync(Integer slot, Codec codec, RedisCommand<T> command, Object ... params);
+
     <T, R> Future<R> readRandomAsync(RedisCommand<T> command, Object ... params);
 
 

src/main/java/org/redisson/command/CommandAsyncService.java

@@ -240,6 +240,12 @@ public <T, R> Future<R> readAsync(String key, Codec codec, RedisCommand<T> comma
         return mainPromise;
     }
 
+    public <T, R> Future<R> readAsync(Integer slot, Codec codec, RedisCommand<T> command, Object ... params) {
+        Promise<R> mainPromise = connectionManager.newPromise();
+        async(true, new NodeSource(slot), codec, command, params, mainPromise, 0);
+        return mainPromise;
+    }
+
     @Override
     public <T, R> Future<R> writeAsync(Integer slot, Codec codec, RedisCommand<T> command, Object ... params) {
         Promise<R> mainPromise = connectionManager.newPromise();
@@ -258,6 +264,11 @@ public <T, R> Future<R> evalReadAsync(String key, Codec codec, RedisCommand<T> e
         return evalAsync(source, true, codec, evalCommandType, script, keys, params);
     }
 
+    @Override
+    public <T, R> Future<R> evalReadAsync(Integer slot, Codec codec, RedisCommand<T> evalCommandType, String script, List<Object> keys, Object ... params) {
+        return evalAsync(new NodeSource(slot), true, codec, evalCommandType, script, keys, params);
+    }
+
     @Override
     public <T, R> Future<R> evalReadAsync(InetSocketAddress client, String key, Codec codec, RedisCommand<T> evalCommandType, String script, List<Object> keys, Object ... params) {
         int slot = connectionManager.calcSlot(key);
@@ -270,6 +281,11 @@ public <T, R> Future<R> evalWriteAsync(String key, Codec codec, RedisCommand<T>
         return evalAsync(source, false, codec, evalCommandType, script, keys, params);
     }
 
+    public <T, R> Future<R> evalWriteAsync(Integer slot, Codec codec, RedisCommand<T> evalCommandType, String script, List<Object> keys, Object ... params) {
+        return evalAsync(new NodeSource(slot), false, codec, evalCommandType, script, keys, params);
+    }
+
+
     @Override
     public <T, R> Future<R> evalWriteAllAsync(RedisCommand<T> command, SlotCallback<T, R> callback, String script, List<Object> keys, Object ... params) {
         return evalAllAsync(false, command, callback, script, keys, params);

src/main/java/org/redisson/core/RKeys.java

@@ -68,23 +68,25 @@ public interface RKeys extends RKeysAsync {
     Collection<String> findKeysByPattern(String pattern);
 
     /**
-     * Delete multiple objects by a key pattern
-     *
+     * Delete multiple objects by a key pattern.
+     * <p/>
+     * Method executes in <b>NON atomic way</b> in cluster mode due to lua script limitations.
+     * <p/>
      *  Supported glob-style patterns:
      *    h?llo subscribes to hello, hallo and hxllo
      *    h*llo subscribes to hllo and heeeello
      *    h[ae]llo subscribes to hello and hallo, but not hillo
      *
      * @param pattern
-     * @return
+     * @return number of removed keys
      */
     long deleteByPattern(String pattern);
 
     /**
      * Delete multiple objects by name
      *
      * @param keys - object names
-     * @return
+     * @return number of removed keys
      */
     long delete(String ... keys);
 

src/main/java/org/redisson/core/RKeysAsync.java

@@ -51,23 +51,25 @@ public interface RKeysAsync {
     Future<Collection<String>> findKeysByPatternAsync(String pattern);
 
     /**
-     * Delete multiple objects by a key pattern in async mode
-     *
+     * Delete multiple objects by a key pattern.
+     * <p/>
+     * Method executes in <b>NON atomic way</b> in cluster mode due to lua script limitations.
+     * <p/>
      *  Supported glob-style patterns:
      *    h?llo subscribes to hello, hallo and hxllo
      *    h*llo subscribes to hllo and heeeello
      *    h[ae]llo subscribes to hello and hallo, but not hillo
      *
      * @param pattern
-     * @return
+     * @return number of removed keys
      */
     Future<Long> deleteByPatternAsync(String pattern);
 
     /**
-     * Delete multiple objects by name in async mode
+     * Delete multiple objects by name
      *
      * @param keys - object names
-     * @return
+     * @return number of removed keys
      */
     Future<Long> deleteAsync(String ... keys);
 

src/test/java/org/redisson/RedissonKeysTest.java

@@ -66,12 +66,25 @@ public void testRandomKey() {
 
     @Test
     public void testDeleteByPattern() {
-        RBucket<String> bucket = redisson.getBucket("test1");
-        bucket.set("someValue");
+        RBucket<String> bucket = redisson.getBucket("test0");
+        bucket.set("someValue3");
+        assertThat(bucket.isExists()).isTrue();
+
+        RBucket<String> bucket2 = redisson.getBucket("test9");
+        bucket2.set("someValue4");
+        assertThat(bucket.isExists()).isTrue();
+
         RMap<String, String> map = redisson.getMap("test2");
         map.fastPut("1", "2");
+        assertThat(map.isExists()).isTrue();
 
-        Assert.assertEquals(2, redisson.getKeys().deleteByPattern("test?"));
+        RMap<String, String> map2 = redisson.getMap("test3");
+        map2.fastPut("1", "5");
+        assertThat(map2.isExists()).isTrue();
+
+
+        Assert.assertEquals(4, redisson.getKeys().deleteByPattern("test?"));
+        Assert.assertEquals(0, redisson.getKeys().deleteByPattern("test?"));
     }
 
     @Test
@@ -90,13 +103,24 @@ public void testFindKeys() {
 
     @Test
     public void testMassDelete() {
-        RBucket<String> bucket = redisson.getBucket("test");
-        bucket.set("someValue");
+        RBucket<String> bucket0 = redisson.getBucket("test0");
+        bucket0.set("someValue");
+        RBucket<String> bucket1 = redisson.getBucket("test1");
+        bucket1.set("someValue");
+        RBucket<String> bucket2 = redisson.getBucket("test2");
+        bucket2.set("someValue");
+        RBucket<String> bucket3 = redisson.getBucket("test3");
+        bucket3.set("someValue");
+        RBucket<String> bucket10 = redisson.getBucket("test10");
+        bucket10.set("someValue");
+
+        RBucket<String> bucket12 = redisson.getBucket("test12");
+        bucket12.set("someValue");
         RMap<String, String> map = redisson.getMap("map2");
         map.fastPut("1", "2");
 
-        Assert.assertEquals(2, redisson.getKeys().delete("test", "map2"));
-        Assert.assertEquals(0, redisson.getKeys().delete("test", "map2"));
+        Assert.assertEquals(7, redisson.getKeys().delete("test0", "test1", "test2", "test3", "test10", "test12", "map2"));
+        Assert.assertEquals(0, redisson.getKeys().delete("test0", "test1", "test2", "test3", "test10", "test12", "map2"));
     }
 
     @Test