DATAREDIS-315 - Polishing.

Update documentation and fix spelling in JavaDoc and reference documentation.

Allow node lookup by host/port and node Id to enable flexibility when passing references to Redis Cluster nodes. Parse multiple slot ranges for a cluster node and move CLUSTER NODES parser to Converters.

Use fixed line separator instead of OS-specific line separator, consolidate duplicate code, support clusterSetSlotStable with lettuce 3.4 and set lettuce driver shutdown timeout to zero for long running tests.

Original pull request: #158.

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

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 <S, T> java.util.Map<RedisClusterNode, T> executeCommandAsyncOnNodes(
 			final ClusterCommandCallback<S, T> callback, Iterable<RedisClusterNode> 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<RedisClusterNode, Future<T>> futures = new LinkedHashMap<RedisClusterNode, Future<T>>();
+
+		List<RedisClusterNode> resolvedRedisClusterNodes = new ArrayList<RedisClusterNode>();
+		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<RedisClusterNode, Future<T>> futures = new LinkedHashMap<RedisClusterNode, Future<T>>();
+		for (final RedisClusterNode node : resolvedRedisClusterNodes) {
 
 			futures.put(node, executor.submit(new Callable<T>() {
 

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}.

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<RedisClusterNode> clusterGetClusterNodes();
+	Iterable<RedisClusterNode> 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<RedisClusterNode> 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);
 

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 {

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.<Integer> 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.<Integer> 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.<Integer> 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.<Integer> 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;

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

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, Properties> STRING_TO_PROPS = new StringToPropertiesConverter();
 	private static final Converter<Long, Boolean> LONG_TO_BOOLEAN = new LongToBooleanConverter();
 	private static final Converter<String, DataType> STRING_TO_DATA_TYPE = new StringToDataTypeConverter();
@@ -120,20 +124,32 @@ abstract public class Converters {
 
 			private SlotRange parseSlotRange(String[] args) {
 
-				SlotRange range = new SlotRange(Collections.<Integer> emptySet());
-				if (args.length > SLOTS_INDEX && !args[SLOTS_INDEX].startsWith("[")) {
+				Set<Integer> slots = new LinkedHashSet<Integer>();
+
+				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<RedisClusterNode> 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<Object> toObjects(Set<Tuple> tuples) {
 		List<Object> tupleArgs = new ArrayList<Object>(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;
+	}
 }

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<byte[]> 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<String>() {
 
@@ -3452,7 +3433,8 @@ public class JedisClusterConnection implements RedisClusterConnection {
 	public void clusterForget(final RedisClusterNode node) {
 
 		Set<RedisClusterNode> nodes = new LinkedHashSet<RedisClusterNode>(topologyProvider.getTopology().getActiveMasterNodes());
-		nodes.remove(node);
+		final RedisClusterNode nodeToRemove = topologyProvider.getTopology().lookup(node);
+		nodes.remove(nodeToRemove);
 
 		clusterCommandExecutor.executeCommandAsyncOnNodes(new JedisClusterCommandCallback<String>() {
 
@@ -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<String>() {
 
@@ -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<String>() {
 
 			@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<RedisClusterNode> clusterGetClusterNodes() {
+	public Set<RedisClusterNode> 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<List<String>>() {
@@ -3837,7 +3823,7 @@ public class JedisClusterConnection implements RedisClusterConnection {
 					jedis = entry.getValue().getResource();
 
 					time = System.currentTimeMillis();
-					Set<RedisClusterNode> nodes = JedisConverters.toSetOfRedisClusterNodes(jedis.clusterNodes());
+					Set<RedisClusterNode> nodes = Converters.toSetOfRedisClusterNodes(jedis.clusterNodes());
 
 					synchronized (lock) {
 						cached = new ClusterTopology(nodes);
@@ -3856,7 +3842,7 @@ public class JedisClusterConnection implements RedisClusterConnection {
 			for (Entry<String, Exception> 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());
 		}
 	}

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<RedisClusterNode> 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);
 	}

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<Set<RedisClusterNode>>() {
@@ -426,14 +424,15 @@ public class LettuceClusterConnection extends LettuceConnection implements
 	@Override
 	public void clusterForget(final RedisClusterNode node) {
 
-		List<RedisClusterNode> nodes = new ArrayList<RedisClusterNode>(clusterGetClusterNodes());
-		nodes.remove(node);
+		List<RedisClusterNode> nodes = new ArrayList<RedisClusterNode>(clusterGetNodes());
+		final RedisClusterNode nodeToRemove = topologyProvider.getTopology().lookup(node);
+		nodes.remove(nodeToRemove);
 
 		this.clusterCommandExecutor.executeCommandAsyncOnNodes(new LettuceClusterCommandCallback<String>() {
 
 			@Override
 			public String doInCluster(RedisClusterConnection<byte[], byte[]> 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<String>() {
 
@@ -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<String>() {
 
@@ -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<String>() {
 
 			@Override
 			public String doInCluster(RedisClusterConnection<byte[], byte[]> 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<RedisClusterNode> nodes = clusterGetClusterNodes();
+		List<RedisClusterNode> nodes = clusterGetNodes();
 		Set<RedisClusterNode> inspectedNodes = new HashSet<RedisClusterNode>(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<RedisClusterNode> clusterGetClusterNodes() {
+	public List<RedisClusterNode> 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<RedisClusterNode> doInCluster(RedisClusterConnection<byte[], byte[]> 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;
 		}
 

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());
 	}

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<K, V> {
@@ -75,7 +80,7 @@ public interface ClusterOperations<K, V> {
 	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()