Retrieving all keys list from a JedisCluster not working

Dasun Nirmitha

unread,

Jan 26, 2016, 7:17:03 AM1/26/16

to Jedis

Hello,

I'm using Jedis v2.8.0 for a project using the Play framework and I'm trying to use the following code to retrieve all the keys within a JedisCluster.

private Set<byte[]> getClusterKeyList(byte[] keyPatternByteArr) throws IOException {

TreeSet<byte[]> keys = new TreeSet<>();

Map<String, JedisPool> clusterNodes = jedisCluster.getClusterNodes();

for(String key : clusterNodes.keySet()) {

JedisPool jedisPool = clusterNodes.get(key);

Jedis nodeConnection = jedisPool.getResource();

keys.addAll(nodeConnection.keys(keyPatternByteArr));

nodeConnection.close();

}

return keys;

}

Running this method throws a java.net.SocketTimeoutException: connect timed out error when it runs the jedisPool.getResource() method.

Here's the stacktrace printed.

play.api.http.HttpErrorHandlerExceptions$$anon$1: Execution exception[[JedisConnectionException: Could not get a resource from the pool]]

at play.api.http.HttpErrorHandlerExceptions$.throwableToUsefulException(HttpErrorHandler.scala:265) ~[play_2.11-2.4.6.jar:2.4.6]

at play.api.http.DefaultHttpErrorHandler.onServerError(HttpErrorHandler.scala:191) ~[play_2.11-2.4.6.jar:2.4.6]

at play.api.GlobalSettings$class.onError(GlobalSettings.scala:179) [play_2.11-2.4.6.jar:2.4.6]

at play.api.DefaultGlobal$.onError(GlobalSettings.scala:212) [play_2.11-2.4.6.jar:2.4.6]

at play.api.http.GlobalSettingsHttpErrorHandler.onServerError(HttpErrorHandler.scala:94) [play_2.11-2.4.6.jar:2.4.6]

at play.core.server.netty.PlayDefaultUpstreamHandler$$anonfun$9$$anonfun$apply$1.applyOrElse(PlayDefaultUpstreamHandler.scala:151) [play-netty-server_2.11-2.4.6.jar:2.4.6]

at play.core.server.netty.PlayDefaultUpstreamHandler$$anonfun$9$$anonfun$apply$1.applyOrElse(PlayDefaultUpstreamHandler.scala:148) [play-netty-server_2.11-2.4.6.jar:2.4.6]

at scala.runtime.AbstractPartialFunction.apply(AbstractPartialFunction.scala:36) [scala-library-2.11.6.jar:na]

at scala.util.Failure$$anonfun$recover$1.apply(Try.scala:215) [scala-library-2.11.6.jar:na]

at scala.util.Try$.apply(Try.scala:191) [scala-library-2.11.6.jar:na]

at scala.util.Failure.recover(Try.scala:215) [scala-library-2.11.6.jar:na]

at scala.concurrent.Future$$anonfun$recover$1.apply(Future.scala:324) [scala-library-2.11.6.jar:na]

at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:32) [scala-library-2.11.6.jar:na]

at play.api.libs.iteratee.Execution$trampoline$.executeScheduled(Execution.scala:109) [play-iteratees_2.11-2.4.6.jar:2.4.6]

at play.api.libs.iteratee.Execution$trampoline$.execute(Execution.scala:71) [play-iteratees_2.11-2.4.6.jar:2.4.6]

at scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:40) [scala-library-2.11.6.jar:na]

at scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:248) [scala-library-2.11.6.jar:na]

at scala.concurrent.Promise$class.complete(Promise.scala:55) [scala-library-2.11.6.jar:na]

at scala.concurrent.impl.Promise$DefaultPromise.complete(Promise.scala:153) [scala-library-2.11.6.jar:na]

at scala.concurrent.impl.Future$PromiseCompletingRunnable.run(Future.scala:23) [scala-library-2.11.6.jar:na]

at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:40) [akka-actor_2.11-2.3.13.jar:na]

at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:397) [akka-actor_2.11-2.3.13.jar:na]

at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260) [scala-library-2.11.6.jar:na]

at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339) [scala-library-2.11.6.jar:na]

at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979) [scala-library-2.11.6.jar:na]

at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107) [scala-library-2.11.6.jar:na]

Caused by: redis.clients.jedis.exceptions.JedisConnectionException: Could not get a resource from the pool

at redis.clients.util.Pool.getResource(Pool.java:50) ~[jedis-2.8.0.jar:na]

at redis.clients.jedis.JedisPool.getResource(JedisPool.java:99) ~[jedis-2.8.0.jar:na]

at cache.JedisCache.getClusterKeyList(JedisCache.java:369) ~[classes/:na]

at cache.JedisCache.deleteAllRecords(JedisCache.java:253) ~[classes/:na]

at controllers.Application.index(Application.java:51) ~[classes/:na]

at router.Routes$$anonfun$routes$1$$anonfun$applyOrElse$1$$anonfun$apply$1.apply(Routes.scala:95) ~[classes/:na]

at play.core.routing.HandlerInvokerFactory$$anon$4.resultCall(HandlerInvoker.scala:136) ~[play_2.11-2.4.6.jar:2.4.6]

at play.core.routing.HandlerInvokerFactory$JavaActionInvokerFactory$$anon$14$$anon$3$$anon$1.invocation(HandlerInvoker.scala:127) ~[play_2.11-2.4.6.jar:2.4.6]

at play.core.j.JavaAction$$anon$1.call(JavaAction.scala:70) ~[play_2.11-2.4.6.jar:2.4.6]

at play.http.DefaultHttpRequestHandler$1.call(DefaultHttpRequestHandler.java:20) ~[play_2.11-2.4.6.jar:2.4.6]

at play.core.j.JavaAction$$anonfun$7.apply(JavaAction.scala:94) ~[play_2.11-2.4.6.jar:2.4.6]

at scala.concurrent.impl.Future$PromiseCompletingRunnable.liftedTree1$1(Future.scala:24) [scala-library-2.11.6.jar:na]

at scala.concurrent.impl.Future$PromiseCompletingRunnable.run(Future.scala:24) [scala-library-2.11.6.jar:na]

at play.core.j.HttpExecutionContext$$anon$2.run(HttpExecutionContext.scala:40) ~[play_2.11-2.4.6.jar:2.4.6]

at play.api.libs.iteratee.Execution$trampoline$.execute(Execution.scala:70) [play-iteratees_2.11-2.4.6.jar:2.4.6]

at play.core.j.HttpExecutionContext.execute(HttpExecutionContext.scala:32) ~[play_2.11-2.4.6.jar:2.4.6]

at scala.concurrent.impl.Future$.apply(Future.scala:31) ~[scala-library-2.11.6.jar:na]

at scala.concurrent.Future$.apply(Future.scala:492) ~[scala-library-2.11.6.jar:na]

at play.core.j.JavaAction.apply(JavaAction.scala:94) ~[play_2.11-2.4.6.jar:2.4.6]

at play.api.mvc.Action$$anonfun$apply$1$$anonfun$apply$4$$anonfun$apply$5.apply(Action.scala:105) ~[play_2.11-2.4.6.jar:2.4.6]

at play.utils.Threads$.withContextClassLoader(Threads.scala:21) ~[play_2.11-2.4.6.jar:2.4.6]

at play.api.mvc.Action$$anonfun$apply$1$$anonfun$apply$4.apply(Action.scala:104) ~[play_2.11-2.4.6.jar:2.4.6]

at play.api.mvc.Action$$anonfun$apply$1$$anonfun$apply$4.apply(Action.scala:103) ~[play_2.11-2.4.6.jar:2.4.6]

at scala.Option.map(Option.scala:146) ~[scala-library-2.11.6.jar:na]

at play.api.mvc.Action$$anonfun$apply$1.apply(Action.scala:103) ~[play_2.11-2.4.6.jar:2.4.6]

at play.api.mvc.Action$$anonfun$apply$1.apply(Action.scala:96) ~[play_2.11-2.4.6.jar:2.4.6]

at play.api.libs.iteratee.Iteratee$$anonfun$mapM$1.apply(Iteratee.scala:524) ~[play-iteratees_2.11-2.4.6.jar:2.4.6]

at play.api.libs.iteratee.Iteratee$$anonfun$flatMapM$1.apply(Iteratee.scala:560) ~[play-iteratees_2.11-2.4.6.jar:2.4.6]

at play.api.libs.iteratee.Iteratee$$anonfun$flatMap$1$$anonfun$apply$13.apply(Iteratee.scala:536) ~[play-iteratees_2.11-2.4.6.jar:2.4.6]

at scala.concurrent.impl.Future$PromiseCompletingRunnable.liftedTree1$1(Future.scala:24) [scala-library-2.11.6.jar:na]

at scala.concurrent.impl.Future$PromiseCompletingRunnable.run(Future.scala:24) [scala-library-2.11.6.jar:na]

... 6 common frames omitted

Caused by: redis.clients.jedis.exceptions.JedisConnectionException: java.net.SocketTimeoutException: connect timed out

at redis.clients.jedis.Connection.connect(Connection.java:164) ~[jedis-2.8.0.jar:na]

at redis.clients.jedis.BinaryClient.connect(BinaryClient.java:80) ~[jedis-2.8.0.jar:na]

at redis.clients.jedis.BinaryJedis.connect(BinaryJedis.java:1676) ~[jedis-2.8.0.jar:na]

at redis.clients.jedis.JedisFactory.makeObject(JedisFactory.java:87) ~[jedis-2.8.0.jar:na]

at org.apache.commons.pool2.impl.GenericObjectPool.create(GenericObjectPool.java:861) ~[commons-pool2-2.3.jar:2.3]

at org.apache.commons.pool2.impl.GenericObjectPool.borrowObject(GenericObjectPool.java:435) ~[commons-pool2-2.3.jar:2.3]

at org.apache.commons.pool2.impl.GenericObjectPool.borrowObject(GenericObjectPool.java:363) ~[commons-pool2-2.3.jar:2.3]

at redis.clients.util.Pool.getResource(Pool.java:48) ~[jedis-2.8.0.jar:na]

... 42 common frames omitted

Caused by: java.net.SocketTimeoutException: connect timed out

at java.net.PlainSocketImpl.socketConnect(Native Method) ~[na:1.8.0_51]

at java.net.AbstractPlainSocketImpl.doConnect(AbstractPlainSocketImpl.java:345) ~[na:1.8.0_51]

at java.net.AbstractPlainSocketImpl.connectToAddress(AbstractPlainSocketImpl.java:206) ~[na:1.8.0_51]

at java.net.AbstractPlainSocketImpl.connect(AbstractPlainSocketImpl.java:188) ~[na:1.8.0_51]

at java.net.SocksSocketImpl.connect(SocksSocketImpl.java:392) ~[na:1.8.0_51]

at java.net.Socket.connect(Socket.java:589) ~[na:1.8.0_51]

at redis.clients.jedis.Connection.connect(Connection.java:158) ~[jedis-2.8.0.jar:na]

Please help if anyone knows the solution to this or another way of retrieving all the keys from the JedisCluster.

Thank You :)

Marcos Lilljedahl

unread,

Jan 26, 2016, 10:57:17 AM1/26/16

to Jedis

Dasun, for the moment this is the only way to retrieve all the keys from a redis cluster. Regarding your code exception, you need to make sure that the machine that's running your Java code can connect to the cluster master node directly and that the `keys` command (which is blocking) is not taking more than the configured timeout.

Marcos

--
You received this message because you are subscribed to the Google Groups "Jedis" group.
To unsubscribe from this group and stop receiving emails from it, send an email to jedis_redis...@googlegroups.com.
For more options, visit https://groups.google.com/d/optout.

Dasun Nirmitha

unread,

Jan 27, 2016, 7:10:16 AM1/27/16

to Jedis

Hello Marcos,

Thanks for the reply :)

I'm currently running 6 nodes running in CentOS VMs for the cluster. So that will rule out the connection issues right? I'm thinking the default timeout set to jedisPool might not be enough can that be the case? If it can be can you please show me how to change the timeout of the jedisPool as I couldn't find it anywhere on the net.

Thanks again!

Allan Wax

unread,

Jan 28, 2016, 2:53:35 PM1/28/16

to Jedis

Try using the following code instead. It also does not use 'keys' so it does not block for a long time

=====================

package com.findology.util.jediscluster;

import org.apache.log4j.ConsoleAppender;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import redis.clients.jedis.*;

import java.net.InetAddress;
import java.util.*;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by allan.wax on 7/27/2015.
 */
public abstract class ClusterScanner {
    private static final Logger log = Logger.getLogger(ClusterScanner.class);
    protected static final ExecutorService executor = Executors.newCachedThreadPool();
    protected static final HashMap<String, String> hostMap = new HashMap<>();
    private JedisCluster jedisCluster = null;

    private static boolean TEST_MODE = false;

    public ClusterScanner(JedisCluster jedisCluster) {
        this.jedisCluster = jedisCluster;
    }

    public abstract void action(String key);

    private String normalizeHostAndPort(String hostAndPort) {
        try {
            String hap = hostMap.get(hostAndPort);

            if (hap == null) {
                int colon = hostAndPort.lastIndexOf(':');
                InetAddress address = InetAddress.getByName(hostAndPort.substring(0, colon));
                hap = address.getHostAddress() + hostAndPort.substring(colon);
                hostMap.put(hostAndPort, hap);
            }

            return hap;
        }
        catch (Exception e) {
            return hostAndPort;
        }
    }

    private class ScannerImpl implements Callable<Long> {
        private String clusterInstance;
        private String match;

        public ScannerImpl(String clusterInstance, String match) {
            this.clusterInstance = clusterInstance;
            this.match = match;
        }

        /**
         * Computes a result, or throws an exception if unable to do so.
         *
         * @return computed result
         * @throws Exception if unable to compute a result
         */
        @Override
        public Long call() throws Exception {
            long count = 0; /* TEST_MODE */
            int scanCount = 0;

            try {
                String hostAndPort = normalizeHostAndPort(clusterInstance);
                String[] parts = hostAndPort.split(":");
                String host = parts[0];
                int port = Integer.valueOf(parts[1]);

                try (Jedis jedis = new Jedis(host, port)) {
                    ScanParams params = new ScanParams().match(match).count(100);
                    String scanMarker = "0";
                    ScanResult<String> results = null;

                    do {
                        results = jedis.scan(scanMarker, params);
                        scanCount++;
                        List<String> keys = results.getResult();
                        if (keys != null && keys.size() > 0) {
                            count += keys.size(); /* TEST_MODE */
                            for (String key : keys) {
                                action(key);
                            }
                        }
                        scanMarker = results.getStringCursor();
                    } while (!scanMarker.equals("0"));
                }

                if (TEST_MODE) {
                    log.info("Found " + count + " keys for " + hostAndPort + " in " + scanCount + " scans"); /* TEST_MODE */
                }
            }
            catch (Exception e) {
                log.error("" + e);
            }

            return count;
        }
    }

    public void scan(String match) {
        scan(match, 15);
    }

    public void scan(String match, int maxSeconds) {
        if (TEST_MODE) {
            log.info("Start scan for '" + match + "'");
        }

        Map<String, JedisPool> jedisPools = jedisCluster.getClusterNodes();

        String nodeList = null;
        Exception screwed = null;

        // get the list of nodes (masters and slaves)

        for (JedisPool pool : jedisPools.values()) {
            try {
                try (Jedis j = pool.getResource()) {
                    nodeList = j.clusterNodes();
                    break;
                }
                catch (Exception e1) {
                    screwed = e1;
                    continue;
                }
            }
            catch (Exception e) {
                // DO SOMETHING
            }
        }

        if (nodeList == null) {
            log.error("The cluster is screwed.  Can't use any members.", screwed);
            return;
        }

        String[] nodes = nodeList.split("\n");
        ArrayList<ScannerImpl> scanners = new ArrayList<>();

        // pick out the masters
        for (String node : nodes) {
            String[] info = node.split("\\s+");

            if (info[2].indexOf("fail") >= 0) {
                continue;
            }

            if (info[2].indexOf("handshake") >= 0) {
                continue;
            }

            if (info[2].indexOf("noaddr") >= 0) {
                continue;
            }

            if (info[2].indexOf("master") >= 0) {
                scanners.add(new ScannerImpl(info[1], match));
            }
        }

        if (!scanners.isEmpty()) {
            try {
                executor.invokeAll(scanners, maxSeconds, TimeUnit.SECONDS);
            }
            catch (Exception e) {
                log.error(e);
            }
        }
    }

    /* TEST */
    public static void main(String[] args) {
        TEST_MODE = true;

        Logger rootLogger = Logger.getRootLogger();
        if (!rootLogger.getAllAppenders().hasMoreElements()) {
            rootLogger.setLevel(Level.INFO);
            rootLogger.addAppender(new ConsoleAppender(new org.apache.log4j.PatternLayout("%d{yyyy-MM-dd HH:mm:ss.SSS} [%t; %C{1}] %-5p -- %m%n")));
        }

        Set<HostAndPort> jcNodes = new HashSet<>();
        jcNodes.add(new HostAndPort("test2", 18000));
        final JedisCluster jc = new JedisCluster(jcNodes);

        final AtomicInteger ai = new AtomicInteger(0);
        ClusterScanner scanner = new ClusterScanner(jc) {
            @Override
            public void action(String key) {
                ai.incrementAndGet();
            }
        };

        String scanFor = "TEST*";
        long start = System.currentTimeMillis();
        scanner.scan(scanFor, 100);

        log.info("Scan for '" + scanFor + "' found " + ai + " keys in " + ((System.currentTimeMillis() - start) / 1000.0) + " seconds");

        System.exit(0);
    }
}

========================

Dasun Nirmitha

unread,

Feb 1, 2016, 1:41:44 AM2/1/16

to Jedis

Hello,

I managed to figure out the issue.

In each VM I've got an Internal Network and a Host-only Adapter running and the jedisCluster.getClusterNodes() call returns 2 entries from each VM because of that.

That's what had caused the exception and filtering out the Internal Network entries and using only the Host-only Adapter network entries within the for loop solved the issue.

Thanks for everyone that helped :)

Reply all

Reply to author

Forward