diff --git a/src/main/asciidoc/index.adoc b/src/main/asciidoc/index.adoc index f3997a7f1..53ebd3bbd 100644 --- a/src/main/asciidoc/index.adoc +++ b/src/main/asciidoc/index.adoc @@ -1,5 +1,5 @@ = Spring Data Redis -Costin Leau, Jennifer Hickey, Christoph Strobl, Thomas Darimont +Costin Leau, Jennifer Hickey, Christoph Strobl, Thomas Darimont, Mark Paluch :revnumber: {version} :revdate: {localdate} :toc: diff --git a/src/asciidoc/reference/redis-cluster.adoc b/src/main/asciidoc/reference/redis-cluster.adoc similarity index 81% rename from src/asciidoc/reference/redis-cluster.adoc rename to src/main/asciidoc/reference/redis-cluster.adoc index 281458962..e7e467dd4 100644 --- a/src/asciidoc/reference/redis-cluster.adoc +++ b/src/main/asciidoc/reference/redis-cluster.adoc @@ -7,28 +7,50 @@ TIP: Redis Cluster is only supported by <> and < initialClusterNodes; + List nodes; + + /** + * Get initial collection of known cluster nodes in format {@code host:port}. + * + * @return + */ + public List getNodes() { + return nodes; + } + + public void setNodes(List nodes) { + this.nodes = nodes; + } +} + +@Configuration +public class AppConfig { + + /** + * Type safe representation of application.properties + */ + @Autowired ClusterConfigurationProperties clusterProperties; public @Bean RedisConnectionFactory connectionFactory() { return new JedisConnectionFactory( - new RedisClusterConfiguration(initialClusterNodes)); + new RedisClusterConfiguration(clusterProperties.getNodes())); } } ---- @@ -39,9 +61,11 @@ NOTE: The initial configuration points driver libraries to an initial set of clu == Working With Redis Cluster Connection As mentioned above Redis Cluster behaves different from single node Redis or even a Sentinel monitored master slave environment. This is reasoned by the automatic sharding that maps a key to one of 16384 slots which are distributed across the nodes. Therefore commands that involve more than one key must assert that all keys map to the exact same slot in order to avoid cross slot execution errors. -Further on, hence a single cluster node, only servers a dedicated set of keys, commands issued against one particular server only return results for those keys served by the server. As a very simple example take the `KEYS` command. When issued to a server in cluster environment it only returns the keys served by the node the request is sent to and not necessarily all keys within the cluster. So to get all keys in cluster environment it is necessary to read the keys from at least all known master nodes. +Further on, hence a single cluster node, only serves a dedicated set of keys, commands issued against one particular server only return results for those keys served by the server. As a very simple example take the `KEYS` command. When issued to a server in cluster environment it only returns the keys served by the node the request is sent to and not necessarily all keys within the cluster. So to get all keys in cluster environment it is necessary to read the keys from at least all known master nodes. -While redirects for to a specific keys to the corresponding slot serving node are handled by the driver libraries, higher level functions like collecting information across nodes, or sending commands to all nodes in the cluster that are covered by `RedisClusterConnection`. Picking up the keys example from just before, this means, that the `keys(pattern)` method picks up every master node in cluster and simultaneously executes the `KEYS` command on every single one, while picking up the results and returning the cumulated set of keys. To just request the keys of a single node `RedisClusterConnection` provides overloads for those (like `keys(node, pattern)` ). +While redirects for to a specific keys to the corresponding slot serving node are handled by the driver libraries, higher level functions like collecting information across nodes, or sending commands to all nodes in the cluster that are covered by `RedisClusterConnection`. Picking up the keys example from just before, this means, that the `keys(pattern)` method picks up every master node in cluster and simultaneously executes the `KEYS` command on every single one, while picking up the results and returning the cumulated set of keys. To just request the keys of a single node `RedisClusterConnection` provides overloads for those (like `keys(node, pattern)` ). + +A `RedisClusterNode` can be obtained from `RedisClusterConnection.clusterGetNodes` or it can be constructed using either host and port or the node Id. .Sample of Running Commands Across the Cluster ==== @@ -113,7 +137,7 @@ connection.mGet("foo", "bar"); < -> 127.0.0.1:7381 GET foo ==== -TIP: The above provides simple examples to demonstrate the general strategy followed by Spring Data Redis. Be aware that some operations might require loading huge amounts of data into memory in order to compute the desired command. Additionally not all cross slot requests can safely be ported to multiple single slot requests and will error if misused (eg. `PFCOUNT` ). +TIP: The above provided simple examples to demonstrate the general strategy followed by Spring Data Redis. Be aware that some operations might require loading huge amounts of data into memory in order to compute the desired command. Additionally not all cross slot requests can safely be ported to multiple single slot requests and will error if misused (eg. `PFCOUNT` ). == Working With RedisTemplate and ClusterOperations diff --git a/src/main/java/org/springframework/data/redis/ClusterStateFailureExeption.java b/src/main/java/org/springframework/data/redis/ClusterStateFailureException.java similarity index 78% rename from src/main/java/org/springframework/data/redis/ClusterStateFailureExeption.java rename to src/main/java/org/springframework/data/redis/ClusterStateFailureException.java index 35a766cbf..003a85457 100644 --- a/src/main/java/org/springframework/data/redis/ClusterStateFailureExeption.java +++ b/src/main/java/org/springframework/data/redis/ClusterStateFailureException.java @@ -23,28 +23,29 @@ import org.springframework.dao.DataAccessResourceFailureException; * and so on. * * @author Christoph Strobl + * @author Mark Paluch * @since 1.7 */ -public class ClusterStateFailureExeption extends DataAccessResourceFailureException { +public class ClusterStateFailureException extends DataAccessResourceFailureException { private static final long serialVersionUID = 333399051713240852L; /** - * Creates new {@link ClusterStateFailureExeption}. + * Creates new {@link ClusterStateFailureException}. * * @param msg */ - public ClusterStateFailureExeption(String msg) { + public ClusterStateFailureException(String msg) { super(msg); } /** - * Creates new {@link ClusterStateFailureExeption}. + * Creates new {@link ClusterStateFailureException}. * * @param msg * @param cause */ - public ClusterStateFailureExeption(String msg, Throwable cause) { + public ClusterStateFailureException(String msg, Throwable cause) { super(msg, cause); } 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 bd70718fb..83bfdef6e 100644 --- a/src/main/java/org/springframework/data/redis/connection/ClusterCommandExecutor.java +++ b/src/main/java/org/springframework/data/redis/connection/ClusterCommandExecutor.java @@ -32,6 +32,7 @@ import org.springframework.beans.factory.DisposableBean; import org.springframework.core.task.AsyncTaskExecutor; import org.springframework.dao.DataAccessException; 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.scheduling.concurrent.ThreadPoolTaskExecutor; @@ -42,6 +43,7 @@ import org.springframework.util.Assert; * {@link AsyncTaskExecutor} the execution behavior can be influenced. * * @author Christoph Strobl + * @author Mark Paluch * @since 1.7 */ public class ClusterCommandExecutor implements DisposableBean { @@ -130,7 +132,9 @@ public class ClusterCommandExecutor implements DisposableBean { redirectCount, maxRedirects)); } - S client = this.resourceProvider.getResourceForSpecificNode(node); + RedisClusterNode nodeToUse = lookupNode(node); + + S client = this.resourceProvider.getResourceForSpecificNode(nodeToUse); Assert.notNull(client, "Could not acquire resource for node. Is your cluster info up to date?"); try { @@ -146,7 +150,21 @@ public class ClusterCommandExecutor implements DisposableBean { throw translatedException != null ? translatedException : ex; } } finally { - this.resourceProvider.returnResourceForSpecificNode(node, client); + this.resourceProvider.returnResourceForSpecificNode(nodeToUse, client); + } + } + + /** + * Lookup node from the topology. + * @param node + * @return + * @throws IllegalArgumentException in case the node could not be resolved to a topology-known node + */ + private RedisClusterNode lookupNode(RedisClusterNode node) { + try { + return topologyProvider.getTopology().lookup(node); + }catch (ClusterStateFailureException e){ + throw new IllegalArgumentException(String.format("Node %s is unknown to cluster", node), e); } } @@ -166,6 +184,7 @@ public class ClusterCommandExecutor implements DisposableBean { * @param nodes * @return * @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) { @@ -173,8 +192,20 @@ public class ClusterCommandExecutor implements DisposableBean { Assert.notNull(callback, "Callback must not be null!"); Assert.notNull(nodes, "Nodes must not be null!"); - Map> futures = new LinkedHashMap>(); + + List resolvedRedisClusterNodes = new ArrayList(); + ClusterTopology topology = topologyProvider.getTopology(); + for (final RedisClusterNode node : nodes) { + try { + resolvedRedisClusterNodes.add(topology.lookup(node)); + } catch (ClusterStateFailureException e) { + throw new IllegalArgumentException(String.format("Node %s is unknown to cluster", node), e); + } + } + + Map> futures = new LinkedHashMap>(); + for (final RedisClusterNode node : resolvedRedisClusterNodes) { futures.put(node, executor.submit(new Callable() { diff --git a/src/main/java/org/springframework/data/redis/connection/ClusterTopology.java b/src/main/java/org/springframework/data/redis/connection/ClusterTopology.java index 70b5f3483..f37938642 100644 --- a/src/main/java/org/springframework/data/redis/connection/ClusterTopology.java +++ b/src/main/java/org/springframework/data/redis/connection/ClusterTopology.java @@ -19,13 +19,15 @@ import java.util.Collections; import java.util.LinkedHashSet; import java.util.Set; -import org.springframework.data.redis.ClusterStateFailureExeption; +import org.springframework.data.redis.ClusterStateFailureException; import org.springframework.util.Assert; +import org.springframework.util.StringUtils; /** * {@link ClusterTopology} holds snapshot like information about {@link RedisClusterNode}s. * * @author Christoph Strobl + * @author Mark Paluch * @since 1.7 */ public class ClusterTopology { @@ -122,7 +124,7 @@ public class ClusterTopology { * * @param key must not be {@literal null}. * @return - * @throws ClusterStateFailureExeption + * @throws ClusterStateFailureException */ public RedisClusterNode getKeyServingMasterNode(byte[] key) { @@ -134,7 +136,7 @@ public class ClusterTopology { return node; } } - throw new ClusterStateFailureExeption(String.format("Could not find master node serving slot %s for key '%s',", + throw new ClusterStateFailureException(String.format("Could not find master node serving slot %s for key '%s',", slot, key)); } @@ -144,7 +146,7 @@ public class ClusterTopology { * @param host must not be {@literal null}. * @param port * @return - * @throws ClusterStateFailureExeption + * @throws ClusterStateFailureException */ public RedisClusterNode lookup(String host, int port) { @@ -153,10 +155,58 @@ public class ClusterTopology { return node; } } - throw new ClusterStateFailureExeption(String.format( + throw new ClusterStateFailureException(String.format( "Could not find node at %s:%s. Is your cluster info up to date?", host, port)); } + /** + * Get the {@link RedisClusterNode} matching given {@literal nodeId}. + * + * @param nodeId must not be {@literal null}. + * @return + * @throws ClusterStateFailureException + */ + public RedisClusterNode lookup(String nodeId) { + + Assert.notNull(nodeId, "NodeId must not be null"); + + for (RedisClusterNode node : nodes) { + if (nodeId.equals(node.getId())) { + return node; + } + } + throw new ClusterStateFailureException(String.format( + "Could not find node at %s. Is your cluster info up to date?", nodeId)); + } + + /** + * Get the {@link RedisClusterNode} matching matching either {@link RedisClusterNode#getHost() host} and {@link RedisClusterNode#getPort() port} + * or {@link RedisClusterNode#getId() nodeId} + * + * @param node must not be {@literal null} + * @return + * @throws ClusterStateFailureException + */ + public RedisClusterNode lookup(RedisClusterNode node) { + + Assert.notNull(node, "RedisClusterNode must not be null"); + + if(nodes.contains(node) && StringUtils.hasText(node.getHost()) && StringUtils.hasText(node.getId())){ + return node; + } + + if (StringUtils.hasText(node.getHost())) { + return lookup(node.getHost(), node.getPort()); + } + + if (StringUtils.hasText(node.getId())) { + return lookup(node.getId()); + } + + throw new ClusterStateFailureException( + String.format("Could not find node at %s. Have you provided either host and port or the nodeId?", node)); + } + /** * @param key must not be {@literal null}. * @return {@literal null}. diff --git a/src/main/java/org/springframework/data/redis/connection/RedisClusterCommands.java b/src/main/java/org/springframework/data/redis/connection/RedisClusterCommands.java index ef6df7ff7..8fd96356e 100644 --- a/src/main/java/org/springframework/data/redis/connection/RedisClusterCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/RedisClusterCommands.java @@ -22,9 +22,12 @@ import java.util.Map; import org.springframework.data.redis.connection.RedisClusterNode.SlotRange; /** - * Interface for the {@literal cluster} commands supported by Redis. + * Interface for the {@literal cluster} commands supported by Redis. A {@link RedisClusterNode} can be obtained from + * {@link #clusterGetNodes()} or it can be constructed using either {@link RedisClusterNode#getHost() host} and + * {@link RedisClusterNode#getPort()} or the {@link RedisClusterNode#getId() node Id}. * * @author Christoph Strobl + * @author Mark Paluch * @since 1.7 */ public interface RedisClusterCommands { @@ -34,12 +37,12 @@ public interface RedisClusterCommands { * * @return never {@literal null}. */ - Iterable clusterGetClusterNodes(); + Iterable clusterGetNodes(); /** * Retrieve information about connected slaves for given master node. * - * @param node must not be {@literal null}. + * @param master must not be {@literal null}. * @return never {@literal null}. */ Collection clusterGetSlaves(RedisClusterNode master); @@ -131,8 +134,9 @@ public interface RedisClusterCommands { /** * Add given {@literal node} to cluster. - * - * @param node must not be {@literal null}. + * + * @param node must contain {@link RedisClusterNode#getHost() host} and {@link RedisClusterNode#getPort()} and must + * not be {@literal null}. */ void clusterMeet(RedisClusterNode node); diff --git a/src/main/java/org/springframework/data/redis/connection/RedisClusterConnection.java b/src/main/java/org/springframework/data/redis/connection/RedisClusterConnection.java index 274cc7b1c..894f77823 100644 --- a/src/main/java/org/springframework/data/redis/connection/RedisClusterConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/RedisClusterConnection.java @@ -22,9 +22,13 @@ import java.util.Set; import org.springframework.data.redis.core.types.RedisClientInfo; /** - * {@link RedisClusterConnection} allows sending commands to dedicated nodes within the cluster. + * {@link RedisClusterConnection} allows sending commands to dedicated nodes within the cluster. A + * {@link RedisClusterNode} can be obtained from {@link #clusterGetNodes()} or it can be constructed using either + * {@link RedisClusterNode#getHost() host} and {@link RedisClusterNode#getPort()} or the {@link RedisClusterNode#getId() + * node Id}. * * @author Christoph Strobl + * @author Mark Paluch * @since 1.7 */ public interface RedisClusterConnection extends RedisConnection, RedisClusterCommands { diff --git a/src/main/java/org/springframework/data/redis/connection/RedisClusterNode.java b/src/main/java/org/springframework/data/redis/connection/RedisClusterNode.java index 417d8b0d7..88c1aedbe 100644 --- a/src/main/java/org/springframework/data/redis/connection/RedisClusterNode.java +++ b/src/main/java/org/springframework/data/redis/connection/RedisClusterNode.java @@ -27,6 +27,7 @@ import org.springframework.util.CollectionUtils; * Representation of a Redis server within the cluster. * * @author Christoph Strobl + * @author Mark Paluch * @since 1.7 */ public class RedisClusterNode extends RedisNode { @@ -49,6 +50,18 @@ public class RedisClusterNode extends RedisNode { this(host, port, new SlotRange(Collections. emptySet())); } + /** + * Creates new {@link RedisClusterNode} with an id and empty {@link SlotRange}. + * + * @param id must not be {@literal null}. + */ + public RedisClusterNode(String id) { + + this(new SlotRange(Collections. emptySet())); + Assert.notNull(id, "Id must not be null"); + this.id = id; + } + /** * Creates new {@link RedisClusterNode} with given {@link SlotRange}. * @@ -62,6 +75,17 @@ public class RedisClusterNode extends RedisNode { this.slotRange = slotRange != null ? slotRange : new SlotRange(Collections. emptySet()); } + /** + * Creates new {@link RedisClusterNode} with given {@link SlotRange}. + * + * @param slotRange can be {@literal null}. + */ + public RedisClusterNode(SlotRange slotRange) { + + super(); + this.slotRange = slotRange != null ? slotRange : new SlotRange(Collections. emptySet()); + } + /** * Get the served {@link SlotRange}. * @@ -328,7 +352,12 @@ public class RedisClusterNode extends RedisNode { RedisNode base = super.build(); - RedisClusterNode node = new RedisClusterNode(base.getHost(), base.getPort(), slotRange); + RedisClusterNode node; + if (base.host != null) { + node = new RedisClusterNode(base.getHost(), base.getPort(), slotRange); + } else { + node = new RedisClusterNode(slotRange); + } node.id = base.id; node.type = base.type; node.masterId = base.masterId; diff --git a/src/main/java/org/springframework/data/redis/connection/RedisNode.java b/src/main/java/org/springframework/data/redis/connection/RedisNode.java index e77e2da8b..b4f47b941 100644 --- a/src/main/java/org/springframework/data/redis/connection/RedisNode.java +++ b/src/main/java/org/springframework/data/redis/connection/RedisNode.java @@ -21,6 +21,7 @@ import org.springframework.util.ObjectUtils; /** * @author Christoph Strobl * @author Thomas Darimont + * @author Mark Paluch * @since 1.4 */ public class RedisNode implements NamedNode { @@ -85,6 +86,15 @@ public class RedisNode implements NamedNode { return id; } + /** + * + * @param id + * @since 1.7 + */ + public void setId(String id) { + this.id = id; + } + /** * @return * @since 1.7 diff --git a/src/main/java/org/springframework/data/redis/connection/convert/Converters.java b/src/main/java/org/springframework/data/redis/connection/convert/Converters.java index edf6c439f..41b3063c9 100644 --- a/src/main/java/org/springframework/data/redis/connection/convert/Converters.java +++ b/src/main/java/org/springframework/data/redis/connection/convert/Converters.java @@ -16,6 +16,7 @@ package org.springframework.data.redis.connection.convert; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collection; import java.util.Collections; import java.util.LinkedHashMap; @@ -34,6 +35,7 @@ import org.springframework.data.redis.connection.RedisClusterNode.RedisClusterNo import org.springframework.data.redis.connection.RedisClusterNode.SlotRange; import org.springframework.data.redis.connection.RedisNode.NodeType; import org.springframework.data.redis.connection.RedisZSetCommands.Tuple; +import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; import org.springframework.util.NumberUtils; import org.springframework.util.StringUtils; @@ -43,11 +45,13 @@ import org.springframework.util.StringUtils; * * @author Jennifer Hickey * @author Thomas Darimont + * @author Mark Paluch */ abstract public class Converters { private static final byte[] ONE = new byte[] { '1' }; private static final byte[] ZERO = new byte[] { '0' }; + private static final String CLUSTER_NODES_LINE_SEPARATOR = "\n"; private static final Converter STRING_TO_PROPS = new StringToPropertiesConverter(); private static final Converter LONG_TO_BOOLEAN = new LongToBooleanConverter(); private static final Converter STRING_TO_DATA_TYPE = new StringToDataTypeConverter(); @@ -120,20 +124,32 @@ abstract public class Converters { private SlotRange parseSlotRange(String[] args) { - SlotRange range = new SlotRange(Collections. emptySet()); - if (args.length > SLOTS_INDEX && !args[SLOTS_INDEX].startsWith("[")) { + Set slots = new LinkedHashSet(); + + for (int i = SLOTS_INDEX; i < args.length; i++) { + + String raw = args[i]; + + if (raw.startsWith("[")) { + continue; + } - String raw = args[SLOTS_INDEX]; if (raw.contains("-")) { String[] slotRange = StringUtils.split(raw, "-"); if (slotRange != null) { - range = new RedisClusterNode.SlotRange(Integer.valueOf(slotRange[0]), Integer.valueOf(slotRange[1])); + int from = Integer.valueOf(slotRange[0]); + int to = Integer.valueOf(slotRange[1]); + for (int slot = from; slot <= to; slot++) { + slots.add(slot); + } } } else { - range = new SlotRange(Integer.valueOf(raw), Integer.valueOf(raw)); + slots.add(Integer.valueOf(raw)); } } + + SlotRange range = new SlotRange(slots); return range; } @@ -184,7 +200,7 @@ abstract public class Converters { } /** - * Converts the result of {@code CLUSTER NODES} into {@link RedisClusterNode}s. + * Converts lines from the result of {@code CLUSTER NODES} into {@link RedisClusterNode}s. * * @param clusterNodes * @return @@ -205,6 +221,22 @@ abstract public class Converters { return nodes; } + /** + * Converts the result of {@code CLUSTER NODES} into {@link RedisClusterNode}s. + * @param clusterNodes + * @return + * @since 1.7 + */ + public static Set toSetOfRedisClusterNodes(String clusterNodes) { + + if (StringUtils.isEmpty(clusterNodes)) { + return Collections.emptySet(); + } + + String[] lines = clusterNodes.split(CLUSTER_NODES_LINE_SEPARATOR); + return toSetOfRedisClusterNodes(Arrays.asList(lines)); + } + public static List toObjects(Set tuples) { List tupleArgs = new ArrayList(tuples.size() * 2); for (Tuple tuple : tuples) { @@ -225,4 +257,21 @@ abstract public class Converters { return NumberUtils.parseNumber(seconds, Long.class) * 1000L + NumberUtils.parseNumber(microseconds, Long.class) / 1000L; } + + /** + * Merge multiple {@code byte} arrays into one array + * @param firstArray must not be {@literal null} + * @param additionalArrays must not be {@literal null} + * @return + */ + public static byte[][] mergeArrays(byte[] firstArray, byte[]... additionalArrays){ + Assert.notNull(firstArray, "first array must not be null"); + Assert.notNull(additionalArrays, "additional arrays must not be null"); + + byte[][] result = new byte[additionalArrays.length + 1][]; + result[0] = firstArray; + System.arraycopy(additionalArrays, 0, result, 1, additionalArrays.length); + + return result; + } } 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 e7ac8740f..b53c32cf3 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 @@ -32,7 +32,7 @@ import java.util.Set; import org.springframework.beans.DirectFieldAccessor; import org.springframework.dao.DataAccessException; import org.springframework.dao.InvalidDataAccessApiUsageException; -import org.springframework.data.redis.ClusterStateFailureExeption; +import org.springframework.data.redis.ClusterStateFailureException; import org.springframework.data.redis.ExceptionTranslationStrategy; import org.springframework.data.redis.PassThroughExceptionTranslationStrategy; import org.springframework.data.redis.connection.ClusterCommandExecutor; @@ -78,6 +78,7 @@ import redis.clients.jedis.ZParams; * {@link Jedis} where needed. * * @author Christoph Strobl + * @author Mark Paluch * @since 1.7 */ public class JedisClusterConnection implements RedisClusterConnection { @@ -849,9 +850,7 @@ public class JedisClusterConnection implements RedisClusterConnection { @Override public Long bitOp(BitOperation op, byte[] destination, byte[]... keys) { - byte[][] allKeys = new byte[keys.length + 1][]; - allKeys[0] = destination; - System.arraycopy(keys, 0, allKeys, 1, keys.length); + byte[][] allKeys = Converters.mergeArrays(destination, keys); if (ClusterSlotHashUtil.isSameSlotForAllKeys(allKeys)) { try { @@ -1306,9 +1305,7 @@ public class JedisClusterConnection implements RedisClusterConnection { @Override public Long sInterStore(byte[] destKey, byte[]... keys) { - byte[][] allKeys = new byte[keys.length + 1][]; - allKeys[0] = destKey; - System.arraycopy(keys, 0, allKeys, 1, keys.length); + byte[][] allKeys = Converters.mergeArrays(destKey, keys); if (ClusterSlotHashUtil.isSameSlotForAllKeys(allKeys)) { try { @@ -1368,11 +1365,9 @@ public class JedisClusterConnection implements RedisClusterConnection { @Override public Long sUnionStore(byte[] destKey, byte[]... keys) { - byte[][] allKeys = new byte[keys.length + 1][]; - allKeys[0] = destKey; - System.arraycopy(keys, 0, allKeys, 1, keys.length); + byte[][] allKeys = Converters.mergeArrays(destKey, keys); - if (ClusterSlotHashUtil.isSameSlotForAllKeys(keys)) { + if (ClusterSlotHashUtil.isSameSlotForAllKeys(allKeys)) { try { return cluster.sunionstore(destKey, keys); } catch (Exception ex) { @@ -1433,9 +1428,7 @@ public class JedisClusterConnection implements RedisClusterConnection { @Override public Long sDiffStore(byte[] destKey, byte[]... keys) { - byte[][] allKeys = new byte[keys.length + 1][]; - allKeys[0] = destKey; - System.arraycopy(keys, 0, allKeys, 1, keys.length); + byte[][] allKeys = Converters.mergeArrays(destKey, keys); if (ClusterSlotHashUtil.isSameSlotForAllKeys(allKeys)) { try { @@ -1489,7 +1482,7 @@ public class JedisClusterConnection implements RedisClusterConnection { public List sRandMember(byte[] key, long count) { if (count > Integer.MAX_VALUE) { - throw new IllegalArgumentException("Count have cannot exceed Integer.MAX_VALUE!"); + throw new IllegalArgumentException("Count cannot exceed Integer.MAX_VALUE!"); } try { @@ -2063,9 +2056,7 @@ public class JedisClusterConnection implements RedisClusterConnection { @Override public Long zUnionStore(byte[] destKey, byte[]... sets) { - byte[][] allKeys = new byte[sets.length + 1][]; - allKeys[0] = destKey; - System.arraycopy(sets, 0, allKeys, 1, sets.length); + byte[][] allKeys = Converters.mergeArrays(destKey, sets); if (ClusterSlotHashUtil.isSameSlotForAllKeys(allKeys)) { @@ -2086,9 +2077,7 @@ public class JedisClusterConnection implements RedisClusterConnection { @Override public Long zUnionStore(byte[] destKey, Aggregate aggregate, int[] weights, byte[]... sets) { - byte[][] allKeys = new byte[sets.length + 1][]; - allKeys[0] = destKey; - System.arraycopy(sets, 0, allKeys, 1, sets.length); + byte[][] allKeys = Converters.mergeArrays(destKey, sets); if (ClusterSlotHashUtil.isSameSlotForAllKeys(allKeys)) { @@ -2107,9 +2096,7 @@ public class JedisClusterConnection implements RedisClusterConnection { @Override public Long zInterStore(byte[] destKey, byte[]... sets) { - byte[][] allKeys = new byte[sets.length + 1][]; - allKeys[0] = destKey; - System.arraycopy(sets, 0, allKeys, 1, sets.length); + byte[][] allKeys = Converters.mergeArrays(destKey, sets); if (ClusterSlotHashUtil.isSameSlotForAllKeys(allKeys)) { @@ -2126,9 +2113,7 @@ public class JedisClusterConnection implements RedisClusterConnection { @Override public Long zInterStore(byte[] destKey, Aggregate aggregate, int[] weights, byte[]... sets) { - byte[][] allKeys = new byte[sets.length + 1][]; - allKeys[0] = destKey; - System.arraycopy(sets, 0, allKeys, 1, sets.length); + byte[][] allKeys = Converters.mergeArrays(destKey, sets); if (ClusterSlotHashUtil.isSameSlotForAllKeys(allKeys)) { @@ -2422,7 +2407,6 @@ public class JedisClusterConnection implements RedisClusterConnection { @Override public void unwatch() { throw new InvalidDataAccessApiUsageException("UNWATCH is currently not supported in cluster mode."); - } /* @@ -2983,7 +2967,6 @@ public class JedisClusterConnection implements RedisClusterConnection { return client.configSet(param, value); } }); - } /* @@ -3072,7 +3055,7 @@ public class JedisClusterConnection implements RedisClusterConnection { Assert.notEmpty(serverTimeInformation, "Received invalid result from server. Expected 2 items in collection."); Assert.isTrue(serverTimeInformation.size() == 2, - "Received invalid nr of arguments from redis server. Expected 2 received " + serverTimeInformation.size()); + "Received invalid number of arguments from redis server. Expected 2 received " + serverTimeInformation.size()); return Converters.toTimeMillis(serverTimeInformation.get(0), serverTimeInformation.get(1)); } @@ -3160,7 +3143,7 @@ public class JedisClusterConnection implements RedisClusterConnection { @Override public void slaveOf(String host, int port) { throw new InvalidDataAccessApiUsageException( - "Slaveof is not supported in cluster environment. Please use CLUSTER REPLICATE."); + "SlaveOf is not supported in cluster environment. Please use CLUSTER REPLICATE."); } /* @@ -3170,7 +3153,7 @@ public class JedisClusterConnection implements RedisClusterConnection { @Override public void slaveOfNoOne() { throw new InvalidDataAccessApiUsageException( - "Slaveof is not supported in cluster environment. Please use CLUSTER REPLICATE."); + "SlaveOf is not supported in cluster environment. Please use CLUSTER REPLICATE."); } /* @@ -3276,9 +3259,7 @@ public class JedisClusterConnection implements RedisClusterConnection { @Override public void pfMerge(byte[] destinationKey, byte[]... sourceKeys) { - byte[][] allKeys = new byte[sourceKeys.length + 1][]; - allKeys[0] = destinationKey; - System.arraycopy(sourceKeys, 0, allKeys, 1, sourceKeys.length); + byte[][] allKeys = Converters.mergeArrays(destinationKey, sourceKeys); if (ClusterSlotHashUtil.isSameSlotForAllKeys(allKeys)) { try { @@ -3320,8 +3301,8 @@ public class JedisClusterConnection implements RedisClusterConnection { Assert.notNull(node, "Node must not be null."); Assert.notNull(mode, "AddSlots mode must not be null."); - final String nodeId = StringUtils.hasText(node.getId()) ? node.getId() : topologyProvider.getTopology() - .lookup(node.getHost(), node.getPort()).getId(); + final RedisClusterNode nodeToUse = topologyProvider.getTopology().lookup(node); + final String nodeId = nodeToUse.getId(); clusterCommandExecutor.executeCommandOnSingleNode(new JedisClusterCommandCallback() { @@ -3452,7 +3433,8 @@ public class JedisClusterConnection implements RedisClusterConnection { public void clusterForget(final RedisClusterNode node) { Set nodes = new LinkedHashSet(topologyProvider.getTopology().getActiveMasterNodes()); - nodes.remove(node); + final RedisClusterNode nodeToRemove = topologyProvider.getTopology().lookup(node); + nodes.remove(nodeToRemove); clusterCommandExecutor.executeCommandAsyncOnNodes(new JedisClusterCommandCallback() { @@ -3470,7 +3452,9 @@ public class JedisClusterConnection implements RedisClusterConnection { @Override public void clusterMeet(final RedisClusterNode node) { - Assert.notNull(node, "Node to meet cluster must not be null!"); + Assert.notNull(node, "Cluster node must not be null for CLUSTER MEET command!"); + Assert.hasText(node.getHost(), "Node to meet cluster must have a host!"); + Assert.isTrue(node.getPort() > 0, "Node to meet cluster must have a port greater 0!"); clusterCommandExecutor.executeCommandOnAllNodes(new JedisClusterCommandCallback() { @@ -3489,12 +3473,14 @@ public class JedisClusterConnection implements RedisClusterConnection { @Override public void clusterReplicate(final RedisClusterNode master, RedisClusterNode slave) { + final RedisClusterNode masterNode = topologyProvider.getTopology().lookup(master); + clusterCommandExecutor.executeCommandOnSingleNode(new JedisClusterCommandCallback() { @Override public String doInCluster(Jedis client) { - return client.clusterReplicate(master.getId()); + return client.clusterReplicate(masterNode.getId()); } }, slave); @@ -3534,10 +3520,10 @@ public class JedisClusterConnection implements RedisClusterConnection { /* * (non-Javadoc) - * @see org.springframework.data.redis.connection.RedisClusterCommands#clusterGetClusterNodes() + * @see org.springframework.data.redis.connection.RedisClusterCommands#clusterGetNodes() */ @Override - public Set clusterGetClusterNodes() { + public Set clusterGetNodes() { return topologyProvider.getTopology().getNodes(); } @@ -3550,8 +3536,8 @@ public class JedisClusterConnection implements RedisClusterConnection { Assert.notNull(master, "Master cannot be null!"); - final RedisClusterNode nodeToUse = StringUtils.hasText(master.getId()) ? master : topologyProvider.getTopology() - .lookup(master.getHost(), master.getPort()); + final RedisClusterNode nodeToUse = topologyProvider.getTopology() + .lookup(master); return JedisConverters.toSetOfRedisClusterNodes(clusterCommandExecutor.executeCommandOnSingleNode( new JedisClusterCommandCallback>() { @@ -3837,7 +3823,7 @@ public class JedisClusterConnection implements RedisClusterConnection { jedis = entry.getValue().getResource(); time = System.currentTimeMillis(); - Set nodes = JedisConverters.toSetOfRedisClusterNodes(jedis.clusterNodes()); + Set nodes = Converters.toSetOfRedisClusterNodes(jedis.clusterNodes()); synchronized (lock) { cached = new ClusterTopology(nodes); @@ -3856,7 +3842,7 @@ public class JedisClusterConnection implements RedisClusterConnection { for (Entry entry : errors.entrySet()) { sb.append(String.format("\r\n\t- %s failed: %s", entry.getKey(), entry.getValue().getMessage())); } - throw new ClusterStateFailureExeption( + throw new ClusterStateFailureException( "Could not retrieve cluster information. CLUSTER NODES returned with error." + sb.toString()); } } diff --git a/src/main/java/org/springframework/data/redis/connection/jedis/JedisConverters.java b/src/main/java/org/springframework/data/redis/connection/jedis/JedisConverters.java index 4efd694f0..e4a6c4101 100644 --- a/src/main/java/org/springframework/data/redis/connection/jedis/JedisConverters.java +++ b/src/main/java/org/springframework/data/redis/connection/jedis/JedisConverters.java @@ -57,6 +57,7 @@ import redis.clients.util.SafeEncoder; * @author Christoph Strobl * @author Thomas Darimont * @author Jungtaek Lim + * @author Mark Paluch */ abstract public class JedisConverters extends Converters { @@ -225,21 +226,6 @@ abstract public class JedisConverters extends Converters { return sentinels; } - /** - * @param clusterNodes - * @return - * @since 1.7 - */ - public static Set toSetOfRedisClusterNodes(String clusterNodes) { - - if (StringUtils.isEmpty(clusterNodes)) { - return Collections.emptySet(); - } - - String[] lines = clusterNodes.split(System.getProperty("line.separator")); - return toSetOfRedisClusterNodes(Arrays.asList(lines)); - } - public static DataAccessException toDataAccessException(Exception ex) { return EXCEPTION_CONVERTER.convert(ex); } 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 acd47943e..8dc9ec468 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 @@ -45,7 +45,6 @@ import org.springframework.data.redis.connection.RedisClusterNode; import org.springframework.data.redis.connection.RedisClusterNode.SlotRange; import org.springframework.data.redis.connection.SortParameters; import org.springframework.data.redis.connection.convert.Converters; -import org.springframework.data.redis.connection.jedis.JedisConverters; import org.springframework.data.redis.connection.util.ByteArraySet; import org.springframework.data.redis.core.Cursor; import org.springframework.data.redis.core.ScanOptions; @@ -53,7 +52,6 @@ import org.springframework.data.redis.core.types.RedisClientInfo; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; import org.springframework.util.ObjectUtils; -import org.springframework.util.StringUtils; import com.lambdaworks.redis.KeyValue; import com.lambdaworks.redis.RedisAsyncConnection; @@ -303,8 +301,8 @@ public class LettuceClusterConnection extends LettuceConnection implements Assert.notNull(master, "Master must not be null!"); - final RedisClusterNode nodeToUse = StringUtils.hasText(master.getId()) ? master : topologyProvider.getTopology() - .lookup(master.getHost(), master.getPort()); + final RedisClusterNode nodeToUse = topologyProvider.getTopology() + .lookup(master); return clusterCommandExecutor.executeCommandOnSingleNode( new LettuceClusterCommandCallback>() { @@ -426,14 +424,15 @@ public class LettuceClusterConnection extends LettuceConnection implements @Override public void clusterForget(final RedisClusterNode node) { - List nodes = new ArrayList(clusterGetClusterNodes()); - nodes.remove(node); + List nodes = new ArrayList(clusterGetNodes()); + final RedisClusterNode nodeToRemove = topologyProvider.getTopology().lookup(node); + nodes.remove(nodeToRemove); this.clusterCommandExecutor.executeCommandAsyncOnNodes(new LettuceClusterCommandCallback() { @Override public String doInCluster(RedisClusterConnection client) { - return client.clusterForget(node.getId()); + return client.clusterForget(nodeToRemove.getId()); } }, nodes); @@ -447,6 +446,8 @@ public class LettuceClusterConnection extends LettuceConnection implements public void clusterMeet(final RedisClusterNode node) { Assert.notNull(node, "Cluster node must not be null for CLUSTER MEET command!"); + Assert.hasText(node.getHost(), "Node to meet cluster must have a host!"); + Assert.isTrue(node.getPort() > 0, "Node to meet cluster must have a port greater 0!"); this.clusterCommandExecutor.executeCommandOnAllNodes(new LettuceClusterCommandCallback() { @@ -467,8 +468,8 @@ public class LettuceClusterConnection extends LettuceConnection implements Assert.notNull(node, "Node must not be null."); Assert.notNull(mode, "AddSlots mode must not be null."); - final String nodeId = StringUtils.hasText(node.getId()) ? node.getId() : topologyProvider.getTopology() - .lookup(node.getHost(), node.getPort()).getId(); + final RedisClusterNode nodeToUse = topologyProvider.getTopology().lookup(node); + final String nodeId = nodeToUse.getId(); clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback() { @@ -482,9 +483,9 @@ public class LettuceClusterConnection extends LettuceConnection implements case NODE: return client.clusterSetSlotNode(slot, nodeId); case STABLE: - throw new IllegalArgumentException("STABLE is not valid when using lettuce."); + return client.clusterSetSlotStable(slot); default: - throw new InvalidDataAccessApiUsageException("Invlid import mode for cluster slot: " + slot); + throw new InvalidDataAccessApiUsageException("Invalid import mode for cluster slot: " + slot); } } }, node); @@ -525,11 +526,12 @@ public class LettuceClusterConnection extends LettuceConnection implements @Override public void clusterReplicate(final RedisClusterNode master, RedisClusterNode slave) { + final RedisClusterNode masterNode = topologyProvider.getTopology().lookup(master); clusterCommandExecutor.executeCommandOnSingleNode(new LettuceClusterCommandCallback() { @Override public String doInCluster(RedisClusterConnection client) { - return client.clusterReplicate(master.getId()); + return client.clusterReplicate(masterNode.getId()); } }, slave); } @@ -762,7 +764,7 @@ public class LettuceClusterConnection extends LettuceConnection implements @Override public byte[] randomKey() { - List nodes = clusterGetClusterNodes(); + List nodes = clusterGetNodes(); Set inspectedNodes = new HashSet(nodes.size()); do { @@ -1092,9 +1094,8 @@ public class LettuceClusterConnection extends LettuceConnection implements @Override public Long sInterStore(byte[] destKey, byte[]... keys) { - byte[][] allKeys = new byte[keys.length + 1][]; - allKeys[0] = destKey; - System.arraycopy(keys, 0, allKeys, 1, keys.length); + byte[][] allKeys = Converters.mergeArrays(destKey, keys); + if (ClusterSlotHashUtil.isSameSlotForAllKeys(allKeys)) { return super.sInterStore(destKey, keys); } @@ -1145,9 +1146,8 @@ public class LettuceClusterConnection extends LettuceConnection implements @Override public Long sUnionStore(byte[] destKey, byte[]... keys) { - byte[][] allKeys = new byte[keys.length + 1][]; - allKeys[0] = destKey; - System.arraycopy(keys, 0, allKeys, 1, keys.length); + byte[][] allKeys = Converters.mergeArrays(destKey, keys); + if (ClusterSlotHashUtil.isSameSlotForAllKeys(allKeys)) { return super.sUnionStore(destKey, keys); } @@ -1201,9 +1201,7 @@ public class LettuceClusterConnection extends LettuceConnection implements @Override public Long sDiffStore(byte[] destKey, byte[]... keys) { - byte[][] allKeys = new byte[keys.length + 1][]; - allKeys[0] = destKey; - System.arraycopy(keys, 0, allKeys, 1, keys.length); + byte[][] allKeys = Converters.mergeArrays(destKey, keys); if (ClusterSlotHashUtil.isSameSlotForAllKeys(allKeys)) { return super.sDiffStore(destKey, keys); @@ -1233,7 +1231,7 @@ public class LettuceClusterConnection extends LettuceConnection implements * @see org.springframework.data.redis.connection.RedisClusterCommands#getClusterNodes() */ @Override - public List clusterGetClusterNodes() { + public List clusterGetNodes() { return LettuceConverters.partitionsToClusterNodes(clusterClient.getPartitions()); } @@ -1263,9 +1261,7 @@ public class LettuceClusterConnection extends LettuceConnection implements @Override public void pfMerge(byte[] destinationKey, byte[]... sourceKeys) { - byte[][] allKeys = new byte[sourceKeys.length + 1][]; - allKeys[0] = destinationKey; - System.arraycopy(sourceKeys, 0, allKeys, 1, sourceKeys.length); + byte[][] allKeys = Converters.mergeArrays(destinationKey, sourceKeys); if (ClusterSlotHashUtil.isSameSlotForAllKeys(allKeys)) { try { @@ -1455,7 +1451,7 @@ public class LettuceClusterConnection extends LettuceConnection implements Assert.notEmpty(serverTimeInformation, "Received invalid result from server. Expected 2 items in collection."); Assert.isTrue(serverTimeInformation.size() == 2, - "Received invalid nr of arguments from redis server. Expected 2 received " + serverTimeInformation.size()); + "Received invalid number of arguments from redis server. Expected 2 received " + serverTimeInformation.size()); return Converters.toTimeMillis(LettuceConverters.toString(serverTimeInformation.get(0)), LettuceConverters.toString(serverTimeInformation.get(1))); @@ -1513,7 +1509,7 @@ public class LettuceClusterConnection extends LettuceConnection implements @Override public Set doInCluster(RedisClusterConnection client) { - return JedisConverters.toSetOfRedisClusterNodes(client.clusterSlaves((String) client.clusterMyId())); + return Converters.toSetOfRedisClusterNodes(client.clusterSlaves(client.clusterMyId())); } }, topologyProvider.getTopology().getActiveMasterNodes()); } @@ -1600,7 +1596,7 @@ public class LettuceClusterConnection extends LettuceConnection implements */ public LettuceClusterTopologyProvider(RedisClusterClient client) { - Assert.notNull(client, "RedisClusteClient must not be null."); + Assert.notNull(client, "RedisClusterClient must not be null."); this.client = client; } diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java index c2f6ab72f..2aa73e57b 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java @@ -511,7 +511,7 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea @Override public RedisSentinelConnection getSentinelConnection() { if (!(client instanceof RedisClient)) { - throw new InvalidDataAccessResourceUsageException("Unable to connect to senitels using " + client.getClass()); + throw new InvalidDataAccessResourceUsageException("Unable to connect to sentinels using " + client.getClass()); } return new LettuceSentinelConnection(((RedisClient) client).connectSentinelAsync()); } diff --git a/src/main/java/org/springframework/data/redis/core/ClusterOperations.java b/src/main/java/org/springframework/data/redis/core/ClusterOperations.java index 3337bfd3e..b016f19e3 100644 --- a/src/main/java/org/springframework/data/redis/core/ClusterOperations.java +++ b/src/main/java/org/springframework/data/redis/core/ClusterOperations.java @@ -18,14 +18,19 @@ package org.springframework.data.redis.core; import java.util.Collection; import java.util.Set; +import org.springframework.data.redis.connection.RedisClusterCommands; import org.springframework.data.redis.connection.RedisClusterNode; import org.springframework.data.redis.connection.RedisClusterNode.SlotRange; import org.springframework.data.redis.connection.RedisConnection; /** - * Redis operations for cluster specific operations. + * Redis operations for cluster specific operations. A {@link RedisClusterNode} can be obtained from + * {@link RedisClusterCommands#clusterGetNodes() a connection} or it can be + * constructed using either {@link RedisClusterNode#getHost() host} and {@link RedisClusterNode#getPort()} or the + * {@link RedisClusterNode#getId() node Id}. * * @author Christoph Strobl + * @author Mark Paluch * @since 1.7 */ public interface ClusterOperations { @@ -75,7 +80,7 @@ public interface ClusterOperations { void addSlots(RedisClusterNode node, SlotRange range); /** - * tart an {@literal Append Only File} rewrite process on given node. + * Start an {@literal Append Only File} rewrite process on given node. * * @param node must not be {@literal null}. * @see RedisConnection#bgReWriteAof() diff --git a/src/test/java/org/springframework/data/redis/RedisTestProfileValueSource.java b/src/test/java/org/springframework/data/redis/RedisTestProfileValueSource.java index 9168bee93..598707150 100644 --- a/src/test/java/org/springframework/data/redis/RedisTestProfileValueSource.java +++ b/src/test/java/org/springframework/data/redis/RedisTestProfileValueSource.java @@ -31,6 +31,7 @@ import redis.clients.jedis.Jedis; * @author Jennifer Hickey * @author Christoph Strobl * @author Thomas Darimont + * @author Mark Paluch */ public class RedisTestProfileValueSource implements ProfileValueSource { @@ -56,7 +57,6 @@ public class RedisTestProfileValueSource implements ProfileValueSource { try { jedis = new Jedis(SettingsUtils.getHost(), SettingsUtils.getPort(), 100); - // jedis.connect(); String info = jedis.info(); String versionString = (String) JedisConverters.stringToProps().convert(info).get("redis_version"); 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 95358dd88..ca025ab40 100644 --- a/src/test/java/org/springframework/data/redis/connection/ClusterCommandExecutorUnitTests.java +++ b/src/test/java/org/springframework/data/redis/connection/ClusterCommandExecutorUnitTests.java @@ -51,6 +51,7 @@ import org.springframework.scheduling.concurrent.ConcurrentTaskExecutor; /** * @author Christoph Strobl + * @author Mark Paluch */ @RunWith(MockitoJUnitRunner.class) public class ClusterCommandExecutorUnitTests { @@ -75,6 +76,9 @@ public class ClusterCommandExecutorUnitTests { .listeningAt(CLUSTER_NODE_3_HOST, CLUSTER_NODE_3_PORT).serving(new SlotRange(10923, 16383)) .withId("3b9b8192a874fa8f1f09dbc0ee20afab5738eee7").promotedAs(NodeType.MASTER).linkState(LinkState.CONNECTED) .build(); + static final RedisClusterNode CLUSTER_NODE_2_LOOKUP = RedisClusterNode.newRedisClusterNode() + .withId("0f2ee5df45d18c50aca07228cc18b1da96fd5e84") + .build(); static final RedisClusterNode UNKNOWN_CLUSTER_NODE = new RedisClusterNode("8.8.8.8", 7379, null); @@ -139,6 +143,28 @@ public class ClusterCommandExecutorUnitTests { verify(con2, times(1)).theWheelWeavesAsTheWheelWills(); } + /** + * @see DATAREDIS-315 + */ + @Test + public void executeCommandOnSingleNodeByHostAndPortShouldBeExecutedCorrectly() { + + executor.executeCommandOnSingleNode(COMMAND_CALLBACK, new RedisClusterNode(CLUSTER_NODE_2_HOST, CLUSTER_NODE_2_PORT)); + + verify(con2, times(1)).theWheelWeavesAsTheWheelWills(); + } + + /** + * @see DATAREDIS-315 + */ + @Test + public void executeCommandOnSingleNodeByNodeIdShouldBeExecutedCorrectly() { + + executor.executeCommandOnSingleNode(COMMAND_CALLBACK, new RedisClusterNode(CLUSTER_NODE_2.id)); + + verify(con2, times(1)).theWheelWeavesAsTheWheelWills(); + } + /** * @see DATAREDIS-315 */ @@ -184,7 +210,57 @@ public class ClusterCommandExecutorUnitTests { * @see DATAREDIS-315 */ @Test - public void executeCommandOnAllNodesShouldExecuteCommandOnEveryKnwonClusterNode() { + public void executeCommandAsyncOnNodesShouldExecuteCommandOnGivenNodesByHostAndPort() { + + ClusterCommandExecutor executor = new ClusterCommandExecutor(new MockClusterNodeProvider(), + 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))); + + verify(con1, times(1)).theWheelWeavesAsTheWheelWills(); + verify(con2, times(1)).theWheelWeavesAsTheWheelWills(); + verify(con3, never()).theWheelWeavesAsTheWheelWills(); + } + + /** + * @see DATAREDIS-315 + */ + @Test + public void executeCommandAsyncOnNodesShouldExecuteCommandOnGivenNodesByNodeId() { + + ClusterCommandExecutor executor = new ClusterCommandExecutor(new MockClusterNodeProvider(), + 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)); + + verify(con1, times(1)).theWheelWeavesAsTheWheelWills(); + verify(con2, times(1)).theWheelWeavesAsTheWheelWills(); + verify(con3, never()).theWheelWeavesAsTheWheelWills(); + } + + /** + * @see DATAREDIS-315 + */ + @Test(expected = IllegalArgumentException.class) + public void executeCommandAsyncOnNodesShouldFailOnGivenUnknownNodes() { + + ClusterCommandExecutor executor = new ClusterCommandExecutor(new MockClusterNodeProvider(), + new MockClusterResourceProvider(), new PassThroughExceptionTranslationStrategy(exceptionConverter), + new ConcurrentTaskExecutor(new SyncTaskExecutor())); + + executor.executeCommandAsyncOnNodes(COMMAND_CALLBACK, Arrays.asList( + new RedisClusterNode("unknown"), CLUSTER_NODE_2_LOOKUP)); + } + + /** + * @see DATAREDIS-315 + */ + @Test + public void executeCommandOnAllNodesShouldExecuteCommandOnEveryKnownClusterNode() { ClusterCommandExecutor executor = new ClusterCommandExecutor(new MockClusterNodeProvider(), new MockClusterResourceProvider(), new PassThroughExceptionTranslationStrategy(exceptionConverter), diff --git a/src/test/java/org/springframework/data/redis/connection/convert/ConvertersUnitTests.java b/src/test/java/org/springframework/data/redis/connection/convert/ConvertersUnitTests.java index 9bee1a6eb..219538f7e 100644 --- a/src/test/java/org/springframework/data/redis/connection/convert/ConvertersUnitTests.java +++ b/src/test/java/org/springframework/data/redis/connection/convert/ConvertersUnitTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * Copyright 2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -27,20 +27,20 @@ import org.springframework.data.redis.connection.RedisClusterNode; import org.springframework.data.redis.connection.RedisClusterNode.Flag; import org.springframework.data.redis.connection.RedisClusterNode.LinkState; import org.springframework.data.redis.connection.RedisNode.NodeType; -import org.springframework.data.redis.connection.jedis.JedisConverters; /** * @author Christoph Strobl + * @author Mark Paluch */ public class ConvertersUnitTests { private static final String CLUSTER_NODES_RESPONSE = "" // - + "ef570f86c7b1a953846668debc177a3a16733420 127.0.0.1:6379 myself,master - 0 0 1 connected 0-5460" - + System.getProperty("line.separator") - + "0f2ee5df45d18c50aca07228cc18b1da96fd5e84 127.0.0.1:6380 master - 0 1427718161587 2 connected 5461-10922" - + System.getProperty("line.separator") + + "ef570f86c7b1a953846668debc177a3a16733420 127.0.0.1:6379 myself,master - 0 0 1 connected 0-5460 5602" + + "\n" + + "0f2ee5df45d18c50aca07228cc18b1da96fd5e84 127.0.0.1:6380 master - 0 1427718161587 2 connected 5461 5603-10922" + + "\n" + "3b9b8192a874fa8f1f09dbc0ee20afab5738eee7 127.0.0.1:6381 master - 0 1427718161587 3 connected 10923-16383" - + System.getProperty("line.separator") + + "\n" + "8cad73f63eb996fedba89f041636f17d88cda075 127.0.0.1:7369 slave ef570f86c7b1a953846668debc177a3a16733420 0 1427718161587 1 connected"; private static final String CLUSTER_NODE_WITH_SINGLE_SLOT_RESPONSE = "ef570f86c7b1a953846668debc177a3a16733420 127.0.0.1:6379 myself,master - 0 0 1 connected 3456"; @@ -55,7 +55,7 @@ public class ConvertersUnitTests { @Test public void toSetOfRedisClusterNodesShouldConvertSingleStringNodesResponseCorrectly() { - Iterator nodes = JedisConverters.toSetOfRedisClusterNodes(CLUSTER_NODES_RESPONSE).iterator(); + Iterator nodes = Converters.toSetOfRedisClusterNodes(CLUSTER_NODES_RESPONSE).iterator(); RedisClusterNode node = nodes.next(); // 127.0.0.1:6379 assertThat(node.getId(), is("ef570f86c7b1a953846668debc177a3a16733420")); @@ -64,6 +64,8 @@ public class ConvertersUnitTests { assertThat(node.getType(), is(NodeType.MASTER)); assertThat(node.getSlotRange().contains(0), is(true)); assertThat(node.getSlotRange().contains(5460), is(true)); + assertThat(node.getSlotRange().contains(5461), is(false)); + assertThat(node.getSlotRange().contains(5602), is(true)); assertThat(node.getFlags(), hasItems(Flag.MASTER, Flag.MYSELF)); assertThat(node.getLinkState(), is(LinkState.CONNECTED)); @@ -72,7 +74,9 @@ public class ConvertersUnitTests { assertThat(node.getHost(), is("127.0.0.1")); assertThat(node.getPort(), is(6380)); assertThat(node.getType(), is(NodeType.MASTER)); + assertThat(node.getSlotRange().contains(5460), is(false)); assertThat(node.getSlotRange().contains(5461), is(true)); + assertThat(node.getSlotRange().contains(5462), is(false)); assertThat(node.getSlotRange().contains(10922), is(true)); assertThat(node.getFlags(), hasItems(Flag.MASTER)); assertThat(node.getLinkState(), is(LinkState.CONNECTED)); @@ -102,9 +106,9 @@ public class ConvertersUnitTests { * @see DATAREDIS-315 */ @Test - public void toSetOfRedisClusterNodesShouldConvertNodesWihtSingleSlotCorrectly() { + public void toSetOfRedisClusterNodesShouldConvertNodesWithSingleSlotCorrectly() { - Iterator nodes = JedisConverters.toSetOfRedisClusterNodes(CLUSTER_NODE_WITH_SINGLE_SLOT_RESPONSE) + Iterator nodes = Converters.toSetOfRedisClusterNodes(CLUSTER_NODE_WITH_SINGLE_SLOT_RESPONSE) .iterator(); RedisClusterNode node = nodes.next(); // 127.0.0.1:6379 @@ -121,7 +125,7 @@ public class ConvertersUnitTests { @Test public void toSetOfRedisClusterNodesShouldParseLinkStateAndDisconnectedCorrectly() { - Iterator nodes = JedisConverters.toSetOfRedisClusterNodes( + Iterator nodes = Converters.toSetOfRedisClusterNodes( CLUSTER_NODE_WITH_FAIL_FLAG_AND_DISCONNECTED_LINK_STATE).iterator(); RedisClusterNode node = nodes.next(); @@ -140,7 +144,7 @@ public class ConvertersUnitTests { @Test public void toSetOfRedisClusterNodesShouldIgnoreImportingSlot() { - Iterator nodes = JedisConverters.toSetOfRedisClusterNodes(CLUSTER_NODE_IMPORTING_SLOT).iterator(); + Iterator nodes = Converters.toSetOfRedisClusterNodes(CLUSTER_NODE_IMPORTING_SLOT).iterator(); RedisClusterNode node = nodes.next(); assertThat(node.getId(), is("ef570f86c7b1a953846668debc177a3a16733420")); 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 6689e8dc7..cef0fe947 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 @@ -59,6 +59,7 @@ import redis.clients.jedis.exceptions.JedisDataException; /** * @author Christoph Strobl + * @author Mark Paluch */ public class JedisClusterConnectionTests implements ClusterConnectionTests { diff --git a/src/test/java/org/springframework/data/redis/connection/jedis/JedisClusterConnectionUnitTests.java b/src/test/java/org/springframework/data/redis/connection/jedis/JedisClusterConnectionUnitTests.java index 16941f849..f80cf578d 100644 --- a/src/test/java/org/springframework/data/redis/connection/jedis/JedisClusterConnectionUnitTests.java +++ b/src/test/java/org/springframework/data/redis/connection/jedis/JedisClusterConnectionUnitTests.java @@ -35,7 +35,7 @@ import org.junit.rules.ExpectedException; import org.junit.runner.RunWith; import org.mockito.Mock; import org.mockito.runners.MockitoJUnitRunner; -import org.springframework.data.redis.ClusterStateFailureExeption; +import org.springframework.data.redis.ClusterStateFailureException; import org.springframework.data.redis.connection.ClusterInfo; import org.springframework.data.redis.connection.RedisClusterCommands.AddSlots; import org.springframework.data.redis.connection.RedisClusterNode; @@ -48,6 +48,7 @@ import redis.clients.jedis.exceptions.JedisConnectionException; /** * @author Christoph Strobl + * @author Mark Paluch */ @RunWith(MockitoJUnitRunner.class) public class JedisClusterConnectionUnitTests { @@ -55,20 +56,20 @@ public class JedisClusterConnectionUnitTests { private static final String CLUSTER_NODES_RESPONSE = "" // + MASTER_NODE_1_ID + " " + CLUSTER_HOST + ":" + MASTER_NODE_1_PORT + " myself,master - 0 0 1 connected 0-5460" - + System.getProperty("line.separator") + MASTER_NODE_2_ID + " " + CLUSTER_HOST + ":" + + "\n" + MASTER_NODE_2_ID + " " + CLUSTER_HOST + ":" + MASTER_NODE_2_PORT - + " master - 0 1427718161587 2 connected 5461-10922" + System.getProperty("line.separator") + + " master - 0 1427718161587 2 connected 5461-10922" + "\n" + MASTER_NODE_2_ID + " " + CLUSTER_HOST + ":" + MASTER_NODE_3_PORT + " master - 0 1427718161587 3 connected 10923-16383"; - static final String CLUSTER_INFO_RESPONSE = "cluster_state:ok" + System.getProperty("line.separator") - + "cluster_slots_assigned:16384" + System.getProperty("line.separator") + "cluster_slots_ok:16384" - + System.getProperty("line.separator") + "cluster_slots_pfail:0" + System.getProperty("line.separator") - + "cluster_slots_fail:0" + System.getProperty("line.separator") + "cluster_known_nodes:4" - + System.getProperty("line.separator") + "cluster_size:3" + System.getProperty("line.separator") - + "cluster_current_epoch:30" + System.getProperty("line.separator") + "cluster_my_epoch:2" - + System.getProperty("line.separator") + "cluster_stats_messages_sent:2560260" - + System.getProperty("line.separator") + "cluster_stats_messages_received:2560086"; + static final String CLUSTER_INFO_RESPONSE = "cluster_state:ok" + "\n" + + "cluster_slots_assigned:16384" + "\n" + "cluster_slots_ok:16384" + + "\n" + "cluster_slots_pfail:0" + "\n" + + "cluster_slots_fail:0" + "\n" + "cluster_known_nodes:4" + + "\n" + "cluster_size:3" + "\n" + + "cluster_current_epoch:30" + "\n" + "cluster_my_epoch:2" + + "\n" + "cluster_stats_messages_sent:2560260" + + "\n" + "cluster_stats_messages_received:2560086"; JedisClusterConnection connection; @@ -159,6 +160,7 @@ public class JedisClusterConnectionUnitTests { connection.clusterReplicate(CLUSTER_NODE_1, CLUSTER_NODE_2); verify(con2Mock, times(1)).clusterReplicate(CLUSTER_NODE_1.getId()); + verify(con1Mock, times(1)).clusterNodes(); verifyZeroInteractions(con1Mock); } @@ -312,6 +314,7 @@ public class JedisClusterConnectionUnitTests { connection.time(CLUSTER_NODE_2); verify(con2Mock, times(1)).time(); + verify(con1Mock, times(1)).clusterNodes(); verifyZeroInteractions(con1Mock, con3Mock); } @@ -347,7 +350,7 @@ public class JedisClusterConnectionUnitTests { @Test public void clusterTopologyProviderShouldCollectErrorsWhenLoadingNodes() { - expectedException.expect(ClusterStateFailureExeption.class); + expectedException.expect(ClusterStateFailureException.class); expectedException.expectMessage("127.0.0.1:7379 failed: o.O"); expectedException.expectMessage("127.0.0.1:7380 failed: o.1"); expectedException.expectMessage("127.0.0.1:7381 failed: o.2"); diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/DefaultLettucePoolTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/DefaultLettucePoolTests.java index 1cb710f8e..ca8b5ae3a 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/DefaultLettucePoolTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/DefaultLettucePoolTests.java @@ -17,6 +17,8 @@ package org.springframework.data.redis.connection.lettuce; import static org.junit.Assert.*; +import java.util.concurrent.TimeUnit; + import org.apache.commons.pool2.impl.GenericObjectPoolConfig; import org.junit.After; import org.junit.Ignore; @@ -34,6 +36,7 @@ import com.lambdaworks.redis.RedisException; * @author Jennifer Hickey * @author Thomas Darimont * @author Christoph Strobl + * @author Mark Paluch */ public class DefaultLettucePoolTests { @@ -44,7 +47,7 @@ public class DefaultLettucePoolTests { if (this.pool != null) { if (this.pool.getClient() != null) { - this.pool.getClient().shutdown(); + this.pool.getClient().shutdown(0, 0, TimeUnit.MILLISECONDS); } this.pool.destroy(); 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 8157c6773..9e7071649 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 @@ -33,7 +33,6 @@ import java.util.concurrent.TimeUnit; import org.junit.After; import org.junit.Before; import org.junit.ClassRule; -import org.junit.Ignore; import org.junit.Test; import org.springframework.dao.DataAccessException; import org.springframework.data.redis.connection.ClusterConnectionTests; @@ -58,6 +57,7 @@ import com.lambdaworks.redis.cluster.RedisClusterClient; /** * @author Christoph Strobl + * @author Mark Paluch */ public class LettuceClusterConnectionTests implements ClusterConnectionTests { @@ -2129,7 +2129,6 @@ public class LettuceClusterConnectionTests implements ClusterConnectionTests { * @see DATAREDIS-315 */ @Test - @Ignore("Should work in 3.4 but does not work in 3.3.2") public void countKeysShouldReturnNumberOfKeysInSlot() { nativeConnection.set(SAME_SLOT_KEY_1, VALUE_1); diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionUnitTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionUnitTests.java index 1059059a8..d9391d57c 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionUnitTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionUnitTests.java @@ -49,6 +49,7 @@ import com.lambdaworks.redis.cluster.models.partitions.RedisClusterNode.NodeFlag /** * @author Christoph Strobl + * @author Mark Paluch */ @RunWith(MockitoJUnitRunner.class) public class LettuceClusterConnectionUnitTests { @@ -116,7 +117,7 @@ public class LettuceClusterConnectionUnitTests { } @Override - public List clusterGetClusterNodes() { + public List clusterGetNodes() { return Arrays.asList(CLUSTER_NODE_1, CLUSTER_NODE_2, CLUSTER_NODE_3); } }; @@ -306,12 +307,11 @@ public class LettuceClusterConnectionUnitTests { * @see DATAREDIS-315 */ @Test - @Ignore("Stable not available for lettuce") public void clusterSetSlotStableShouldBeExecutedCorrectly() { connection.clusterSetSlot(CLUSTER_NODE_1, 100, AddSlots.STABLE); - // verify(clusterConnection1Mock, times(1)).clusterSetSlotStable(eq(100)); + verify(clusterConnection1Mock, times(1)).clusterSetSlotStable(eq(100)); } /** diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionTransactionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionTransactionIntegrationTests.java index 38fcc60fc..e6535e00c 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionTransactionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionTransactionIntegrationTests.java @@ -46,6 +46,7 @@ public class LettuceConnectionTransactionIntegrationTests extends AbstractConnec verifyResults(Arrays.asList(new Object[] { true })); // Lettuce does not support select when using shared conn, use a new conn factory LettuceConnectionFactory factory2 = new LettuceConnectionFactory(); + factory2.setShutdownTimeout(0); factory2.setDatabase(1); factory2.afterPropertiesSet(); StringRedisConnection conn2 = new DefaultStringRedisConnection(factory2.getConnection()); diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelIntegrationTests.java index 67cb38835..6cac5d13c 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelIntegrationTests.java @@ -54,6 +54,7 @@ public class LettuceSentinelIntegrationTests extends AbstractConnectionIntegrati LettuceConnectionFactory lettuceConnectionFactory = new LettuceConnectionFactory(SENTINEL_CONFIG); lettuceConnectionFactory.setShareNativeConnection(false); + lettuceConnectionFactory.setShutdownTimeout(0); lettuceConnectionFactory.afterPropertiesSet(); connectionFactory = lettuceConnectionFactory; diff --git a/src/test/java/org/springframework/data/redis/core/script/lettuce/LettuceDefaultScriptExecutorTests.java b/src/test/java/org/springframework/data/redis/core/script/lettuce/LettuceDefaultScriptExecutorTests.java index 37670a87b..1485090cd 100644 --- a/src/test/java/org/springframework/data/redis/core/script/lettuce/LettuceDefaultScriptExecutorTests.java +++ b/src/test/java/org/springframework/data/redis/core/script/lettuce/LettuceDefaultScriptExecutorTests.java @@ -27,6 +27,7 @@ import org.springframework.data.redis.core.script.DefaultScriptExecutor; * Integration test of {@link DefaultScriptExecutor} with Lettuce. * * @author Thomas Darimont + * @author Mark Paluch */ public class LettuceDefaultScriptExecutorTests extends AbstractDefaultScriptExecutorTests { @@ -36,6 +37,7 @@ public class LettuceDefaultScriptExecutorTests extends AbstractDefaultScriptExec public void setup() { connectionFactory = new LettuceConnectionFactory(SettingsUtils.getHost(), SettingsUtils.getPort()); + connectionFactory.setShutdownTimeout(0); connectionFactory.afterPropertiesSet(); } diff --git a/src/test/resources/org/springframework/data/redis/connection/lettuce/LettuceConnectionIntegrationTests-context.xml b/src/test/resources/org/springframework/data/redis/connection/lettuce/LettuceConnectionIntegrationTests-context.xml index 75da6c57d..e38de3879 100644 --- a/src/test/resources/org/springframework/data/redis/connection/lettuce/LettuceConnectionIntegrationTests-context.xml +++ b/src/test/resources/org/springframework/data/redis/connection/lettuce/LettuceConnectionIntegrationTests-context.xml @@ -11,6 +11,7 @@ + \ No newline at end of file