From 02742e3b4921cdda886a9455839bc5f691b2d0af Mon Sep 17 00:00:00 2001 From: Christoph Strobl Date: Wed, 24 Feb 2016 14:55:11 +0100 Subject: [PATCH] DATAREDIS-467 - Fix cluster multi key commands. We moved away from returning raw map types and now return dedicated objects for command executions in cluster environment. This allows to maintain node information and collecting results from each and every callback. MGET now returns values according to the key position. Additionally we now prefix info commands in the cluster with the nodes host:port. Original pull request: #174. --- .../connection/ClusterCommandExecutor.java | 341 +++++++++++++++--- .../jedis/JedisClusterConnection.java | 293 +++++++-------- .../lettuce/LettuceClusterConnection.java | 238 ++++++------ .../data/redis/core/RedisTemplate.java | 2 +- .../ClusterCommandExecutorUnitTests.java | 42 +-- .../jedis/JedisClusterConnectionTests.java | 29 +- .../LettuceClusterConnectionTests.java | 7 +- 7 files changed, 606 insertions(+), 346 deletions(-) diff --git a/src/main/java/org/springframework/data/redis/connection/ClusterCommandExecutor.java b/src/main/java/org/springframework/data/redis/connection/ClusterCommandExecutor.java index 961da321f..610d103d5 100644 --- a/src/main/java/org/springframework/data/redis/connection/ClusterCommandExecutor.java +++ b/src/main/java/org/springframework/data/redis/connection/ClusterCommandExecutor.java @@ -16,7 +16,12 @@ package org.springframework.data.redis.connection; import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.Comparator; import java.util.HashMap; +import java.util.HashSet; import java.util.LinkedHashMap; import java.util.LinkedHashSet; import java.util.List; @@ -35,13 +40,17 @@ import org.springframework.data.redis.ClusterRedirectException; import org.springframework.data.redis.ClusterStateFailureException; import org.springframework.data.redis.ExceptionTranslationStrategy; import org.springframework.data.redis.TooManyClusterRedirectionsException; +import org.springframework.data.redis.connection.util.ByteArraySet; +import org.springframework.data.redis.connection.util.ByteArrayWrapper; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.util.Assert; +import org.springframework.util.CollectionUtils; +import org.springframework.util.ObjectUtils; /** * {@link ClusterCommandExecutor} takes care of running commands across the known cluster nodes. By providing an * {@link AsyncTaskExecutor} the execution behavior can be influenced. - * + * * @author Christoph Strobl * @author Mark Paluch * @since 1.7 @@ -56,7 +65,7 @@ public class ClusterCommandExecutor implements DisposableBean { /** * Create a new instance of {@link ClusterCommandExecutor}. - * + * * @param topologyProvider must not be {@literal null}. * @param resourceProvider must not be {@literal null}. * @param exceptionTranslation must not be {@literal null}. @@ -96,11 +105,11 @@ public class ClusterCommandExecutor implements DisposableBean { /** * Run {@link ClusterCommandCallback} on a random node. - * + * * @param cmd must not be {@literal null}. * @return */ - public T executeCommandOnArbitraryNode(ClusterCommandCallback cmd) { + public NodeResult executeCommandOnArbitraryNode(ClusterCommandCallback cmd) { Assert.notNull(cmd, "ClusterCommandCallback must not be null!"); List nodes = new ArrayList(getClusterTopology().getActiveNodes()); @@ -109,27 +118,26 @@ public class ClusterCommandExecutor implements DisposableBean { /** * Run {@link ClusterCommandCallback} on given {@link RedisClusterNode}. - * + * * @param cmd must not be {@literal null}. * @param node must not be {@literal null}. * @throws IllegalArgumentException in case no resource can be acquired for given node. * @return */ - public T executeCommandOnSingleNode(ClusterCommandCallback cmd, RedisClusterNode node) { + public NodeResult executeCommandOnSingleNode(ClusterCommandCallback cmd, RedisClusterNode node) { return executeCommandOnSingleNode(cmd, node, 0); } - private T executeCommandOnSingleNode(ClusterCommandCallback cmd, RedisClusterNode node, int redirectCount) { + private NodeResult executeCommandOnSingleNode(ClusterCommandCallback cmd, RedisClusterNode node, + int redirectCount) { Assert.notNull(cmd, "ClusterCommandCallback must not be null!"); Assert.notNull(node, "RedisClusterNode must not be null!"); if (redirectCount > maxRedirects) { - throw new TooManyClusterRedirectionsException( - String - .format( - "Cannot follow Cluster Redirects over more than %s legs. Please consider increasing the number of redirects to follow. Current value is: %s.", - redirectCount, maxRedirects)); + throw new TooManyClusterRedirectionsException(String.format( + "Cannot follow Cluster Redirects over more than %s legs. Please consider increasing the number of redirects to follow. Current value is: %s.", + redirectCount, maxRedirects)); } RedisClusterNode nodeToUse = lookupNode(node); @@ -138,7 +146,7 @@ public class ClusterCommandExecutor implements DisposableBean { Assert.notNull(client, "Could not acquire resource for node. Is your cluster info up to date?"); try { - return cmd.doInCluster(client); + return new NodeResult(node, cmd.doInCluster(client)); } catch (RuntimeException ex) { RuntimeException translatedException = convertToDataAccessExeption(ex); @@ -156,7 +164,7 @@ public class ClusterCommandExecutor implements DisposableBean { /** * Lookup node from the topology. - * + * * @param node * @return * @throws IllegalArgumentException in case the node could not be resolved to a topology-known node @@ -171,12 +179,12 @@ public class ClusterCommandExecutor implements DisposableBean { /** * Run {@link ClusterCommandCallback} on all reachable master nodes. - * + * * @param cmd * @return * @throws ClusterCommandExecutionFailureException */ - public Map executeCommandOnAllNodes(final ClusterCommandCallback cmd) { + public MulitNodeResult executeCommandOnAllNodes(final ClusterCommandCallback cmd) { return executeCommandAsyncOnNodes(cmd, getClusterTopology().getActiveMasterNodes()); } @@ -187,8 +195,8 @@ public class ClusterCommandExecutor implements DisposableBean { * @throws ClusterCommandExecutionFailureException * @throws IllegalArgumentException in case the node could not be resolved to a topology-known node */ - public java.util.Map executeCommandAsyncOnNodes( - final ClusterCommandCallback callback, Iterable nodes) { + public MulitNodeResult executeCommandAsyncOnNodes(final ClusterCommandCallback callback, + Iterable nodes) { Assert.notNull(callback, "Callback must not be null!"); Assert.notNull(nodes, "Nodes must not be null!"); @@ -204,13 +212,13 @@ public class ClusterCommandExecutor implements DisposableBean { } } - Map> futures = new LinkedHashMap>(); + Map>> futures = new LinkedHashMap>>(); for (final RedisClusterNode node : resolvedRedisClusterNodes) { - futures.put(node, executor.submit(new Callable() { + futures.put(new NodeExecution(node), executor.submit(new Callable>() { @Override - public T call() throws Exception { + public NodeResult call() throws Exception { return executeCommandOnSingleNode(callback, node); } })); @@ -219,35 +227,40 @@ public class ClusterCommandExecutor implements DisposableBean { return collectResults(futures); } - private Map collectResults(Map> futures) { + private MulitNodeResult collectResults(Map>> futures) { boolean done = false; - Map result = new HashMap(); + MulitNodeResult result = new MulitNodeResult(); Map exceptions = new HashMap(); + + Set saveGuard = new HashSet(); while (!done) { done = true; - for (Map.Entry> entry : futures.entrySet()) { + for (Map.Entry>> entry : futures.entrySet()) { if (!entry.getValue().isDone() && !entry.getValue().isCancelled()) { done = false; } else { - if (!result.containsKey(entry.getKey()) && !exceptions.containsKey(entry.getKey())) { - try { - result.put(entry.getKey(), entry.getValue().get()); - } catch (ExecutionException e) { + try { - RuntimeException ex = convertToDataAccessExeption((Exception) e.getCause()); - exceptions.put(entry.getKey(), ex != null ? ex : e.getCause()); - } catch (InterruptedException e) { - - Thread.currentThread().interrupt(); - - RuntimeException ex = convertToDataAccessExeption((Exception) e.getCause()); - exceptions.put(entry.getKey(), ex != null ? ex : e.getCause()); - break; + String futureId = ObjectUtils.getIdentityHexString(entry.getValue()); + if (!saveGuard.contains(futureId)) { + result.add(entry.getValue().get()); + saveGuard.add(futureId); } + } catch (ExecutionException e) { + + RuntimeException ex = convertToDataAccessExeption((Exception) e.getCause()); + exceptions.put(entry.getKey().getNode(), ex != null ? ex : e.getCause()); + } catch (InterruptedException e) { + + Thread.currentThread().interrupt(); + + RuntimeException ex = convertToDataAccessExeption((Exception) e.getCause()); + exceptions.put(entry.getKey().getNode(), ex != null ? ex : e.getCause()); + break; } } } @@ -268,12 +281,12 @@ public class ClusterCommandExecutor implements DisposableBean { /** * Run {@link MultiKeyClusterCommandCallback} with on a curated set of nodes serving one or more keys. - * + * * @param cmd * @return * @throws ClusterCommandExecutionFailureException */ - public Map executeMuliKeyCommand(final MultiKeyClusterCommandCallback cmd, + public MulitNodeResult executeMuliKeyCommand(final MultiKeyClusterCommandCallback cmd, Iterable keys) { Map> nodeKeyMap = new HashMap>(); @@ -291,26 +304,28 @@ public class ClusterCommandExecutor implements DisposableBean { } } - Map> futures = new LinkedHashMap>(); + Map>> futures = new LinkedHashMap>>(); + for (final Entry> entry : nodeKeyMap.entrySet()) { if (entry.getKey().isMaster()) { for (final byte[] key : entry.getValue()) { - futures.put(entry.getKey(), executor.submit(new Callable() { + futures.put(new NodeExecution(entry.getKey(), key), executor.submit(new Callable>() { @Override - public T call() throws Exception { - return (T) executeMultiKeyCommandOnSingleNode(cmd, entry.getKey(), key); + public NodeResult call() throws Exception { + return executeMultiKeyCommandOnSingleNode(cmd, entry.getKey(), key); } })); } } } + return collectResults(futures); } - private T executeMultiKeyCommandOnSingleNode(MultiKeyClusterCommandCallback cmd, RedisClusterNode node, - byte[] key) { + private NodeResult executeMultiKeyCommandOnSingleNode(MultiKeyClusterCommandCallback cmd, + RedisClusterNode node, byte[] key) { Assert.notNull(cmd, "MultiKeyCommandCallback must not be null!"); Assert.notNull(node, "RedisClusterNode must not be null!"); @@ -320,7 +335,7 @@ public class ClusterCommandExecutor implements DisposableBean { Assert.notNull(client, "Could not acquire resource for node. Is your cluster info up to date?"); try { - return cmd.doInCluster(client, key); + return new NodeResult(node, cmd.doInCluster(client, key), key); } catch (RuntimeException ex) { RuntimeException translatedException = convertToDataAccessExeption(ex); @@ -340,7 +355,7 @@ public class ClusterCommandExecutor implements DisposableBean { /** * Set the maximum number of redirects to follow on {@code MOVED} or {@code ASK}. - * + * * @param maxRedirects set to zero to suspend redirects. */ public void setMaxRedirects(int maxRedirects) { @@ -362,7 +377,7 @@ public class ClusterCommandExecutor implements DisposableBean { /** * Callback interface for Redis 'low level' code using the cluster client directly. To be used with * {@link ClusterCommandExecutor} execution methods. - * + * * @author Christoph Strobl * @param native driver connection * @param @@ -374,7 +389,7 @@ public class ClusterCommandExecutor implements DisposableBean { /** * Callback interface for Redis 'low level' code using the cluster client to execute multi key commands. - * + * * @author Christoph Strobl * @param native driver connection * @param @@ -383,4 +398,232 @@ public class ClusterCommandExecutor implements DisposableBean { S doInCluster(T client, byte[] key); } + /** + * {@link NodeExecution} encapsulates the execution of a command on a specific node along with arguments, such as + * keys, involved. + * + * @author Christoph Strobl + * @since 1.7 + */ + private static class NodeExecution { + + private RedisClusterNode node; + private Object[] args; + + public NodeExecution(RedisClusterNode node, Object... args) { + + this.node = node; + this.args = args; + } + + /** + * Get the {@link RedisClusterNode} the execution happens on. + */ + public RedisClusterNode getNode() { + return node; + } + + /* + * (non-Javadoc) + * @see java.lang.Object#hashCode() + */ + @Override + public int hashCode() { + + int result = ObjectUtils.nullSafeHashCode(node); + return result + ObjectUtils.nullSafeHashCode(args); + } + + /* + * (non-Javadoc) + * @see java.lang.Object#equals(java.lang.Object) + */ + @Override + public boolean equals(Object obj) { + + if (this == obj) { + return true; + } + + if (obj == null) { + return false; + } + + if (!(obj instanceof NodeExecution)) { + return false; + } + + NodeExecution that = (NodeExecution) obj; + + if (!ObjectUtils.nullSafeEquals(this.node, that.node)) { + return false; + } + + return ObjectUtils.nullSafeEquals(this.args, that.args); + } + + } + + /** + * {@link NodeResult} encapsules the actual value returned by a {@link ClusterCommandCallback} on a given + * {@link RedisClusterNode}. + * + * @author Christoph Strobl + * @param + * @since 1.7 + */ + public static class NodeResult { + + private RedisClusterNode node; + private T value; + private ByteArrayWrapper key; + + /** + * Create new {@link NodeResult}. + * + * @param node + * @param value + */ + public NodeResult(RedisClusterNode node, T value) { + this(node, value, new byte[] {}); + } + + /** + * Create new {@link NodeResult}. + * + * @param node + * @param value + * @parm key + */ + public NodeResult(RedisClusterNode node, T value, byte[] key) { + + this.node = node; + this.value = value; + + this.key = new ByteArrayWrapper(key); + } + + /** + * Get the actual value of the command execution. + * + * @return can be {@literal null}. + */ + public T getValue() { + return value; + } + + /** + * Get the {@link RedisClusterNode} the command was executed on. + * + * @return never {@literal null}. + */ + public RedisClusterNode getNode() { + return node; + } + + /** + * @return + */ + public byte[] getKey() { + return key.getArray(); + } + } + + /** + * {@link MulitNodeResult} holds all {@link NodeResult} of a command executed on multiple {@link RedisClusterNode}. + * + * @author Christoph Strobl + * @param + * @since 1.7 + */ + public static class MulitNodeResult { + + List> nodeResults = new ArrayList>(); + + private void add(NodeResult result) { + nodeResults.add(result); + } + + /** + * @return never {@literal null}. + */ + public List> getResults() { + return Collections.unmodifiableList(nodeResults); + } + + /** + * Get {@link List} of all individual {@link NodeResult#value}.
+ * The resulting {@link List} may contain {@literal null} values. + * + * @return never {@literal null}. + */ + public List resultsAsList() { + return toList(nodeResults); + } + + /** + * Get {@link List} of all individual {@link NodeResult#value}.
+ * The resulting {@link List} may contain {@literal null} values. + * + * @return never {@literal null}. + */ + public List resultsAsListSortBy(byte[]... keys) { + + ArrayList> clone = new ArrayList>(nodeResults); + Collections.sort(clone, new ResultByReferenceKeyPositionComperator(keys)); + + return toList(clone); + } + + /** + * @param returnValue + * @return + */ + public T getFirstNonNullNotEmptyOrDefault(T returnValue) { + + for (NodeResult nodeResult : nodeResults) { + if (nodeResult.getValue() != null) { + if (nodeResult.getValue() instanceof Map) { + if (CollectionUtils.isEmpty((Map) nodeResult.getValue())) { + return nodeResult.getValue(); + } + } else if (CollectionUtils.isEmpty((Collection) nodeResult.getValue())) { + return nodeResult.getValue(); + } else { + return nodeResult.getValue(); + } + } + } + + return returnValue; + } + + private List toList(Collection> source) { + + ArrayList result = new ArrayList(); + for (NodeResult nodeResult : source) { + result.add(nodeResult.getValue()); + } + return result; + } + + /** + * {@link Comparator} for sorting {@link NodeResult} by reference keys. + * + * @author Christoph Strobl + */ + private static class ResultByReferenceKeyPositionComperator implements Comparator> { + + List reference; + + public ResultByReferenceKeyPositionComperator(byte[]... keys) { + reference = new ArrayList(new ByteArraySet(Arrays.asList(keys))); + } + + @Override + public int compare(NodeResult o1, NodeResult o2) { + return Integer.compare(reference.indexOf(o1.key), reference.indexOf(o2.key)); + } + } + } } diff --git a/src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterConnection.java b/src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterConnection.java index 841ee7e7a..cea347005 100644 --- a/src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterConnection.java @@ -39,6 +39,7 @@ import org.springframework.data.redis.PassThroughExceptionTranslationStrategy; import org.springframework.data.redis.connection.ClusterCommandExecutor; import org.springframework.data.redis.connection.ClusterCommandExecutor.ClusterCommandCallback; import org.springframework.data.redis.connection.ClusterCommandExecutor.MultiKeyClusterCommandCallback; +import org.springframework.data.redis.connection.ClusterCommandExecutor.NodeResult; import org.springframework.data.redis.connection.ClusterInfo; import org.springframework.data.redis.connection.ClusterNodeResourceProvider; import org.springframework.data.redis.connection.ClusterSlotHashUtil; @@ -110,8 +111,8 @@ public class JedisClusterConnection implements RedisClusterConnection { closed = false; topologyProvider = new JedisClusterTopologyProvider(cluster); - clusterCommandExecutor = new ClusterCommandExecutor(topologyProvider, - new JedisClusterNodeResourceProvider(cluster), EXCEPTION_TRANSLATION); + clusterCommandExecutor = new ClusterCommandExecutor(topologyProvider, new JedisClusterNodeResourceProvider(cluster), + EXCEPTION_TRANSLATION); try { DirectFieldAccessor dfa = new DirectFieldAccessor(cluster); @@ -168,14 +169,14 @@ public class JedisClusterConnection implements RedisClusterConnection { } } - return Long.valueOf(this.clusterCommandExecutor.executeMuliKeyCommand( - new JedisMultiKeyClusterCommandCallback() { + return Long + .valueOf(this.clusterCommandExecutor.executeMuliKeyCommand(new JedisMultiKeyClusterCommandCallback() { @Override public Long doInCluster(Jedis client, byte[] key) { return client.del(key); } - }, Arrays.asList(keys)).size()); + }, Arrays.asList(keys)).resultsAsList().size()); } /* @@ -201,14 +202,14 @@ public class JedisClusterConnection implements RedisClusterConnection { Assert.notNull(pattern, "Pattern must not be null!"); - Collection> keysPerNode = clusterCommandExecutor.executeCommandOnAllNodes( - new JedisClusterCommandCallback>() { + Collection> keysPerNode = clusterCommandExecutor + .executeCommandOnAllNodes(new JedisClusterCommandCallback>() { @Override public Set doInCluster(Jedis client) { return client.keys(pattern); } - }).values(); + }).resultsAsList(); Set keys = new HashSet(); for (Set keySet : keysPerNode) { @@ -232,7 +233,7 @@ public class JedisClusterConnection implements RedisClusterConnection { public Set doInCluster(Jedis client) { return client.keys(pattern); } - }, node); + }, node).getValue(); } /* @@ -251,8 +252,8 @@ public class JedisClusterConnection implements RedisClusterConnection { @Override public byte[] randomKey() { - List nodes = new ArrayList(topologyProvider.getTopology() - .getActiveMasterNodes()); + List nodes = new ArrayList( + topologyProvider.getTopology().getActiveMasterNodes()); Set inspectedNodes = new HashSet(nodes.size()); do { @@ -286,7 +287,7 @@ public class JedisClusterConnection implements RedisClusterConnection { public byte[] doInCluster(Jedis client) { return client.randomBinaryKey(); } - }, node); + }, node).getValue(); } /* @@ -451,7 +452,7 @@ public class JedisClusterConnection implements RedisClusterConnection { public Long doInCluster(Jedis client) { return client.pttl(key); } - }, topologyProvider.getTopology().getKeyServingMasterNode(key)); + }, topologyProvider.getTopology().getKeyServingMasterNode(key)).getValue(); } /* @@ -507,7 +508,7 @@ public class JedisClusterConnection implements RedisClusterConnection { public byte[] doInCluster(Jedis client) { return client.dump(key); } - }, topologyProvider.getTopology().getKeyServingMasterNode(key)); + }, topologyProvider.getTopology().getKeyServingMasterNode(key)).getValue(); } /* @@ -571,16 +572,13 @@ public class JedisClusterConnection implements RedisClusterConnection { return cluster.mget(keys); } - Map nodeResult = this.clusterCommandExecutor.executeMuliKeyCommand( - new JedisMultiKeyClusterCommandCallback() { + return this.clusterCommandExecutor.executeMuliKeyCommand(new JedisMultiKeyClusterCommandCallback() { - @Override - public byte[] doInCluster(Jedis client, byte[] key) { - return client.get(key); - } - }, Arrays.asList(keys)); - - return new ArrayList(nodeResult.values()); + @Override + public byte[] doInCluster(Jedis client, byte[] key) { + return client.get(key); + } + }, Arrays.asList(keys)).resultsAsListSortBy(keys); } /* @@ -1120,22 +1118,13 @@ public class JedisClusterConnection implements RedisClusterConnection { } } - Map> nodeResult = this.clusterCommandExecutor.executeMuliKeyCommand( - new JedisMultiKeyClusterCommandCallback>() { + return this.clusterCommandExecutor.executeMuliKeyCommand(new JedisMultiKeyClusterCommandCallback>() { - @Override - public List doInCluster(Jedis client, byte[] key) { - return client.blpop(timeout, key); - } - }, Arrays.asList(keys)); - - for (List partial : nodeResult.values()) { - if (!partial.isEmpty()) { - return partial; + @Override + public List doInCluster(Jedis client, byte[] key) { + return client.blpop(timeout, key); } - } - - return Collections.emptyList(); + }, Arrays.asList(keys)).getFirstNonNullNotEmptyOrDefault(Collections. emptyList()); } /* @@ -1145,22 +1134,13 @@ public class JedisClusterConnection implements RedisClusterConnection { @Override public List bRPop(final int timeout, byte[]... keys) { - Map> nodeResult = this.clusterCommandExecutor.executeMuliKeyCommand( - new JedisMultiKeyClusterCommandCallback>() { + return this.clusterCommandExecutor.executeMuliKeyCommand(new JedisMultiKeyClusterCommandCallback>() { - @Override - public List doInCluster(Jedis client, byte[] key) { - return client.brpop(timeout, key); - } - }, Arrays.asList(keys)); - - for (List partial : nodeResult.values()) { - if (!partial.isEmpty()) { - return partial; + @Override + public List doInCluster(Jedis client, byte[] key) { + return client.brpop(timeout, key); } - } - - return Collections.emptyList(); + }, Arrays.asList(keys)).getFirstNonNullNotEmptyOrDefault(Collections. emptyList()); } /* @@ -1314,19 +1294,20 @@ public class JedisClusterConnection implements RedisClusterConnection { } } - Map> nodeResult = this.clusterCommandExecutor.executeMuliKeyCommand( - new JedisMultiKeyClusterCommandCallback>() { + Collection> resultList = this.clusterCommandExecutor + .executeMuliKeyCommand(new JedisMultiKeyClusterCommandCallback>() { @Override public Set doInCluster(Jedis client, byte[] key) { return client.smembers(key); } - }, Arrays.asList(keys)); + }, Arrays.asList(keys)).resultsAsList(); ByteArraySet result = null; - for (Entry> entry : nodeResult.entrySet()) { - ByteArraySet tmp = new ByteArraySet(entry.getValue()); + for (Set value : resultList) { + + ByteArraySet tmp = new ByteArraySet(value); if (result == null) { result = tmp; } else { @@ -1383,18 +1364,18 @@ public class JedisClusterConnection implements RedisClusterConnection { } } - Map> nodeResult = this.clusterCommandExecutor.executeMuliKeyCommand( - new JedisMultiKeyClusterCommandCallback>() { + Collection> resultList = this.clusterCommandExecutor + .executeMuliKeyCommand(new JedisMultiKeyClusterCommandCallback>() { @Override public Set doInCluster(Jedis client, byte[] key) { return client.smembers(key); } - }, Arrays.asList(keys)); + }, Arrays.asList(keys)).resultsAsList(); ByteArraySet result = new ByteArraySet(); - for (Entry> entry : nodeResult.entrySet()) { - result.addAll(entry.getValue()); + for (Set entry : resultList) { + result.addAll(entry); } if (result.isEmpty()) { @@ -1447,21 +1428,21 @@ public class JedisClusterConnection implements RedisClusterConnection { byte[][] others = Arrays.copyOfRange(keys, 1, keys.length - 1); ByteArraySet values = new ByteArraySet(sMembers(source)); - Map> nodeResult = clusterCommandExecutor.executeMuliKeyCommand( - new JedisMultiKeyClusterCommandCallback>() { + Collection> resultList = clusterCommandExecutor + .executeMuliKeyCommand(new JedisMultiKeyClusterCommandCallback>() { @Override public Set doInCluster(Jedis client, byte[] key) { return client.smembers(key); } - }, Arrays.asList(others)); + }, Arrays.asList(others)).resultsAsList(); if (values.isEmpty()) { return Collections.emptySet(); } - for (Set toSubstract : nodeResult.values()) { - values.removeAll(toSubstract); + for (Set singleNodeValue : resultList) { + values.removeAll(singleNodeValue); } return values.asRawSet(); @@ -1552,8 +1533,8 @@ public class JedisClusterConnection implements RedisClusterConnection { redis.clients.jedis.ScanResult result = cluster.sscan(JedisConverters.toString(key), Long.toString(cursorId)); - return new ScanIteration(Long.valueOf(result.getCursor()), JedisConverters.stringListToByteList() - .convert(result.getResult())); + return new ScanIteration(Long.valueOf(result.getCursor()), + JedisConverters.stringListToByteList().convert(result.getResult())); } }.open(); } @@ -1672,8 +1653,8 @@ public class JedisClusterConnection implements RedisClusterConnection { try { if (limit != null) { - return JedisConverters.toTupleSet(cluster.zrangeByScoreWithScores(key, min, max, limit.getOffset(), - limit.getCount())); + return JedisConverters + .toTupleSet(cluster.zrangeByScoreWithScores(key, min, max, limit.getOffset(), limit.getCount())); } return JedisConverters.toTupleSet(cluster.zrangeByScoreWithScores(key, min, max)); } catch (Exception ex) { @@ -1734,8 +1715,8 @@ public class JedisClusterConnection implements RedisClusterConnection { try { if (limit != null) { - return JedisConverters.toTupleSet(cluster.zrevrangeByScoreWithScores(key, max, min, limit.getOffset(), - limit.getCount())); + return JedisConverters + .toTupleSet(cluster.zrevrangeByScoreWithScores(key, max, min, limit.getOffset(), limit.getCount())); } return JedisConverters.toTupleSet(cluster.zrevrangeByScoreWithScores(key, max, min)); } catch (Exception ex) { @@ -2018,8 +1999,8 @@ public class JedisClusterConnection implements RedisClusterConnection { } try { - return JedisConverters.toTupleSet(cluster.zrevrangeByScoreWithScores(key, max, min, Long.valueOf(offset) - .intValue(), Long.valueOf(count).intValue())); + return JedisConverters.toTupleSet(cluster.zrevrangeByScoreWithScores(key, max, min, + Long.valueOf(offset).intValue(), Long.valueOf(count).intValue())); } catch (Exception ex) { throw convertJedisAccessException(ex); } @@ -2206,8 +2187,8 @@ public class JedisClusterConnection implements RedisClusterConnection { } try { - return cluster.zrangeByScore(key, JedisConverters.toBytes(min), JedisConverters.toBytes(max), Long - .valueOf(offset).intValue(), Long.valueOf(count).intValue()); + return cluster.zrangeByScore(key, JedisConverters.toBytes(min), JedisConverters.toBytes(max), + Long.valueOf(offset).intValue(), Long.valueOf(count).intValue()); } catch (Exception ex) { throw convertJedisAccessException(ex); } @@ -2553,7 +2534,7 @@ public class JedisClusterConnection implements RedisClusterConnection { public String doInCluster(Jedis client) { return client.ping(); } - }).isEmpty() ? "PONG" : null; + }).resultsAsList().isEmpty() ? "PONG" : null; } @@ -2570,7 +2551,7 @@ public class JedisClusterConnection implements RedisClusterConnection { public String doInCluster(Jedis client) { return client.ping(); } - }, node); + }, node).getValue(); } /* @@ -2660,14 +2641,14 @@ public class JedisClusterConnection implements RedisClusterConnection { @Override public Long lastSave() { - List result = new ArrayList(clusterCommandExecutor.executeCommandOnAllNodes( - new JedisClusterCommandCallback() { + List result = new ArrayList( + clusterCommandExecutor.executeCommandOnAllNodes(new JedisClusterCommandCallback() { @Override public Long doInCluster(Jedis client) { return client.lastsave(); } - }).values()); + }).resultsAsList()); if (CollectionUtils.isEmpty(result)) { return null; @@ -2690,7 +2671,7 @@ public class JedisClusterConnection implements RedisClusterConnection { public Long doInCluster(Jedis client) { return client.lastsave(); } - }, node); + }, node).getValue(); } /* @@ -2732,21 +2713,20 @@ public class JedisClusterConnection implements RedisClusterConnection { @Override public Long dbSize() { - Map dbSizes = clusterCommandExecutor - .executeCommandOnAllNodes(new JedisClusterCommandCallback() { + Collection dbSizes = clusterCommandExecutor.executeCommandOnAllNodes(new JedisClusterCommandCallback() { - @Override - public Long doInCluster(Jedis client) { - return client.dbSize(); - } - }); + @Override + public Long doInCluster(Jedis client) { + return client.dbSize(); + } + }).resultsAsList(); if (CollectionUtils.isEmpty(dbSizes)) { return 0L; } Long size = 0L; - for (Long value : dbSizes.values()) { + for (Long value : dbSizes) { size += value; } return size; @@ -2765,7 +2745,7 @@ public class JedisClusterConnection implements RedisClusterConnection { public Long doInCluster(Jedis client) { return client.dbSize(); } - }, node); + }, node).getValue(); } /* @@ -2841,13 +2821,20 @@ public class JedisClusterConnection implements RedisClusterConnection { Properties infos = new Properties(); - infos.putAll(clusterCommandExecutor.executeCommandOnAllNodes(new JedisClusterCommandCallback() { + List> nodeResults = clusterCommandExecutor + .executeCommandOnAllNodes(new JedisClusterCommandCallback() { - @Override - public Properties doInCluster(Jedis client) { - return JedisConverters.toProperties(client.info()); + @Override + public Properties doInCluster(Jedis client) { + return JedisConverters.toProperties(client.info()); + } + }).getResults(); + + for (NodeResult nodePorperties : nodeResults) { + for (Entry entry : nodePorperties.getValue().entrySet()) { + infos.put(nodePorperties.getNode().asString() + "." + entry.getKey(), entry.getValue()); } - })); + } return infos; } @@ -2859,14 +2846,14 @@ public class JedisClusterConnection implements RedisClusterConnection { @Override public Properties info(RedisClusterNode node) { - return JedisConverters.toProperties(clusterCommandExecutor.executeCommandOnSingleNode( - new JedisClusterCommandCallback() { + return JedisConverters + .toProperties(clusterCommandExecutor.executeCommandOnSingleNode(new JedisClusterCommandCallback() { @Override public String doInCluster(Jedis client) { return client.info(); } - }, node)); + }, node).getValue()); } /* @@ -2878,13 +2865,20 @@ public class JedisClusterConnection implements RedisClusterConnection { Properties infos = new Properties(); - infos.putAll(clusterCommandExecutor.executeCommandOnAllNodes(new JedisClusterCommandCallback() { + List> nodeResults = clusterCommandExecutor + .executeCommandOnAllNodes(new JedisClusterCommandCallback() { - @Override - public Properties doInCluster(Jedis client) { - return JedisConverters.toProperties(client.info(section)); + @Override + public Properties doInCluster(Jedis client) { + return JedisConverters.toProperties(client.info(section)); + } + }).getResults(); + + for (NodeResult nodePorperties : nodeResults) { + for (Entry entry : nodePorperties.getValue().entrySet()) { + infos.put(nodePorperties.getNode().asString() + "." + entry.getKey(), entry.getValue()); } - })); + } return infos; } @@ -2896,14 +2890,14 @@ public class JedisClusterConnection implements RedisClusterConnection { @Override public Properties info(RedisClusterNode node, final String section) { - return JedisConverters.toProperties(clusterCommandExecutor.executeCommandOnSingleNode( - new JedisClusterCommandCallback() { + return JedisConverters + .toProperties(clusterCommandExecutor.executeCommandOnSingleNode(new JedisClusterCommandCallback() { @Override public String doInCluster(Jedis client) { return client.info(section); } - }, node)); + }, node).getValue()); } /* @@ -2961,19 +2955,19 @@ public class JedisClusterConnection implements RedisClusterConnection { @Override public List getConfig(final String pattern) { - Map> mapResult = clusterCommandExecutor + List>> mapResult = clusterCommandExecutor .executeCommandOnAllNodes(new JedisClusterCommandCallback>() { @Override public List doInCluster(Jedis client) { return client.configGet(pattern); } - }); + }).getResults(); List result = new ArrayList(); - for (Entry> entry : mapResult.entrySet()) { + for (NodeResult> entry : mapResult) { - String prefix = entry.getKey().asString(); + String prefix = entry.getNode().asString(); int i = 0; for (String value : entry.getValue()) { result.add((i++ % 2 == 0 ? (prefix + ".") : "") + value); @@ -2996,7 +2990,7 @@ public class JedisClusterConnection implements RedisClusterConnection { public List doInCluster(Jedis client) { return client.configGet(pattern); } - }, node); + }, node).getValue(); } /* @@ -3070,14 +3064,14 @@ public class JedisClusterConnection implements RedisClusterConnection { @Override public Long time() { - return convertListOfStringToTime(clusterCommandExecutor - .executeCommandOnArbitraryNode(new JedisClusterCommandCallback>() { + return convertListOfStringToTime( + clusterCommandExecutor.executeCommandOnArbitraryNode(new JedisClusterCommandCallback>() { @Override public List doInCluster(Jedis client) { return client.time(); } - })); + }).getValue()); } /* @@ -3087,14 +3081,14 @@ public class JedisClusterConnection implements RedisClusterConnection { @Override public Long time(RedisClusterNode node) { - return convertListOfStringToTime(clusterCommandExecutor.executeCommandOnSingleNode( - new JedisClusterCommandCallback>() { + return convertListOfStringToTime( + clusterCommandExecutor.executeCommandOnSingleNode(new JedisClusterCommandCallback>() { @Override public List doInCluster(Jedis client) { return client.time(); } - }, node)); + }, node).getValue()); } private Long convertListOfStringToTime(List serverTimeInformation) { @@ -3149,17 +3143,16 @@ public class JedisClusterConnection implements RedisClusterConnection { @Override public List getClientList() { - Map map = clusterCommandExecutor - .executeCommandOnAllNodes(new JedisClusterCommandCallback() { + Collection map = clusterCommandExecutor.executeCommandOnAllNodes(new JedisClusterCommandCallback() { - @Override - public String doInCluster(Jedis client) { - return client.clientList(); - } - }); + @Override + public String doInCluster(Jedis client) { + return client.clientList(); + } + }).resultsAsList(); ArrayList result = new ArrayList(); - for (String infos : map.values()) { + for (String infos : map) { result.addAll(JedisConverters.toListOfRedisClientInformation(infos)); } return result; @@ -3172,14 +3165,14 @@ public class JedisClusterConnection implements RedisClusterConnection { @Override public List getClientList(RedisClusterNode node) { - return JedisConverters.toListOfRedisClientInformation(clusterCommandExecutor.executeCommandOnSingleNode( - new JedisClusterCommandCallback() { + return JedisConverters.toListOfRedisClientInformation( + clusterCommandExecutor.executeCommandOnSingleNode(new JedisClusterCommandCallback() { @Override public String doInCluster(Jedis client) { return client.clientList(); } - }, node)); + }, node).getValue()); } /* @@ -3385,8 +3378,8 @@ public class JedisClusterConnection implements RedisClusterConnection { @Override public List doInCluster(Jedis client) { - return JedisConverters.stringListToByteList().convert( - client.clusterGetKeysInSlot(slot, count != null ? count.intValue() : Integer.MAX_VALUE)); + return JedisConverters.stringListToByteList() + .convert(client.clusterGetKeysInSlot(slot, count != null ? count.intValue() : Integer.MAX_VALUE)); } }, node); return null; @@ -3438,7 +3431,7 @@ public class JedisClusterConnection implements RedisClusterConnection { return client.clusterCountKeysInSlot(slot); } - }, node); + }, node).getValue(); } @@ -3478,8 +3471,8 @@ public class JedisClusterConnection implements RedisClusterConnection { @Override public void clusterForget(final RedisClusterNode node) { - Set nodes = new LinkedHashSet(topologyProvider.getTopology() - .getActiveMasterNodes()); + Set nodes = new LinkedHashSet( + topologyProvider.getTopology().getActiveMasterNodes()); final RedisClusterNode nodeToRemove = topologyProvider.getTopology().lookup(node); nodes.remove(nodeToRemove); @@ -3546,7 +3539,7 @@ public class JedisClusterConnection implements RedisClusterConnection { public Integer doInCluster(Jedis client) { return client.clusterKeySlot(JedisConverters.toString(key)).intValue(); } - }); + }).getValue(); } /* @@ -3585,14 +3578,14 @@ public class JedisClusterConnection implements RedisClusterConnection { final RedisClusterNode nodeToUse = topologyProvider.getTopology().lookup(master); - return JedisConverters.toSetOfRedisClusterNodes(clusterCommandExecutor.executeCommandOnSingleNode( - new JedisClusterCommandCallback>() { + return JedisConverters.toSetOfRedisClusterNodes( + clusterCommandExecutor.executeCommandOnSingleNode(new JedisClusterCommandCallback>() { @Override public List doInCluster(Jedis client) { return client.clusterSlaves(nodeToUse.getId()); } - }, master)); + }, master).getValue()); } /* @@ -3601,17 +3594,25 @@ public class JedisClusterConnection implements RedisClusterConnection { */ public Map> clusterGetMasterSlaveMap() { - return clusterCommandExecutor.executeCommandAsyncOnNodes( - new JedisClusterCommandCallback>() { + List>> nodeResults = clusterCommandExecutor + .executeCommandAsyncOnNodes(new JedisClusterCommandCallback>() { @Override public Set doInCluster(Jedis client) { // TODO: remove client.eval as soon as Jedis offers support for myid - return JedisConverters.toSetOfRedisClusterNodes(client.clusterSlaves((String) client.eval( - "return redis.call('cluster', 'myid')", 0))); + return JedisConverters.toSetOfRedisClusterNodes( + client.clusterSlaves((String) client.eval("return redis.call('cluster', 'myid')", 0))); } - }, topologyProvider.getTopology().getActiveMasterNodes()); + }, topologyProvider.getTopology().getActiveMasterNodes()).getResults(); + + Map> result = new LinkedHashMap>(); + + for (NodeResult> nodeResult : nodeResults) { + result.put(nodeResult.getNode(), nodeResult.getValue()); + } + + return result; } /* @@ -3630,14 +3631,14 @@ public class JedisClusterConnection implements RedisClusterConnection { @Override public ClusterInfo clusterGetClusterInfo() { - return new ClusterInfo(JedisConverters.toProperties(clusterCommandExecutor - .executeCommandOnArbitraryNode(new JedisClusterCommandCallback() { + return new ClusterInfo(JedisConverters + .toProperties(clusterCommandExecutor.executeCommandOnArbitraryNode(new JedisClusterCommandCallback() { @Override public String doInCluster(Jedis client) { return client.clusterInfo(); } - }))); + }).getValue())); } /* diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnection.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnection.java index 8dc9ec468..be02fdc41 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnection.java @@ -20,6 +20,7 @@ import java.util.Arrays; import java.util.Collection; import java.util.Collections; import java.util.HashSet; +import java.util.LinkedHashMap; import java.util.LinkedHashSet; import java.util.List; import java.util.Map; @@ -36,6 +37,7 @@ import org.springframework.data.redis.PassThroughExceptionTranslationStrategy; import org.springframework.data.redis.connection.ClusterCommandExecutor; import org.springframework.data.redis.connection.ClusterCommandExecutor.ClusterCommandCallback; import org.springframework.data.redis.connection.ClusterCommandExecutor.MultiKeyClusterCommandCallback; +import org.springframework.data.redis.connection.ClusterCommandExecutor.NodeResult; import org.springframework.data.redis.connection.ClusterInfo; import org.springframework.data.redis.connection.ClusterNodeResourceProvider; import org.springframework.data.redis.connection.ClusterSlotHashUtil; @@ -69,8 +71,8 @@ import com.lambdaworks.redis.codec.RedisCodec; * @author Mark Paluch * @since 1.7 */ -public class LettuceClusterConnection extends LettuceConnection implements - org.springframework.data.redis.connection.RedisClusterConnection { +public class LettuceClusterConnection extends LettuceConnection + implements org.springframework.data.redis.connection.RedisClusterConnection { static final ExceptionTranslationStrategy exceptionConverter = new PassThroughExceptionTranslationStrategy( new LettuceExceptionConverter()); @@ -93,8 +95,8 @@ public class LettuceClusterConnection extends LettuceConnection implements this.clusterClient = clusterClient; topologyProvider = new LettuceClusterTopologyProvider(clusterClient); - clusterCommandExecutor = new ClusterCommandExecutor(topologyProvider, new LettuceClusterNodeResourceProvider( - clusterClient), exceptionConverter); + clusterCommandExecutor = new ClusterCommandExecutor(topologyProvider, + new LettuceClusterNodeResourceProvider(clusterClient), exceptionConverter); } /** @@ -133,14 +135,14 @@ public class LettuceClusterConnection extends LettuceConnection implements Assert.notNull(pattern, "Pattern must not be null!"); - Collection> keysPerNode = clusterCommandExecutor.executeCommandOnAllNodes( - new LettuceClusterCommandCallback>() { + Collection> keysPerNode = clusterCommandExecutor + .executeCommandOnAllNodes(new LettuceClusterCommandCallback>() { @Override public List doInCluster(RedisClusterConnection connection) { return connection.keys(pattern); } - }).values(); + }).resultsAsList(); Set keys = new HashSet(); @@ -188,7 +190,7 @@ public class LettuceClusterConnection extends LettuceConnection implements @Override public Long dbSize() { - Map dbSizes = clusterCommandExecutor + Collection dbSizes = clusterCommandExecutor .executeCommandOnAllNodes(new LettuceClusterCommandCallback() { @Override @@ -196,14 +198,14 @@ public class LettuceClusterConnection extends LettuceConnection implements return client.dbsize(); } - }); + }).resultsAsList(); if (CollectionUtils.isEmpty(dbSizes)) { return 0L; } Long size = 0L; - for (Long value : dbSizes.values()) { + for (Long value : dbSizes) { size += value; } return size; @@ -218,13 +220,20 @@ public class LettuceClusterConnection extends LettuceConnection implements Properties infos = new Properties(); - infos.putAll(clusterCommandExecutor.executeCommandOnAllNodes(new LettuceClusterCommandCallback() { + List> nodeResults = clusterCommandExecutor + .executeCommandOnAllNodes(new LettuceClusterCommandCallback() { - @Override - public Properties doInCluster(RedisClusterConnection client) { - return LettuceConverters.toProperties(client.info()); + @Override + public Properties doInCluster(RedisClusterConnection client) { + return LettuceConverters.toProperties(client.info()); + } + }).getResults(); + + for (NodeResult nodePorperties : nodeResults) { + for (Entry entry : nodePorperties.getValue().entrySet()) { + infos.put(nodePorperties.getNode().asString() + "." + entry.getKey(), entry.getValue()); } - })); + } return infos; } @@ -233,14 +242,20 @@ public class LettuceClusterConnection extends LettuceConnection implements public Properties info(final String section) { Properties infos = new Properties(); + List> nodeResults = clusterCommandExecutor + .executeCommandOnAllNodes(new LettuceClusterCommandCallback() { - infos.putAll(clusterCommandExecutor.executeCommandOnAllNodes(new LettuceClusterCommandCallback() { + @Override + public Properties doInCluster(RedisClusterConnection client) { + return LettuceConverters.toProperties(client.info(section)); + } + }).getResults(); - @Override - public Properties doInCluster(RedisClusterConnection client) { - return LettuceConverters.toProperties(client.info(section)); + for (NodeResult nodePorperties : nodeResults) { + for (Entry entry : nodePorperties.getValue().entrySet()) { + infos.put(nodePorperties.getNode().asString() + "." + entry.getKey(), entry.getValue()); } - })); + } return infos; } @@ -252,14 +267,14 @@ public class LettuceClusterConnection extends LettuceConnection implements @Override public Properties info(RedisClusterNode node, final String section) { - return LettuceConverters.toProperties(clusterCommandExecutor.executeCommandOnSingleNode( - new LettuceClusterCommandCallback() { + return LettuceConverters + .toProperties(clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback() { @Override public String doInCluster(RedisClusterConnection client) { return client.info(section); } - }, node)); + }, node).getValue()); } /* @@ -301,17 +316,16 @@ public class LettuceClusterConnection extends LettuceConnection implements Assert.notNull(master, "Master must not be null!"); - final RedisClusterNode nodeToUse = topologyProvider.getTopology() - .lookup(master); + final RedisClusterNode nodeToUse = topologyProvider.getTopology().lookup(master); - return clusterCommandExecutor.executeCommandOnSingleNode( - new LettuceClusterCommandCallback>() { + return clusterCommandExecutor + .executeCommandOnSingleNode(new LettuceClusterCommandCallback>() { @Override public Set doInCluster(RedisClusterConnection client) { return LettuceConverters.toSetOfRedisClusterNodes(client.clusterSlaves(nodeToUse.getId())); } - }, master); + }, master).getValue(); } /* @@ -331,8 +345,8 @@ public class LettuceClusterConnection extends LettuceConnection implements public RedisClusterNode clusterGetNodeForSlot(int slot) { DirectFieldAccessor accessor = new DirectFieldAccessor(clusterClient); - return LettuceConverters.toRedisClusterNode(((Partitions) accessor.getPropertyValue("partitions")) - .getPartitionBySlot(slot)); + return LettuceConverters + .toRedisClusterNode(((Partitions) accessor.getPropertyValue("partitions")).getPartitionBySlot(slot)); } /* @@ -357,7 +371,7 @@ public class LettuceClusterConnection extends LettuceConnection implements public ClusterInfo doInCluster(RedisClusterConnection client) { return new ClusterInfo(LettuceConverters.toProperties(client.clusterInfo())); } - }); + }).getValue(); } /* @@ -542,14 +556,14 @@ public class LettuceClusterConnection extends LettuceConnection implements */ @Override public String ping() { - Collection ping = clusterCommandExecutor.executeCommandOnAllNodes( - new LettuceClusterCommandCallback() { + Collection ping = clusterCommandExecutor + .executeCommandOnAllNodes(new LettuceClusterCommandCallback() { @Override public String doInCluster(RedisClusterConnection connection) { return doPing(connection); } - }).values(); + }).resultsAsList(); for (String result : ping) { if (!ObjectUtils.nullSafeEquals("PONG", result)) { @@ -573,7 +587,7 @@ public class LettuceClusterConnection extends LettuceConnection implements public String doInCluster(RedisClusterConnection client) { return doPing(client); } - }, node); + }, node).getValue(); } protected String doPing(RedisClusterConnection client) { @@ -638,7 +652,7 @@ public class LettuceClusterConnection extends LettuceConnection implements public Long doInCluster(RedisClusterConnection client) { return client.lastsave().getTime(); } - }, node); + }, node).getValue(); } /* @@ -671,7 +685,7 @@ public class LettuceClusterConnection extends LettuceConnection implements public Long doInCluster(RedisClusterConnection client) { return client.dbsize(); } - }, node); + }, node).getValue(); } /* @@ -714,14 +728,14 @@ public class LettuceClusterConnection extends LettuceConnection implements @Override public Properties info(RedisClusterNode node) { - return LettuceConverters.toProperties(clusterCommandExecutor.executeCommandOnSingleNode( - new LettuceClusterCommandCallback() { + return LettuceConverters + .toProperties(clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback() { @Override public String doInCluster(RedisClusterConnection client) { return client.info(); } - }, node)); + }, node).getValue()); } /* @@ -731,14 +745,14 @@ public class LettuceClusterConnection extends LettuceConnection implements @Override public Set keys(RedisClusterNode node, final byte[] pattern) { - return LettuceConverters.toBytesSet(clusterCommandExecutor.executeCommandOnSingleNode( - new LettuceClusterCommandCallback>() { + return LettuceConverters.toBytesSet( + clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback>() { @Override public List doInCluster(RedisClusterConnection client) { return client.keys(pattern); } - }, node)); + }, node).getValue()); } /* @@ -754,7 +768,7 @@ public class LettuceClusterConnection extends LettuceConnection implements public byte[] doInCluster(RedisClusterConnection client) { return client.randomkey(); } - }, node); + }, node).getValue(); } /* @@ -886,16 +900,13 @@ public class LettuceClusterConnection extends LettuceConnection implements return super.mGet(keys); } - Map nodeResult = this.clusterCommandExecutor.executeMuliKeyCommand( - new LettuceMultiKeyClusterCommandCallback() { + return this.clusterCommandExecutor.executeMuliKeyCommand(new LettuceMultiKeyClusterCommandCallback() { - @Override - public byte[] doInCluster(RedisClusterConnection client, byte[] key) { - return client.get(key); - } - }, Arrays.asList(keys)); - - return new ArrayList(nodeResult.values()); + @Override + public byte[] doInCluster(RedisClusterConnection client, byte[] key) { + return client.get(key); + } + }, Arrays.asList(keys)).resultsAsListSortBy(keys); } /* @@ -948,17 +959,19 @@ public class LettuceClusterConnection extends LettuceConnection implements return super.bLPop(timeout, keys); } - Map> nodeResult = this.clusterCommandExecutor.executeMuliKeyCommand( - new LettuceMultiKeyClusterCommandCallback>() { + List> resultList = this.clusterCommandExecutor + .executeMuliKeyCommand(new LettuceMultiKeyClusterCommandCallback>() { @Override public KeyValue doInCluster(RedisClusterConnection client, byte[] key) { return client.blpop(timeout, key); } - }, Arrays.asList(keys)); + }, Arrays.asList(keys)).resultsAsList(); - for (KeyValue partial : nodeResult.values()) { - return LettuceConverters.toBytesList(partial); + for (KeyValue kv : resultList) { + if (kv != null) { + return LettuceConverters.toBytesList(kv); + } } return Collections.emptyList(); @@ -975,17 +988,19 @@ public class LettuceClusterConnection extends LettuceConnection implements return super.bRPop(timeout, keys); } - Map> nodeResult = this.clusterCommandExecutor.executeMuliKeyCommand( - new LettuceMultiKeyClusterCommandCallback>() { + List> resultList = this.clusterCommandExecutor + .executeMuliKeyCommand(new LettuceMultiKeyClusterCommandCallback>() { @Override public KeyValue doInCluster(RedisClusterConnection client, byte[] key) { return client.brpop(timeout, key); } - }, Arrays.asList(keys)); + }, Arrays.asList(keys)).resultsAsList(); - for (KeyValue partial : nodeResult.values()) { - return LettuceConverters.toBytesList(partial); + for (KeyValue kv : resultList) { + if (kv != null) { + return LettuceConverters.toBytesList(kv); + } } return Collections.emptyList(); @@ -1057,19 +1072,19 @@ public class LettuceClusterConnection extends LettuceConnection implements return super.sInter(keys); } - Map> nodeResult = this.clusterCommandExecutor.executeMuliKeyCommand( - new LettuceMultiKeyClusterCommandCallback>() { + Collection> nodeResult = this.clusterCommandExecutor + .executeMuliKeyCommand(new LettuceMultiKeyClusterCommandCallback>() { @Override public Set doInCluster(RedisClusterConnection client, byte[] key) { return client.smembers(key); } - }, Arrays.asList(keys)); + }, Arrays.asList(keys)).resultsAsList(); ByteArraySet result = null; - for (Entry> entry : nodeResult.entrySet()) { + for (Set entry : nodeResult) { - ByteArraySet tmp = new ByteArraySet(entry.getValue()); + ByteArraySet tmp = new ByteArraySet(entry); if (result == null) { result = tmp; } else { @@ -1118,18 +1133,18 @@ public class LettuceClusterConnection extends LettuceConnection implements return super.sUnion(keys); } - Map> nodeResult = this.clusterCommandExecutor.executeMuliKeyCommand( - new LettuceMultiKeyClusterCommandCallback>() { + Collection> nodeResult = this.clusterCommandExecutor + .executeMuliKeyCommand(new LettuceMultiKeyClusterCommandCallback>() { @Override public Set doInCluster(RedisClusterConnection client, byte[] key) { return client.smembers(key); } - }, Arrays.asList(keys)); + }, Arrays.asList(keys)).resultsAsList(); ByteArraySet result = new ByteArraySet(); - for (Entry> entry : nodeResult.entrySet()) { - result.addAll(entry.getValue()); + for (Set entry : nodeResult) { + result.addAll(entry); } if (result.isEmpty()) { @@ -1174,20 +1189,20 @@ public class LettuceClusterConnection extends LettuceConnection implements byte[][] others = Arrays.copyOfRange(keys, 1, keys.length - 1); ByteArraySet values = new ByteArraySet(sMembers(source)); - Map> nodeResult = clusterCommandExecutor.executeMuliKeyCommand( - new LettuceMultiKeyClusterCommandCallback>() { + Collection> nodeResult = clusterCommandExecutor + .executeMuliKeyCommand(new LettuceMultiKeyClusterCommandCallback>() { @Override public Set doInCluster(RedisClusterConnection client, byte[] key) { return client.smembers(key); } - }, Arrays.asList(others)); + }, Arrays.asList(others)).resultsAsList(); if (values.isEmpty()) { return Collections.emptySet(); } - for (Set toSubstract : nodeResult.values()) { + for (Set toSubstract : nodeResult) { values.removeAll(toSubstract); } @@ -1310,19 +1325,19 @@ public class LettuceClusterConnection extends LettuceConnection implements @Override public List getConfig(final String pattern) { - Map> mapResult = clusterCommandExecutor + List>> mapResult = clusterCommandExecutor .executeCommandOnAllNodes(new LettuceClusterCommandCallback>() { @Override public List doInCluster(RedisClusterConnection client) { return client.configGet(pattern); } - }); + }).getResults(); List result = new ArrayList(); - for (Entry> entry : mapResult.entrySet()) { + for (NodeResult> entry : mapResult) { - String prefix = entry.getKey().asString(); + String prefix = entry.getNode().asString(); int i = 0; for (String value : entry.getValue()) { result.add((i++ % 2 == 0 ? (prefix + ".") : "") + value); @@ -1345,7 +1360,7 @@ public class LettuceClusterConnection extends LettuceConnection implements public List doInCluster(RedisClusterConnection client) { return client.configGet(pattern); } - }, node); + }, node).getValue(); } /* @@ -1420,14 +1435,14 @@ public class LettuceClusterConnection extends LettuceConnection implements @Override public Long time() { - return convertListOfStringToTime(clusterCommandExecutor - .executeCommandOnArbitraryNode(new LettuceClusterCommandCallback>() { + return convertListOfStringToTime( + clusterCommandExecutor.executeCommandOnArbitraryNode(new LettuceClusterCommandCallback>() { @Override public List doInCluster(RedisClusterConnection client) { return client.time(); } - })); + }).getValue()); } /* @@ -1437,14 +1452,14 @@ public class LettuceClusterConnection extends LettuceConnection implements @Override public Long time(RedisClusterNode node) { - return convertListOfStringToTime(clusterCommandExecutor.executeCommandOnSingleNode( - new LettuceClusterCommandCallback>() { + return convertListOfStringToTime( + clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback>() { @Override public List doInCluster(RedisClusterConnection client) { return client.time(); } - }, node)); + }, node).getValue()); } private Long convertListOfStringToTime(List serverTimeInformation) { @@ -1464,17 +1479,16 @@ public class LettuceClusterConnection extends LettuceConnection implements @Override public List getClientList() { - Map map = clusterCommandExecutor - .executeCommandOnAllNodes(new LettuceClusterCommandCallback() { + List map = clusterCommandExecutor.executeCommandOnAllNodes(new LettuceClusterCommandCallback() { - @Override - public String doInCluster(RedisClusterConnection client) { - return client.clientList(); - } - }); + @Override + public String doInCluster(RedisClusterConnection client) { + return client.clientList(); + } + }).resultsAsList(); ArrayList result = new ArrayList(); - for (String infos : map.values()) { + for (String infos : map) { result.addAll(LettuceConverters.toListOfRedisClientInformation(infos)); } return result; @@ -1487,14 +1501,14 @@ public class LettuceClusterConnection extends LettuceConnection implements @Override public List getClientList(RedisClusterNode node) { - return LettuceConverters.toListOfRedisClientInformation(clusterCommandExecutor.executeCommandOnSingleNode( - new LettuceClusterCommandCallback() { + return LettuceConverters.toListOfRedisClientInformation( + clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback() { @Override public String doInCluster(RedisClusterConnection client) { return client.clientList(); } - }, node)); + }, node).getValue()); } /* @@ -1504,14 +1518,22 @@ public class LettuceClusterConnection extends LettuceConnection implements @Override public Map> clusterGetMasterSlaveMap() { - return clusterCommandExecutor.executeCommandAsyncOnNodes( - new LettuceClusterCommandCallback>() { + List>> nodeResults = clusterCommandExecutor + .executeCommandAsyncOnNodes(new LettuceClusterCommandCallback>() { @Override public Set doInCluster(RedisClusterConnection client) { return Converters.toSetOfRedisClusterNodes(client.clusterSlaves(client.clusterMyId())); } - }, topologyProvider.getTopology().getActiveMasterNodes()); + }, topologyProvider.getTopology().getActiveMasterNodes()).getResults(); + + Map> result = new LinkedHashMap>(); + + for (NodeResult> nodeResult : nodeResults) { + result.put(nodeResult.getNode(), nodeResult.getValue()); + } + + return result; } /** @@ -1521,8 +1543,8 @@ public class LettuceClusterConnection extends LettuceConnection implements * @param * @since 1.7 */ - protected interface LettuceClusterCommandCallback extends - ClusterCommandCallback, T> {} + protected interface LettuceClusterCommandCallback + extends ClusterCommandCallback, T> {} /** * Lettuce specific implementation of {@link MultiKeyClusterCommandCallback}. @@ -1531,8 +1553,8 @@ public class LettuceClusterConnection extends LettuceConnection implements * @param * @since 1.7 */ - protected interface LettuceMultiKeyClusterCommandCallback extends - MultiKeyClusterCommandCallback, T> { + protected interface LettuceMultiKeyClusterCommandCallback + extends MultiKeyClusterCommandCallback, T> { } @@ -1606,8 +1628,8 @@ public class LettuceClusterConnection extends LettuceConnection implements */ @Override public ClusterTopology getTopology() { - return new ClusterTopology(new LinkedHashSet(LettuceConverters.partitionsToClusterNodes(client - .getPartitions()))); + return new ClusterTopology( + new LinkedHashSet(LettuceConverters.partitionsToClusterNodes(client.getPartitions()))); } } diff --git a/src/main/java/org/springframework/data/redis/core/RedisTemplate.java b/src/main/java/org/springframework/data/redis/core/RedisTemplate.java index 933343ca5..7a9dcb9ad 100644 --- a/src/main/java/org/springframework/data/redis/core/RedisTemplate.java +++ b/src/main/java/org/springframework/data/redis/core/RedisTemplate.java @@ -722,7 +722,7 @@ public class RedisTemplate extends RedisAccessor implements RedisOperation } }, true); - return keySerializer != null ? SerializationUtils.deserialize(rawKeys, keySerializer) : rawKeys; + return keySerializer != null ? SerializationUtils.deserialize(rawKeys, keySerializer) : (Set) rawKeys; } public Boolean persist(K key) { diff --git a/src/test/java/org/springframework/data/redis/connection/ClusterCommandExecutorUnitTests.java b/src/test/java/org/springframework/data/redis/connection/ClusterCommandExecutorUnitTests.java index ca025ab40..fd1bd31d4 100644 --- a/src/test/java/org/springframework/data/redis/connection/ClusterCommandExecutorUnitTests.java +++ b/src/test/java/org/springframework/data/redis/connection/ClusterCommandExecutorUnitTests.java @@ -25,7 +25,6 @@ import static org.springframework.data.redis.test.util.MockitoUtils.*; import java.util.Arrays; import java.util.HashSet; import java.util.LinkedHashSet; -import java.util.Map; import org.hamcrest.core.IsInstanceOf; import org.junit.After; @@ -43,6 +42,7 @@ import org.springframework.data.redis.ClusterRedirectException; import org.springframework.data.redis.PassThroughExceptionTranslationStrategy; import org.springframework.data.redis.TooManyClusterRedirectionsException; import org.springframework.data.redis.connection.ClusterCommandExecutor.ClusterCommandCallback; +import org.springframework.data.redis.connection.ClusterCommandExecutor.MulitNodeResult; import org.springframework.data.redis.connection.ClusterCommandExecutor.MultiKeyClusterCommandCallback; import org.springframework.data.redis.connection.RedisClusterNode.LinkState; import org.springframework.data.redis.connection.RedisClusterNode.SlotRange; @@ -77,8 +77,7 @@ public class ClusterCommandExecutorUnitTests { .withId("3b9b8192a874fa8f1f09dbc0ee20afab5738eee7").promotedAs(NodeType.MASTER).linkState(LinkState.CONNECTED) .build(); static final RedisClusterNode CLUSTER_NODE_2_LOOKUP = RedisClusterNode.newRedisClusterNode() - .withId("0f2ee5df45d18c50aca07228cc18b1da96fd5e84") - .build(); + .withId("0f2ee5df45d18c50aca07228cc18b1da96fd5e84").build(); static final RedisClusterNode UNKNOWN_CLUSTER_NODE = new RedisClusterNode("8.8.8.8", 7379, null); @@ -149,7 +148,8 @@ public class ClusterCommandExecutorUnitTests { @Test public void executeCommandOnSingleNodeByHostAndPortShouldBeExecutedCorrectly() { - executor.executeCommandOnSingleNode(COMMAND_CALLBACK, new RedisClusterNode(CLUSTER_NODE_2_HOST, CLUSTER_NODE_2_PORT)); + executor.executeCommandOnSingleNode(COMMAND_CALLBACK, + new RedisClusterNode(CLUSTER_NODE_2_HOST, CLUSTER_NODE_2_PORT)); verify(con2, times(1)).theWheelWeavesAsTheWheelWills(); } @@ -216,8 +216,9 @@ public class ClusterCommandExecutorUnitTests { new MockClusterResourceProvider(), new PassThroughExceptionTranslationStrategy(exceptionConverter), new ConcurrentTaskExecutor(new SyncTaskExecutor())); - executor.executeCommandAsyncOnNodes(COMMAND_CALLBACK, Arrays.asList(new RedisClusterNode(CLUSTER_NODE_1_HOST, CLUSTER_NODE_1_PORT), - new RedisClusterNode(CLUSTER_NODE_2_HOST, CLUSTER_NODE_2_PORT))); + executor.executeCommandAsyncOnNodes(COMMAND_CALLBACK, + Arrays.asList(new RedisClusterNode(CLUSTER_NODE_1_HOST, CLUSTER_NODE_1_PORT), + new RedisClusterNode(CLUSTER_NODE_2_HOST, CLUSTER_NODE_2_PORT))); verify(con1, times(1)).theWheelWeavesAsTheWheelWills(); verify(con2, times(1)).theWheelWeavesAsTheWheelWills(); @@ -234,8 +235,8 @@ public class ClusterCommandExecutorUnitTests { new MockClusterResourceProvider(), new PassThroughExceptionTranslationStrategy(exceptionConverter), new ConcurrentTaskExecutor(new SyncTaskExecutor())); - executor.executeCommandAsyncOnNodes(COMMAND_CALLBACK, Arrays.asList(new RedisClusterNode(CLUSTER_NODE_1.id), - CLUSTER_NODE_2_LOOKUP)); + executor.executeCommandAsyncOnNodes(COMMAND_CALLBACK, + Arrays.asList(new RedisClusterNode(CLUSTER_NODE_1.id), CLUSTER_NODE_2_LOOKUP)); verify(con1, times(1)).theWheelWeavesAsTheWheelWills(); verify(con2, times(1)).theWheelWeavesAsTheWheelWills(); @@ -252,8 +253,8 @@ public class ClusterCommandExecutorUnitTests { new MockClusterResourceProvider(), new PassThroughExceptionTranslationStrategy(exceptionConverter), new ConcurrentTaskExecutor(new SyncTaskExecutor())); - executor.executeCommandAsyncOnNodes(COMMAND_CALLBACK, Arrays.asList( - new RedisClusterNode("unknown"), CLUSTER_NODE_2_LOOKUP)); + executor.executeCommandAsyncOnNodes(COMMAND_CALLBACK, + Arrays.asList(new RedisClusterNode("unknown"), CLUSTER_NODE_2_LOOKUP)); } /** @@ -306,32 +307,29 @@ public class ClusterCommandExecutorUnitTests { when(con2.theWheelWeavesAsTheWheelWills()).thenReturn("mat"); when(con3.theWheelWeavesAsTheWheelWills()).thenReturn("perrin"); - Map result = executor.executeCommandOnAllNodes(COMMAND_CALLBACK); + MulitNodeResult result = executor.executeCommandOnAllNodes(COMMAND_CALLBACK); - assertThat(result.keySet(), hasItems(CLUSTER_NODE_1, CLUSTER_NODE_2, CLUSTER_NODE_3)); - assertThat(result.values(), hasItems("rand", "mat", "perrin")); + assertThat(result.resultsAsList(), hasItems("rand", "mat", "perrin")); } /** * @see DATAREDIS-315 + * @see DATAREDIS-467 */ @Test public void executeMultikeyCommandShouldRunCommandAcrossCluster() { // key-1 and key-9 map both to node1 ArgumentCaptor captor = ArgumentCaptor.forClass(byte[].class); - when(con1.bloodAndAshes(captor.capture())).thenReturn("rand"); + when(con1.bloodAndAshes(captor.capture())).thenReturn("rand").thenReturn("egwene"); when(con2.bloodAndAshes(any(byte[].class))).thenReturn("mat"); when(con3.bloodAndAshes(any(byte[].class))).thenReturn("perrin"); - Map result = executor.executeMuliKeyCommand( - MULTIKEY_CALLBACK, - new HashSet(Arrays.asList("key-1".getBytes(), "key-2".getBytes(), "key-3".getBytes(), - "key-9".getBytes()))); + MulitNodeResult result = executor.executeMuliKeyCommand(MULTIKEY_CALLBACK, new HashSet( + Arrays.asList("key-1".getBytes(), "key-2".getBytes(), "key-3".getBytes(), "key-9".getBytes()))); - assertThat(result.keySet(), hasItems(CLUSTER_NODE_1, CLUSTER_NODE_2, CLUSTER_NODE_3)); - assertThat(result.values(), hasItems("rand", "mat", "perrin")); + assertThat(result.resultsAsList(), hasItems("rand", "mat", "perrin", "egwene")); // check that 2 keys have been routed to node1 assertThat(captor.getAllValues().size(), is(2)); @@ -389,8 +387,8 @@ public class ClusterCommandExecutorUnitTests { @Override public ClusterTopology getTopology() { - return new ClusterTopology(new LinkedHashSet(Arrays.asList(CLUSTER_NODE_1, CLUSTER_NODE_2, - CLUSTER_NODE_3))); + return new ClusterTopology( + new LinkedHashSet(Arrays.asList(CLUSTER_NODE_1, CLUSTER_NODE_2, CLUSTER_NODE_3))); } } diff --git a/src/test/java/org/springframework/data/redis/connection/jedis/JedisClusterConnectionTests.java b/src/test/java/org/springframework/data/redis/connection/jedis/JedisClusterConnectionTests.java index c1418be06..da207dab9 100644 --- a/src/test/java/org/springframework/data/redis/connection/jedis/JedisClusterConnectionTests.java +++ b/src/test/java/org/springframework/data/redis/connection/jedis/JedisClusterConnectionTests.java @@ -16,6 +16,7 @@ package org.springframework.data.redis.connection.jedis; import static org.hamcrest.CoreMatchers.*; +import static org.hamcrest.collection.IsIterableContainingInOrder.*; import static org.hamcrest.number.IsCloseTo.*; import static org.junit.Assert.*; import static org.springframework.data.redis.connection.ClusterTestVariables.*; @@ -536,7 +537,7 @@ public class JedisClusterConnectionTests implements ClusterConnectionTests { nativeConnection.set(SAME_SLOT_KEY_2_BYTES, VALUE_2_BYTES); assertThat(clusterConnection.mGet(SAME_SLOT_KEY_1_BYTES, SAME_SLOT_KEY_2_BYTES), - hasItems(VALUE_1_BYTES, VALUE_2_BYTES)); + contains(VALUE_1_BYTES, VALUE_2_BYTES)); } /** @@ -548,7 +549,7 @@ public class JedisClusterConnectionTests implements ClusterConnectionTests { nativeConnection.set(KEY_1_BYTES, VALUE_1_BYTES); nativeConnection.set(KEY_2_BYTES, VALUE_2_BYTES); - assertThat(clusterConnection.mGet(KEY_1_BYTES, KEY_2_BYTES), hasItems(VALUE_1_BYTES, VALUE_2_BYTES)); + assertThat(clusterConnection.mGet(KEY_1_BYTES, KEY_2_BYTES), contains(VALUE_1_BYTES, VALUE_2_BYTES)); } /** @@ -2088,29 +2089,23 @@ public class JedisClusterConnectionTests implements ClusterConnectionTests { assertThat(values, hasItems(JedisConverters.toBytes("a"), JedisConverters.toBytes("b"), JedisConverters.toBytes("c"))); - assertThat( - values, - not(hasItems(JedisConverters.toBytes("d"), JedisConverters.toBytes("e"), JedisConverters.toBytes("f"), - JedisConverters.toBytes("g")))); + assertThat(values, not(hasItems(JedisConverters.toBytes("d"), JedisConverters.toBytes("e"), + JedisConverters.toBytes("f"), JedisConverters.toBytes("g")))); values = clusterConnection.zRangeByLex(KEY_1_BYTES, Range.range().lt("c")); assertThat(values, hasItems(JedisConverters.toBytes("a"), JedisConverters.toBytes("b"))); assertThat(values, not(hasItem(JedisConverters.toBytes("c")))); values = clusterConnection.zRangeByLex(KEY_1_BYTES, Range.range().gte("aaa").lt("g")); - assertThat( - values, - hasItems(JedisConverters.toBytes("b"), JedisConverters.toBytes("c"), JedisConverters.toBytes("d"), - JedisConverters.toBytes("e"), JedisConverters.toBytes("f"))); + assertThat(values, hasItems(JedisConverters.toBytes("b"), JedisConverters.toBytes("c"), + JedisConverters.toBytes("d"), JedisConverters.toBytes("e"), JedisConverters.toBytes("f"))); assertThat(values, not(hasItems(JedisConverters.toBytes("a"), JedisConverters.toBytes("g")))); values = clusterConnection.zRangeByLex(KEY_1_BYTES, Range.range().gte("e")); assertThat(values, hasItems(JedisConverters.toBytes("e"), JedisConverters.toBytes("f"), JedisConverters.toBytes("g"))); - assertThat( - values, - not(hasItems(JedisConverters.toBytes("a"), JedisConverters.toBytes("b"), JedisConverters.toBytes("c"), - JedisConverters.toBytes("d")))); + assertThat(values, not(hasItems(JedisConverters.toBytes("a"), JedisConverters.toBytes("b"), + JedisConverters.toBytes("c"), JedisConverters.toBytes("d")))); } /** @@ -2118,7 +2113,7 @@ public class JedisClusterConnectionTests implements ClusterConnectionTests { */ @Test public void infoShouldCollectionInfoFromAllClusterNodes() { - assertThat(clusterConnection.info().size(), is(3)); + assertThat(Double.valueOf(clusterConnection.info().size()), closeTo(245d, 35d)); } /** @@ -2288,8 +2283,8 @@ public class JedisClusterConnectionTests implements ClusterConnectionTests { @Test public void clusterGetSlavesShouldReturnSlaveCorrectly() { - Set slaves = clusterConnection.clusterGetSlaves(new RedisClusterNode(CLUSTER_HOST, - MASTER_NODE_1_PORT)); + Set slaves = clusterConnection + .clusterGetSlaves(new RedisClusterNode(CLUSTER_HOST, MASTER_NODE_1_PORT)); assertThat(slaves.size(), is(1)); assertThat(slaves, hasItem(new RedisClusterNode(CLUSTER_HOST, SLAVEOF_NODE_1_PORT))); diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionTests.java index 93e568158..412873bdf 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionTests.java @@ -16,6 +16,7 @@ package org.springframework.data.redis.connection.lettuce; import static org.hamcrest.CoreMatchers.*; +import static org.hamcrest.collection.IsIterableContainingInOrder.*; import static org.hamcrest.number.IsCloseTo.*; import static org.junit.Assert.*; import static org.springframework.data.redis.connection.ClusterTestVariables.*; @@ -529,7 +530,7 @@ public class LettuceClusterConnectionTests implements ClusterConnectionTests { nativeConnection.set(SAME_SLOT_KEY_2, VALUE_2); assertThat(clusterConnection.mGet(SAME_SLOT_KEY_1_BYTES, SAME_SLOT_KEY_2_BYTES), - hasItems(VALUE_1_BYTES, VALUE_2_BYTES)); + contains(VALUE_1_BYTES, VALUE_2_BYTES)); } /** @@ -541,7 +542,7 @@ public class LettuceClusterConnectionTests implements ClusterConnectionTests { nativeConnection.set(KEY_1, VALUE_1); nativeConnection.set(KEY_2, VALUE_2); - assertThat(clusterConnection.mGet(KEY_1_BYTES, KEY_2_BYTES), hasItems(VALUE_1_BYTES, VALUE_2_BYTES)); + assertThat(clusterConnection.mGet(KEY_1_BYTES, KEY_2_BYTES), contains(VALUE_1_BYTES, VALUE_2_BYTES)); } /** @@ -2103,7 +2104,7 @@ public class LettuceClusterConnectionTests implements ClusterConnectionTests { */ @Test public void infoShouldCollectionInfoFromAllClusterNodes() { - assertThat(clusterConnection.info().size(), is(3)); + assertThat(Double.valueOf(clusterConnection.info().size()), closeTo(245d, 35d)); } /**