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)); } /**