DATAREDIS-315 - Add initial support for redis cluster (jedis/lettuce).

Cluster support is based on the very same building blocks as non clustered communication. RedisClusterConnection and extension to RedisConnection handles the communication with the Redis Cluster and translates errors into the Spring DAO exception hierarchy.

Redirects for to a specific keys to the corresponding slot serving node are handled by the driver libraries, higher level functions like collecting information accross nodes, or sending commands to all nodes in the cluster that are covered by RedisClusterConnection utilizing a ClusterCommandExecutor distributing commands accross the cluster and collecting results.

RedisTemplate provides access to cluster specific operations via the ClusterOperations interface that can be obtained via RedisTemplate.opsForCluster(). This allows to execute commands explicitly on a single node within the cluster while retaining de-/serialization features configured for the template.

Original pull request: #158.

src/main/java/org/springframework/data/redis/ClusterRedirectException.java

@@ -0,0 +1,79 @@
+/*
+ * Copyright 2015 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis;
+
+import org.springframework.dao.DataRetrievalFailureException;
+
+/**
+ * {@link ClusterRedirectException} indicates that a requested slot is not served by the targeted server but can be
+ * obtained on another one.
+ * 
+ * @author Christoph Strobl
+ * @since 1.7
+ */
+public class ClusterRedirectException extends DataRetrievalFailureException {
+
+	private static final long serialVersionUID = -857075813794333965L;
+
+	private final int slot;
+	private final String host;
+	private final int port;
+
+	/**
+	 * Creates new {@link ClusterRedirectException}.
+	 * 
+	 * @param slot the slot to redirect to.
+	 * @param targetHost the host to redirect to.
+	 * @param targetPort the port on the host.
+	 * @param e the root cause from the data access API in use
+	 */
+	public ClusterRedirectException(int slot, String targetHost, int targetPort, Throwable e) {
+
+		super(String.format("Redirect: slot %s to %s:%s.", slot, targetHost, targetPort), e);
+
+		this.slot = slot;
+		this.host = targetHost;
+		this.port = targetPort;
+	}
+
+	/**
+	 * Get slot to go for.
+	 * 
+	 * @return
+	 */
+	public int getSlot() {
+		return slot;
+	}
+
+	/**
+	 * Get host serving the slot.
+	 * 
+	 * @return
+	 */
+	public String getTargetHost() {
+		return host;
+	}
+
+	/**
+	 * Get port on host serving the slot.
+	 * 
+	 * @return
+	 */
+	public int getTargetPort() {
+		return port;
+	}
+
+}

src/main/java/org/springframework/data/redis/ClusterStateFailureExeption.java

@@ -0,0 +1,51 @@
+/*
+ * Copyright 2015 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis;
+
+import org.springframework.dao.DataAccessResourceFailureException;
+
+/**
+ * {@link DataAccessResourceFailureException} indicating the current local snapshot of cluster state does no longer
+ * represent the actual remote state. This can happen nodes are removed from cluster, slots get migrated to other nodes
+ * and so on.
+ * 
+ * @author Christoph Strobl
+ * @since 1.7
+ */
+public class ClusterStateFailureExeption extends DataAccessResourceFailureException {
+
+	private static final long serialVersionUID = 333399051713240852L;
+
+	/**
+	 * Creates new {@link ClusterStateFailureExeption}.
+	 * 
+	 * @param msg
+	 */
+	public ClusterStateFailureExeption(String msg) {
+		super(msg);
+	}
+
+	/**
+	 * Creates new {@link ClusterStateFailureExeption}.
+	 * 
+	 * @param msg
+	 * @param cause
+	 */
+	public ClusterStateFailureExeption(String msg, Throwable cause) {
+		super(msg, cause);
+	}
+
+}

src/main/java/org/springframework/data/redis/TooManyClusterRedirectionsException.java

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2015 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis;
+
+import org.springframework.dao.DataRetrievalFailureException;
+
+/**
+ * {@link DataRetrievalFailureException} thrown when following cluster redirects exceeds the max number of edges.
+ * 
+ * @author Christoph Strobl
+ * @since 1.7
+ */
+public class TooManyClusterRedirectionsException extends DataRetrievalFailureException {
+
+	private static final long serialVersionUID = -2818933672669154328L;
+
+	/**
+	 * Creates new {@link TooManyClusterRedirectionsException}.
+	 * 
+	 * @param msg
+	 */
+	public TooManyClusterRedirectionsException(String msg) {
+		super(msg);
+	}
+
+	/**
+	 * Creates new {@link TooManyClusterRedirectionsException}.
+	 * 
+	 * @param msg
+	 * @param cause
+	 */
+	public TooManyClusterRedirectionsException(String msg, Throwable cause) {
+		super(msg, cause);
+	}
+
+}

src/main/java/org/springframework/data/redis/connection/ClusterCommandExecutionFailureException.java

@@ -0,0 +1,64 @@
+/*
+ * Copyright 2015 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.connection;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+
+import org.springframework.dao.UncategorizedDataAccessException;
+
+/**
+ * Exception thrown when at least one call to a clustered redis environment fails.
+ * 
+ * @author Christoph Strobl
+ * @since 1.7
+ */
+public class ClusterCommandExecutionFailureException extends UncategorizedDataAccessException {
+
+	private static final long serialVersionUID = 5727044227040368955L;
+
+	private final Collection<? extends Throwable> causes;
+
+	/**
+	 * Creates new {@link ClusterCommandExecutionFailureException}.
+	 * 
+	 * @param cause must not be {@literal null}.
+	 */
+	public ClusterCommandExecutionFailureException(Throwable cause) {
+		this(Collections.singletonList(cause));
+	}
+
+	/**
+	 * Creates new {@link ClusterCommandExecutionFailureException}.
+	 * 
+	 * @param causes must not be {@literal empty}.
+	 */
+	public ClusterCommandExecutionFailureException(List<? extends Throwable> causes) {
+
+		super(causes.get(0).getMessage(), causes.get(0));
+		this.causes = causes;
+	}
+
+	/**
+	 * Get the collected errors.
+	 * 
+	 * @return never {@literal null}.
+	 */
+	public Collection<? extends Throwable> getCauses() {
+		return causes;
+	}
+}

src/main/java/org/springframework/data/redis/connection/ClusterCommandExecutor.java

@@ -0,0 +1,352 @@
+/*
+ * Copyright 2015 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.connection;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+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.ExceptionTranslationStrategy;
+import org.springframework.data.redis.TooManyClusterRedirectionsException;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import org.springframework.util.Assert;
+
+/**
+ * {@link ClusterCommandExecutor} takes care of running commands across the known cluster nodes. By providing an
+ * {@link AsyncTaskExecutor} the execution behavior can be influenced.
+ * 
+ * @author Christoph Strobl
+ * @since 1.7
+ */
+public class ClusterCommandExecutor implements DisposableBean {
+
+	private AsyncTaskExecutor executor;
+	private final ClusterTopologyProvider topologyProvider;
+	private final ClusterNodeResourceProvider resourceProvider;
+	private final ExceptionTranslationStrategy exceptionTranslationStrategy;
+	private int maxRedirects = 5;
+
+	/**
+	 * Create a new instance of {@link ClusterCommandExecutor}.
+	 * 
+	 * @param topologyProvider must not be {@literal null}.
+	 * @param resourceProvider must not be {@literal null}.
+	 * @param exceptionTranslation must not be {@literal null}.
+	 */
+	public ClusterCommandExecutor(ClusterTopologyProvider topologyProvider, ClusterNodeResourceProvider resourceProvider,
+			ExceptionTranslationStrategy exceptionTranslation) {
+
+		Assert.notNull(topologyProvider);
+		Assert.notNull(resourceProvider);
+		Assert.notNull(exceptionTranslation);
+
+		this.topologyProvider = topologyProvider;
+		this.resourceProvider = resourceProvider;
+		this.exceptionTranslationStrategy = exceptionTranslation;
+	}
+
+	/**
+	 * @param topologyProvider must not be {@literal null}.
+	 * @param resourceProvider must not be {@literal null}.
+	 * @param exceptionTranslation must not be {@literal null}.
+	 * @param executor can be {@literal null}.
+	 */
+	public ClusterCommandExecutor(ClusterTopologyProvider topologyProvider, ClusterNodeResourceProvider resourceProvider,
+			ExceptionTranslationStrategy exceptionTranslation, AsyncTaskExecutor executor) {
+
+		this(topologyProvider, resourceProvider, exceptionTranslation);
+		this.executor = executor;
+	}
+
+	{
+		if (executor == null) {
+			ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
+			threadPoolTaskExecutor.initialize();
+			this.executor = threadPoolTaskExecutor;
+		}
+	}
+
+	/**
+	 * Run {@link ClusterCommandCallback} on a random node.
+	 * 
+	 * @param cmd must not be {@literal null}.
+	 * @return
+	 */
+	public <T> T executeCommandOnArbitraryNode(ClusterCommandCallback<?, T> cmd) {
+
+		Assert.notNull(cmd, "ClusterCommandCallback must not be null!");
+		List<RedisClusterNode> nodes = new ArrayList<RedisClusterNode>(getClusterTopology().getActiveNodes());
+		return executeCommandOnSingleNode(cmd, nodes.get(new Random().nextInt(nodes.size())));
+	}
+
+	/**
+	 * Run {@link ClusterCommandCallback} on given {@link RedisClusterNode}.
+	 * 
+	 * @param cmd must not be {@literal null}.
+	 * @param node must not be {@literal null}.
+	 * @throws IllegalArgumentException in case no resource can be acquired for given node.
+	 * @return
+	 */
+	public <S, T> T executeCommandOnSingleNode(ClusterCommandCallback<S, T> cmd, RedisClusterNode node) {
+		return executeCommandOnSingleNode(cmd, node, 0);
+	}
+
+	private <S, T> T executeCommandOnSingleNode(ClusterCommandCallback<S, T> cmd, RedisClusterNode node, int redirectCount) {
+
+		Assert.notNull(cmd, "ClusterCommandCallback must not be null!");
+		Assert.notNull(node, "RedisClusterNode must not be null!");
+
+		if (redirectCount > maxRedirects) {
+			throw new TooManyClusterRedirectionsException(
+					String
+							.format(
+									"Cannot follow Cluster Redirects over more than %s legs. Please consider increasing the number of redirects to follow. Current value is: %s.",
+									redirectCount, maxRedirects));
+		}
+
+		S client = this.resourceProvider.getResourceForSpecificNode(node);
+		Assert.notNull(client, "Could not acquire resource for node. Is your cluster info up to date?");
+
+		try {
+			return cmd.doInCluster(client);
+		} catch (RuntimeException ex) {
+
+			RuntimeException translatedException = convertToDataAccessExeption(ex);
+			if (translatedException instanceof ClusterRedirectException) {
+				ClusterRedirectException cre = (ClusterRedirectException) translatedException;
+				return executeCommandOnSingleNode(cmd,
+						topologyProvider.getTopology().lookup(cre.getTargetHost(), cre.getTargetPort()), redirectCount + 1);
+			} else {
+				throw translatedException != null ? translatedException : ex;
+			}
+		} finally {
+			this.resourceProvider.returnResourceForSpecificNode(node, client);
+		}
+	}
+
+	/**
+	 * Run {@link ClusterCommandCallback} on all reachable master nodes.
+	 * 
+	 * @param cmd
+	 * @return
+	 * @throws ClusterCommandExecutionFailureException
+	 */
+	public <S, T> Map<RedisClusterNode, T> executeCommandOnAllNodes(final ClusterCommandCallback<S, T> cmd) {
+		return executeCommandAsyncOnNodes(cmd, getClusterTopology().getActiveMasterNodes());
+	}
+
+	/**
+	 * @param callback
+	 * @param nodes
+	 * @return
+	 * @throws ClusterCommandExecutionFailureException
+	 */
+	public <S, T> java.util.Map<RedisClusterNode, T> executeCommandAsyncOnNodes(
+			final ClusterCommandCallback<S, T> callback, Iterable<RedisClusterNode> nodes) {
+
+		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>>();
+		for (final RedisClusterNode node : nodes) {
+
+			futures.put(node, executor.submit(new Callable<T>() {
+
+				@Override
+				public T call() throws Exception {
+					return executeCommandOnSingleNode(callback, node);
+				}
+			}));
+		}
+
+		return collectResults(futures);
+	}
+
+	private <T> Map<RedisClusterNode, T> collectResults(Map<RedisClusterNode, Future<T>> futures) {
+
+		boolean done = false;
+
+		Map<RedisClusterNode, T> result = new HashMap<RedisClusterNode, T>();
+		Map<RedisClusterNode, Throwable> exceptions = new HashMap<RedisClusterNode, Throwable>();
+		while (!done) {
+
+			done = true;
+			for (Map.Entry<RedisClusterNode, Future<T>> entry : futures.entrySet()) {
+
+				if (!entry.getValue().isDone() && !entry.getValue().isCancelled()) {
+					done = false;
+				} else {
+					if (!result.containsKey(entry.getKey()) && !exceptions.containsKey(entry.getKey())) {
+						try {
+							result.put(entry.getKey(), entry.getValue().get());
+						} catch (ExecutionException e) {
+
+							RuntimeException ex = convertToDataAccessExeption((Exception) e.getCause());
+							exceptions.put(entry.getKey(), ex != null ? ex : e.getCause());
+						} catch (InterruptedException e) {
+
+							RuntimeException ex = convertToDataAccessExeption((Exception) e.getCause());
+							exceptions.put(entry.getKey(), ex != null ? ex : e.getCause());
+						}
+					}
+				}
+			}
+			try {
+				Thread.sleep(10);
+			} catch (InterruptedException e) {
+
+				done = true;
+				Thread.currentThread().interrupt();
+			}
+		}
+
+		if (!exceptions.isEmpty()) {
+			throw new ClusterCommandExecutionFailureException(new ArrayList<Throwable>(exceptions.values()));
+		}
+		return result;
+	}
+
+	/**
+	 * Run {@link MultiKeyClusterCommandCallback} with on a curated set of nodes serving one or more keys.
+	 * 
+	 * @param cmd
+	 * @return
+	 * @throws ClusterCommandExecutionFailureException
+	 */
+	public <S, T> Map<RedisClusterNode, T> executeMuliKeyCommand(final MultiKeyClusterCommandCallback<S, T> cmd,
+			Iterable<byte[]> keys) {
+
+		Map<RedisClusterNode, Set<byte[]>> nodeKeyMap = new HashMap<RedisClusterNode, Set<byte[]>>();
+
+		for (byte[] key : keys) {
+			for (RedisClusterNode node : getClusterTopology().getKeyServingNodes(key)) {
+
+				if (nodeKeyMap.containsKey(node)) {
+					nodeKeyMap.get(node).add(key);
+				} else {
+					Set<byte[]> keySet = new LinkedHashSet<byte[]>();
+					keySet.add(key);
+					nodeKeyMap.put(node, keySet);
+				}
+			}
+		}
+
+		Map<RedisClusterNode, Future<T>> futures = new LinkedHashMap<RedisClusterNode, Future<T>>();
+		for (final Entry<RedisClusterNode, Set<byte[]>> entry : nodeKeyMap.entrySet()) {
+
+			if (entry.getKey().isMaster()) {
+				for (final byte[] key : entry.getValue()) {
+					futures.put(entry.getKey(), executor.submit(new Callable<T>() {
+
+						@Override
+						public T call() throws Exception {
+							return (T) executeMultiKeyCommandOnSingleNode(cmd, entry.getKey(), key);
+						}
+					}));
+				}
+			}
+		}
+		return collectResults(futures);
+	}
+
+	private <S, T> T executeMultiKeyCommandOnSingleNode(MultiKeyClusterCommandCallback<S, T> cmd, RedisClusterNode node,
+			byte[] key) {
+
+		Assert.notNull(cmd, "MultiKeyCommandCallback must not be null!");
+		Assert.notNull(node, "RedisClusterNode must not be null!");
+		Assert.notNull(key, "Keys for execution must not be null!");
+
+		S client = this.resourceProvider.getResourceForSpecificNode(node);
+		Assert.notNull(client, "Could not acquire resource for node. Is your cluster info up to date?");
+
+		try {
+			return cmd.doInCluster(client, key);
+		} catch (RuntimeException ex) {
+
+			RuntimeException translatedException = convertToDataAccessExeption(ex);
+			throw translatedException != null ? translatedException : ex;
+		} finally {
+			this.resourceProvider.returnResourceForSpecificNode(node, client);
+		}
+	}
+
+	private ClusterTopology getClusterTopology() {
+		return this.topologyProvider.getTopology();
+	}
+
+	private DataAccessException convertToDataAccessExeption(Exception e) {
+		return exceptionTranslationStrategy.translate(e);
+	}
+
+	/**
+	 * Set the maximum number of redirects to follow on {@code MOVED} or {@code ASK}.
+	 * 
+	 * @param maxRedirects set to zero to suspend redirects.
+	 */
+	public void setMaxRedirects(int maxRedirects) {
+		this.maxRedirects = maxRedirects;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.beans.factory.DisposableBean#destroy()
+	 */
+	@Override
+	public void destroy() throws Exception {
+
+		if (executor instanceof DisposableBean) {
+			((DisposableBean) executor).destroy();
+		}
+	}
+
+	/**
+	 * Callback interface for Redis 'low level' code using the cluster client directly. To be used with
+	 * {@link ClusterCommandExecutor} execution methods.
+	 * 
+	 * @author Christoph Strobl
+	 * @param <T> native driver connection
+	 * @param <S>
+	 * @since 1.7
+	 */
+	public static interface ClusterCommandCallback<T, S> {
+		S doInCluster(T client);
+	}
+
+	/**
+	 * Callback interface for Redis 'low level' code using the cluster client to execute multi key commands.
+	 * 
+	 * @author Christoph Strobl
+	 * @param <T> native driver connection
+	 * @param <S>
+	 */
+	public static interface MultiKeyClusterCommandCallback<T, S> {
+		S doInCluster(T client, byte[] key);
+	}
+
+}

src/main/java/org/springframework/data/redis/connection/ClusterInfo.java

@@ -0,0 +1,173 @@
+/*
+ * Copyright 2015 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.connection;
+
+import java.util.Properties;
+
+import org.springframework.data.redis.core.types.RedisClientInfo.INFO;
+import org.springframework.util.Assert;
+
+/**
+ * {@link ClusterInfo} gives access to cluster information such as {@code cluster_state} and
+ * {@code cluster_slots_assigned} provided by the {@code CLUSTER INFO} command.
+ * 
+ * @author Christoph Strobl
+ * @since 1.7
+ */
+public class ClusterInfo {
+
+	public static enum Info {
+		STATE("cluster_state"), SLOTS_ASSIGNED("cluster_slots_assigned"), SLOTS_OK("cluster_slots_ok"), SLOTS_PFAIL(
+				"cluster_slots_pfail"), SLOTS_FAIL("cluster_slots_fail"), KNOWN_NODES("cluster_known_nodes"), SIZE(
+				"cluster_size"), CURRENT_EPOCH("cluster_current_epoch"), MY_EPOCH("cluster_my_epoch"), MESSAGES_SENT(
+				"cluster_stats_messages_sent"), MESSAGES_RECEIVED("cluster_stats_messages_received");
+
+		String key;
+
+		Info(String key) {
+			this.key = key;
+		}
+	}
+
+	private final Properties clusterProperties;
+
+	/**
+	 * Creates new {@link ClusterInfo} for given {@link Properties}.
+	 * 
+	 * @param clusterProperties must not be {@literal null}.
+	 */
+	public ClusterInfo(Properties clusterProperties) {
+
+		Assert.notNull(clusterProperties, "ClusterProperties must not be null!");
+		this.clusterProperties = clusterProperties;
+	}
+
+	/**
+	 * @see Info#STATE
+	 * @return
+	 */
+	public String getState() {
+		return get(Info.STATE);
+	}
+
+	/**
+	 * @see Info#SLOTS_ASSIGNED
+	 * @return
+	 */
+	public Long getSlotsAssigned() {
+		return getLongValueOf(Info.SLOTS_ASSIGNED);
+	}
+
+	/**
+	 * @see Info#SLOTS_OK
+	 * @return
+	 */
+	public Long getSlotsOk() {
+		return getLongValueOf(Info.SLOTS_OK);
+	}
+
+	/**
+	 * @see Info#SLOTS_PFAIL
+	 * @return
+	 */
+	public Long getSlotsPfail() {
+		return getLongValueOf(Info.SLOTS_PFAIL);
+	}
+
+	/**
+	 * @see Info#SLOTS_FAIL
+	 * @return
+	 */
+	public Long getSlotsFail() {
+		return getLongValueOf(Info.SLOTS_FAIL);
+	}
+
+	/**
+	 * @see Info#KNOWN_NODES
+	 * @return
+	 */
+	public Long getKnownNodes() {
+		return getLongValueOf(Info.KNOWN_NODES);
+	}
+
+	/**
+	 * @see Info#SIZE
+	 * @return
+	 */
+	public Long getClusterSize() {
+		return getLongValueOf(Info.SIZE);
+	}
+
+	/**
+	 * @see Info#CURRENT_EPOCH
+	 * @return
+	 */
+	public Long getCurrentEpoch() {
+		return getLongValueOf(Info.CURRENT_EPOCH);
+	}
+
+	/**
+	 * @see Info#MESSAGES_SENT
+	 * @return
+	 */
+	public Long getMessagesSent() {
+		return getLongValueOf(Info.MESSAGES_SENT);
+	}
+
+	/**
+	 * @see Info#MESSAGES_RECEIVED
+	 * @return
+	 */
+	public Long getMessagesReceived() {
+		return getLongValueOf(Info.MESSAGES_RECEIVED);
+	}
+
+	/**
+	 * @param info must not be null
+	 * @return {@literal null} if no entry found for requested {@link INFO}.
+	 */
+	public String get(Info info) {
+
+		Assert.notNull(info, "Cannot retrieve cluster information for 'null'.");
+		return get(info.key);
+	}
+
+	/**
+	 * @param key must not be {@literal null} or {@literal empty}.
+	 * @return {@literal null} if no entry found for requested {@code key}.
+	 */
+	public String get(String key) {
+
+		Assert.hasText(key, "Cannot get cluster information for 'empty' / 'null' key.");
+		return this.clusterProperties.getProperty(key);
+	}
+
+	private Long getLongValueOf(Info info) {
+
+		String value = get(info);
+		return value == null ? null : Long.valueOf(value);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see java.lang.Object#toString()
+	 */
+	@Override
+	public String toString() {
+		return this.clusterProperties.toString();
+	}
+
+}

src/main/java/org/springframework/data/redis/connection/ClusterNodeResourceProvider.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2015 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.connection;
+
+/**
+ * {@link ClusterNodeResourceProvider} provides access to low level client api to directly execute operations against a
+ * Redis instance.
+ * 
+ * @author Christoph Strobl
+ * @since 1.7
+ */
+public interface ClusterNodeResourceProvider {
+
+	/**
+	 * Get the client resource for the given node.
+	 * 
+	 * @param node must not be {@literal null}.
+	 * @return
+	 */
+	<S> S getResourceForSpecificNode(RedisClusterNode node);
+
+	/**
+	 * Return the resource object for the given node. This can mean free up resources or return elements back to a pool.
+	 * 
+	 * @param node must not be {@literal null}.
+	 * @param resource must not be {@literal null}.
+	 */
+	void returnResourceForSpecificNode(RedisClusterNode node, Object resource);
+
+}

src/main/java/org/springframework/data/redis/connection/ClusterSlotHashUtil.java

@@ -0,0 +1,137 @@
+/*
+ * Copyright 2015 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.connection;
+
+import org.springframework.util.Assert;
+
+/**
+ * @author Christoph Strobl
+ * @since 1.7
+ */
+public final class ClusterSlotHashUtil {
+
+	private static final int SLOT_COUNT = 16384;
+
+	private static final byte SUBKEY_START = '{';
+	private static final byte SUBKEY_END = '}';
+
+	private static final int[] LOOKUP_TABLE = { 0x0000, 0x1021, 0x2042, 0x3063, 0x4084, 0x50A5, 0x60C6, 0x70E7, 0x8108,
+			0x9129, 0xA14A, 0xB16B, 0xC18C, 0xD1AD, 0xE1CE, 0xF1EF, 0x1231, 0x0210, 0x3273, 0x2252, 0x52B5, 0x4294, 0x72F7,
+			0x62D6, 0x9339, 0x8318, 0xB37B, 0xA35A, 0xD3BD, 0xC39C, 0xF3FF, 0xE3DE, 0x2462, 0x3443, 0x0420, 0x1401, 0x64E6,
+			0x74C7, 0x44A4, 0x5485, 0xA56A, 0xB54B, 0x8528, 0x9509, 0xE5EE, 0xF5CF, 0xC5AC, 0xD58D, 0x3653, 0x2672, 0x1611,
+			0x0630, 0x76D7, 0x66F6, 0x5695, 0x46B4, 0xB75B, 0xA77A, 0x9719, 0x8738, 0xF7DF, 0xE7FE, 0xD79D, 0xC7BC, 0x48C4,
+			0x58E5, 0x6886, 0x78A7, 0x0840, 0x1861, 0x2802, 0x3823, 0xC9CC, 0xD9ED, 0xE98E, 0xF9AF, 0x8948, 0x9969, 0xA90A,
+			0xB92B, 0x5AF5, 0x4AD4, 0x7AB7, 0x6A96, 0x1A71, 0x0A50, 0x3A33, 0x2A12, 0xDBFD, 0xCBDC, 0xFBBF, 0xEB9E, 0x9B79,
+			0x8B58, 0xBB3B, 0xAB1A, 0x6CA6, 0x7C87, 0x4CE4, 0x5CC5, 0x2C22, 0x3C03, 0x0C60, 0x1C41, 0xEDAE, 0xFD8F, 0xCDEC,
+			0xDDCD, 0xAD2A, 0xBD0B, 0x8D68, 0x9D49, 0x7E97, 0x6EB6, 0x5ED5, 0x4EF4, 0x3E13, 0x2E32, 0x1E51, 0x0E70, 0xFF9F,
+			0xEFBE, 0xDFDD, 0xCFFC, 0xBF1B, 0xAF3A, 0x9F59, 0x8F78, 0x9188, 0x81A9, 0xB1CA, 0xA1EB, 0xD10C, 0xC12D, 0xF14E,
+			0xE16F, 0x1080, 0x00A1, 0x30C2, 0x20E3, 0x5004, 0x4025, 0x7046, 0x6067, 0x83B9, 0x9398, 0xA3FB, 0xB3DA, 0xC33D,
+			0xD31C, 0xE37F, 0xF35E, 0x02B1, 0x1290, 0x22F3, 0x32D2, 0x4235, 0x5214, 0x6277, 0x7256, 0xB5EA, 0xA5CB, 0x95A8,
+			0x8589, 0xF56E, 0xE54F, 0xD52C, 0xC50D, 0x34E2, 0x24C3, 0x14A0, 0x0481, 0x7466, 0x6447, 0x5424, 0x4405, 0xA7DB,
+			0xB7FA, 0x8799, 0x97B8, 0xE75F, 0xF77E, 0xC71D, 0xD73C, 0x26D3, 0x36F2, 0x0691, 0x16B0, 0x6657, 0x7676, 0x4615,
+			0x5634, 0xD94C, 0xC96D, 0xF90E, 0xE92F, 0x99C8, 0x89E9, 0xB98A, 0xA9AB, 0x5844, 0x4865, 0x7806, 0x6827, 0x18C0,
+			0x08E1, 0x3882, 0x28A3, 0xCB7D, 0xDB5C, 0xEB3F, 0xFB1E, 0x8BF9, 0x9BD8, 0xABBB, 0xBB9A, 0x4A75, 0x5A54, 0x6A37,
+			0x7A16, 0x0AF1, 0x1AD0, 0x2AB3, 0x3A92, 0xFD2E, 0xED0F, 0xDD6C, 0xCD4D, 0xBDAA, 0xAD8B, 0x9DE8, 0x8DC9, 0x7C26,
+			0x6C07, 0x5C64, 0x4C45, 0x3CA2, 0x2C83, 0x1CE0, 0x0CC1, 0xEF1F, 0xFF3E, 0xCF5D, 0xDF7C, 0xAF9B, 0xBFBA, 0x8FD9,
+			0x9FF8, 0x6E17, 0x7E36, 0x4E55, 0x5E74, 0x2E93, 0x3EB2, 0x0ED1, 0x1EF0 };
+
+	private ClusterSlotHashUtil() {
+
+	}
+
+	/**
+	 * @param keys must not be {@literal null}.
+	 * @return
+	 */
+	public static boolean isSameSlotForAllKeys(byte[]... keys) {
+
+		Assert.notNull(keys, "Keys must not be null!");
+
+		if (keys.length <= 1) {
+			return true;
+		}
+
+		int slot = calculateSlot(keys[0]);
+		for (int i = 1; i < keys.length; i++) {
+			if (slot != calculateSlot(keys[i])) {
+				return false;
+			}
+		}
+		return true;
+	}
+
+	/**
+	 * Calculate the slot from the given key.
+	 * 
+	 * @param key must not be {@literal null} or empty.
+	 * @return
+	 */
+	public static int calculateSlot(String key) {
+
+		Assert.hasText(key, "Key must not be null or empty!");
+		return calculateSlot(key.getBytes());
+	}
+
+	/**
+	 * Calculate the slot from the given key.
+	 * 
+	 * @param key must not be {@literal null}.
+	 * @return
+	 */
+	public static int calculateSlot(byte[] key) {
+
+		Assert.notNull(key, "Key must not be null!");
+
+		byte[] finalKey = key;
+		int start = indexOf(key, SUBKEY_START);
+		if (start != -1) {
+			int end = indexOf(key, start + 1, SUBKEY_END);
+			if (end != -1 && end != start + 1) {
+
+				finalKey = new byte[end - (start + 1)];
+				System.arraycopy(key, start + 1, finalKey, 0, finalKey.length);
+			}
+		}
+		return crc16(finalKey) % SLOT_COUNT;
+	}
+
+	private static int indexOf(byte[] haystack, byte needle) {
+		return indexOf(haystack, 0, needle);
+	}
+
+	private static int indexOf(byte[] haystack, int start, byte needle) {
+
+		for (int i = start; i < haystack.length; i++) {
+
+			if (haystack[i] == needle) {
+				return i;
+			}
+		}
+
+		return -1;
+	}
+
+	private static int crc16(byte[] bytes) {
+
+		int crc = 0x0000;
+
+		for (byte b : bytes) {
+			crc = ((crc << 8) ^ LOOKUP_TABLE[((crc >>> 8) ^ (b & 0xFF)) & 0xFF]);
+		}
+		return crc & 0xFFFF;
+	}
+
+}

src/main/java/org/springframework/data/redis/connection/ClusterTopology.java

@@ -0,0 +1,169 @@
+/*
+ * Copyright 2015 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.connection;
+
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.Set;
+
+import org.springframework.data.redis.ClusterStateFailureExeption;
+import org.springframework.util.Assert;
+
+/**
+ * {@link ClusterTopology} holds snapshot like information about {@link RedisClusterNode}s.
+ * 
+ * @author Christoph Strobl
+ * @since 1.7
+ */
+public class ClusterTopology {
+
+	private final Set<RedisClusterNode> nodes;
+
+	/**
+	 * Creates new instance of {@link ClusterTopology}.
+	 * 
+	 * @param nodes can be {@literal null}.
+	 */
+	public ClusterTopology(Set<RedisClusterNode> nodes) {
+		this.nodes = nodes != null ? nodes : Collections.<RedisClusterNode> emptySet();
+	}
+
+	/**
+	 * Get all {@link RedisClusterNode}s.
+	 * 
+	 * @return never {@literal null}.
+	 */
+	public Set<RedisClusterNode> getNodes() {
+		return Collections.unmodifiableSet(nodes);
+	}
+
+	/**
+	 * Get all nodes (master and slave) in cluster where {@code link-state} is {@literal connected} and {@code flags} does
+	 * not contain {@literal fail} or {@literal fail?}.
+	 * 
+	 * @return never {@literal null}.
+	 */
+	public Set<RedisClusterNode> getActiveNodes() {
+
+		Set<RedisClusterNode> activeNodes = new LinkedHashSet<RedisClusterNode>(nodes.size());
+		for (RedisClusterNode node : nodes) {
+			if (node.isConnected() && !node.isMarkedAsFail()) {
+				activeNodes.add(node);
+			}
+		}
+		return activeNodes;
+	}
+
+	/**
+	 * Get all master nodes in cluster where {@code link-state} is {@literal connected} and {@code flags} does not contain
+	 * {@literal fail} or {@literal fail?}.
+	 * 
+	 * @return never {@literal null}.
+	 */
+	public Set<RedisClusterNode> getActiveMasterNodes() {
+
+		Set<RedisClusterNode> activeMasterNodes = new LinkedHashSet<RedisClusterNode>(nodes.size());
+		for (RedisClusterNode node : nodes) {
+			if (node.isMaster() && node.isConnected() && !node.isMarkedAsFail()) {
+				activeMasterNodes.add(node);
+			}
+		}
+		return activeMasterNodes;
+	}
+
+	/**
+	 * Get all master nodes in cluster.
+	 * 
+	 * @return never {@literal null}.
+	 */
+	public Set<RedisClusterNode> getMasterNodes() {
+
+		Set<RedisClusterNode> masterNodes = new LinkedHashSet<RedisClusterNode>(nodes.size());
+		for (RedisClusterNode node : nodes) {
+			if (node.isMaster()) {
+				masterNodes.add(node);
+			}
+		}
+		return masterNodes;
+	}
+
+	/**
+	 * Get the {@link RedisClusterNode}s (master and slave) serving s specific slot.
+	 * 
+	 * @param slot
+	 * @return never {@literal null}.
+	 */
+	public Set<RedisClusterNode> getSlotServingNodes(int slot) {
+
+		Set<RedisClusterNode> slotServingNodes = new LinkedHashSet<RedisClusterNode>(nodes.size());
+		for (RedisClusterNode node : nodes) {
+			if (node.servesSlot(slot)) {
+				slotServingNodes.add(node);
+			}
+		}
+		return slotServingNodes;
+	}
+
+	/**
+	 * Get the {@link RedisClusterNode} that is the current master serving the given key.
+	 * 
+	 * @param key must not be {@literal null}.
+	 * @return
+	 * @throws ClusterStateFailureExeption
+	 */
+	public RedisClusterNode getKeyServingMasterNode(byte[] key) {
+
+		Assert.notNull(key, "Key for node lookup must not be null!");
+
+		int slot = ClusterSlotHashUtil.calculateSlot(key);
+		for (RedisClusterNode node : nodes) {
+			if (node.isMaster() && node.servesSlot(slot)) {
+				return node;
+			}
+		}
+		throw new ClusterStateFailureExeption(String.format("Could not find master node serving slot %s for key '%s',",
+				slot, key));
+	}
+
+	/**
+	 * Get the {@link RedisClusterNode} matching given {@literal host} and {@literal port}.
+	 * 
+	 * @param host must not be {@literal null}.
+	 * @param port
+	 * @return
+	 * @throws ClusterStateFailureExeption
+	 */
+	public RedisClusterNode lookup(String host, int port) {
+
+		for (RedisClusterNode node : nodes) {
+			if (host.equals(node.getHost()) && port == node.getPort()) {
+				return node;
+			}
+		}
+		throw new ClusterStateFailureExeption(String.format(
+				"Could not find node at %s:%s. Is your cluster info up to date?", host, port));
+	}
+
+	/**
+	 * @param key must not be {@literal null}.
+	 * @return {@literal null}.
+	 */
+	public Set<RedisClusterNode> getKeyServingNodes(byte[] key) {
+
+		Assert.notNull(key, "Key must not be null for Cluster Node lookup.");
+		return getSlotServingNodes(ClusterSlotHashUtil.calculateSlot(key));
+	}
+}

src/main/java/org/springframework/data/redis/connection/ClusterTopologyProvider.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2015 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.connection;
+
+/**
+ * {@link ClusterTopologyProvider} manages the current cluster topology and makes sure to refresh cluster information.
+ * 
+ * @author Christoph Strobl
+ * @since 1.7
+ */
+public interface ClusterTopologyProvider {
+
+	/**
+	 * Get the current known {@link ClusterTopology}.
+	 * 
+	 * @return never {@null}.
+	 */
+	ClusterTopology getTopology();
+
+}

src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java

@@ -2688,4 +2688,22 @@ public class DefaultStringRedisConnection implements StringRedisConnection {
 		return byteSetToStringSet.convert(results);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisServerCommands#migrate(byte[], org.springframework.data.redis.connection.RedisNode, int, org.springframework.data.redis.connection.RedisServerCommands.MigrateOption)
+	 */
+	@Override
+	public void migrate(byte[] key, RedisNode target, int dbIndex, MigrateOption option) {
+		delegate.migrate(key, target, dbIndex, option);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisServerCommands#migrate(byte[], org.springframework.data.redis.connection.RedisNode, int, org.springframework.data.redis.connection.RedisServerCommands.MigrateOption, long)
+	 */
+	@Override
+	public void migrate(byte[] key, RedisNode target, int dbIndex, MigrateOption option, long timeout) {
+		delegate.migrate(key, target, dbIndex, option, timeout);
+	}
+
 }

src/main/java/org/springframework/data/redis/connection/RedisClusterCommands.java

@@ -0,0 +1,167 @@
+/*
+ * Copyright 2015 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.connection;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+
+import org.springframework.data.redis.connection.RedisClusterNode.SlotRange;
+
+/**
+ * Interface for the {@literal cluster} commands supported by Redis.
+ * 
+ * @author Christoph Strobl
+ * @since 1.7
+ */
+public interface RedisClusterCommands {
+
+	/**
+	 * Retrieve cluster node information such as {@literal id}, {@literal host}, {@literal port} and {@literal slots}.
+	 * 
+	 * @return never {@literal null}.
+	 */
+	Iterable<RedisClusterNode> clusterGetClusterNodes();
+
+	/**
+	 * Retrieve information about connected slaves for given master node.
+	 * 
+	 * @param node must not be {@literal null}.
+	 * @return never {@literal null}.
+	 */
+	Collection<RedisClusterNode> clusterGetSlaves(RedisClusterNode master);
+
+	/**
+	 * Retrieve information about masters and their connected slaves.
+	 * 
+	 * @return never {@literal null}.
+	 */
+	Map<RedisClusterNode, Collection<RedisClusterNode>> clusterGetMasterSlaveMap();
+
+	/**
+	 * Find the slot for a given {@code key}.
+	 * 
+	 * @param key must not be {@literal null}.
+	 * @return
+	 */
+	Integer clusterGetSlotForKey(byte[] key);
+
+	/**
+	 * Find the {@link RedisClusterNode} serving given {@literal slot}.
+	 * 
+	 * @param slot
+	 * @return
+	 */
+	RedisClusterNode clusterGetNodeForSlot(int slot);
+
+	/**
+	 * Find the {@link RedisClusterNode} serving given {@literal key}.
+	 * 
+	 * @param key must not be {@literal null}.
+	 * @return
+	 */
+	RedisClusterNode clusterGetNodeForKey(byte[] key);
+
+	/**
+	 * Get cluster information.
+	 * 
+	 * @return
+	 */
+	ClusterInfo clusterGetClusterInfo();
+
+	/**
+	 * Assign slots to given {@link RedisClusterNode}.
+	 * 
+	 * @param node must not be {@literal null}.
+	 * @param slots
+	 */
+	void clusterAddSlots(RedisClusterNode node, int... slots);
+
+	/**
+	 * Assign {@link SlotRange#getSlotsArray()} to given {@link RedisClusterNode}.
+	 * 
+	 * @param node must not be {@literal null}.
+	 * @param range must not be {@literal null}.
+	 */
+	void clusterAddSlots(RedisClusterNode node, SlotRange range);
+
+	/**
+	 * Count the number of keys assigned to one {@literal slot}.
+	 * 
+	 * @param slot
+	 * @return
+	 */
+	Long clusterCountKeysInSlot(int slot);
+
+	/**
+	 * Remove slots from {@link RedisClusterNode}.
+	 * 
+	 * @param node must not be {@literal null}.
+	 * @param slots
+	 */
+	void clusterDeleteSlots(RedisClusterNode node, int... slots);
+
+	/**
+	 * Removes {@link SlotRange#getSlotsArray()} from given {@link RedisClusterNode}.
+	 * 
+	 * @param node must not be {@literal null}.
+	 * @param range must not be {@literal null}.
+	 */
+	void clusterDeleteSlotsInRange(RedisClusterNode node, SlotRange range);
+
+	/**
+	 * Remove given {@literal node} from cluster.
+	 * 
+	 * @param node must not be {@literal null}.
+	 */
+	void clusterForget(RedisClusterNode node);
+
+	/**
+	 * Add given {@literal node} to cluster.
+	 * 
+	 * @param node must not be {@literal null}.
+	 */
+	void clusterMeet(RedisClusterNode node);
+
+	/**
+	 * @param node must not be {@literal null}.
+	 * @param slot
+	 * @param mode must not be{@literal null}.
+	 */
+	void clusterSetSlot(RedisClusterNode node, int slot, AddSlots mode);
+
+	/**
+	 * Get {@literal keys} served by slot.
+	 * 
+	 * @param slot
+	 * @param count must not be {@literal null}.
+	 * @return
+	 */
+	List<byte[]> clusterGetKeysInSlot(int slot, Integer count);
+
+	/**
+	 * Assign a {@literal slave} to given {@literal master}.
+	 * 
+	 * @param master must not be {@literal null}.
+	 * @param slave must not be {@literal null}.
+	 */
+	void clusterReplicate(RedisClusterNode master, RedisClusterNode slave);
+
+	public enum AddSlots {
+		MIGRATING, IMPORTING, STABLE, NODE
+	}
+
+}

src/main/java/org/springframework/data/redis/connection/RedisClusterConfiguration.java

@@ -0,0 +1,211 @@
+/*
+ * Copyright 2015 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.connection;
+
+import static org.springframework.util.Assert.*;
+import static org.springframework.util.StringUtils.*;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedHashSet;
+import java.util.Map;
+import java.util.Set;
+
+import org.springframework.core.env.MapPropertySource;
+import org.springframework.core.env.PropertySource;
+import org.springframework.util.NumberUtils;
+import org.springframework.util.StringUtils;
+
+/**
+ * Configuration class used for setting up {@link RedisConnection} via {@link RedisConnectionFactory} using connecting
+ * to <a href="http://redis.io/topics/cluster-spec">Redis Cluster</a>. Useful when setting up a high availability Redis
+ * environment.
+ * 
+ * @author Christoph Strobl
+ * @since 1.7
+ */
+public class RedisClusterConfiguration {
+
+	private static final String REDIS_CLUSTER_NODES_CONFIG_PROPERTY = "spring.redis.cluster.nodes";
+	private static final String REDIS_CLUSTER_TIMEOUT_CONFIG_PROPERTY = "spring.redis.cluster.timeout";
+	private static final String REDIS_CLUSTER_MAX_REDIRECTS_CONFIG_PROPERTY = "spring.redis.cluster.max-redirects";
+
+	private Set<RedisNode> clusterNodes;
+	private Long clusterTimeout;
+	private Integer maxRedirects;
+
+	/**
+	 * Creates new {@link RedisClusterConfiguration}.
+	 */
+	public RedisClusterConfiguration() {
+		this(new MapPropertySource("RedisClusterConfiguration", Collections.<String, Object> emptyMap()));
+	}
+
+	/**
+	 * Creates {@link RedisClusterConfiguration} for given hostPort combinations.
+	 * 
+	 * <pre>
+	 * clusterHostAndPorts[0] = 127.0.0.1:23679
+	 * clusterHostAndPorts[1] = 127.0.0.1:23680
+	 * ...
+	 * 
+	 * <pre>
+	 * 
+	 * @param cluster must not be
+	 * {@literal null}.
+	 */
+	public RedisClusterConfiguration(Collection<String> clusterNodes) {
+		this(new MapPropertySource("RedisClusterConfiguration", asMap(clusterNodes, -1, -1)));
+	}
+
+	/**
+	 * Creates {@link RedisClusterConfiguration} looking up values in given {@link PropertySource}.
+	 * 
+	 * <pre>
+	 * <code>
+	 * spring.redis.cluster.nodes=127.0.0.1:23679,127.0.0.1:23680,127.0.0.1:23681
+	 * spring.redis.cluster.timeout=5
+	 * spring.redis.cluster.max-redirects=3
+	 * </code>
+	 * </pre>
+	 * 
+	 * @param propertySource must not be {@literal null}.
+	 */
+	public RedisClusterConfiguration(PropertySource<?> propertySource) {
+
+		notNull(propertySource, "PropertySource must not be null!");
+
+		this.clusterNodes = new LinkedHashSet<RedisNode>();
+
+		if (propertySource.containsProperty(REDIS_CLUSTER_NODES_CONFIG_PROPERTY)) {
+			appendClusterNodes(commaDelimitedListToSet(propertySource.getProperty(REDIS_CLUSTER_NODES_CONFIG_PROPERTY)
+					.toString()));
+		}
+		if (propertySource.containsProperty(REDIS_CLUSTER_TIMEOUT_CONFIG_PROPERTY)) {
+			this.clusterTimeout = NumberUtils.parseNumber(propertySource.getProperty(REDIS_CLUSTER_TIMEOUT_CONFIG_PROPERTY)
+					.toString(), Long.class);
+		}
+		if (propertySource.containsProperty(REDIS_CLUSTER_MAX_REDIRECTS_CONFIG_PROPERTY)) {
+			this.maxRedirects = NumberUtils.parseNumber(
+					propertySource.getProperty(REDIS_CLUSTER_MAX_REDIRECTS_CONFIG_PROPERTY).toString(), Integer.class);
+		}
+	}
+
+	/**
+	 * Set {@literal cluster nodes} to connect to.
+	 * 
+	 * @param nodes must not be {@literal null}.
+	 */
+	public void setClusterNodes(Iterable<RedisNode> nodes) {
+
+		notNull(nodes, "Cannot set cluster nodes to 'null'.");
+
+		this.clusterNodes.clear();
+
+		for (RedisNode clusterNode : nodes) {
+			addClusterNode(clusterNode);
+		}
+	}
+
+	/**
+	 * Returns an {@link Collections#unmodifiableSet(Set)} of {@literal cluster nodes}.
+	 * 
+	 * @return {@link Set} of nodes. Never {@literal null}.
+	 */
+	public Set<RedisNode> getClusterNodes() {
+		return Collections.unmodifiableSet(clusterNodes);
+	}
+
+	/**
+	 * Add a cluster node to configuration.
+	 * 
+	 * @param node must not be {@literal null}.
+	 */
+	public void addClusterNode(RedisNode node) {
+
+		notNull(node, "ClusterNode must not be 'null'.");
+		this.clusterNodes.add(node);
+	}
+
+	/**
+	 * @return
+	 */
+	public RedisClusterConfiguration clusterNode(RedisNode node) {
+		this.clusterNodes.add(node);
+		return this;
+	}
+
+	/**
+	 * @return
+	 */
+	public Long getClusterTimeout() {
+		return clusterTimeout != null && clusterTimeout > Long.MIN_VALUE ? clusterTimeout : null;
+	}
+
+	/**
+	 * @return
+	 */
+	public Integer getMaxRedirects() {
+		return maxRedirects != null && maxRedirects > Integer.MIN_VALUE ? maxRedirects : null;
+	}
+
+	/**
+	 * @param host
+	 * @param port
+	 * @return
+	 */
+	public RedisClusterConfiguration clusterNode(String host, Integer port) {
+		return clusterNode(new RedisNode(host, port));
+	}
+
+	private void appendClusterNodes(Set<String> hostAndPorts) {
+
+		for (String hostAndPort : hostAndPorts) {
+			addClusterNode(readHostAndPortFromString(hostAndPort));
+		}
+	}
+
+	private RedisNode readHostAndPortFromString(String hostAndPort) {
+
+		String[] args = split(hostAndPort, ":");
+
+		notNull(args, "HostAndPort need to be seperated by  ':'.");
+		isTrue(args.length == 2, "Host and Port String needs to specified as host:port");
+		return new RedisNode(args[0], Integer.valueOf(args[1]).intValue());
+	}
+
+	/**
+	 * @param master must not be {@literal null} or empty.
+	 * @param clusterHostAndPorts must not be {@literal null}.
+	 * @return
+	 */
+	private static Map<String, Object> asMap(Collection<String> clusterHostAndPorts, long timeout, int redirects) {
+
+		notNull(clusterHostAndPorts, "ClusterHostAndPorts must not be null!");
+
+		Map<String, Object> map = new HashMap<String, Object>();
+		map.put(REDIS_CLUSTER_NODES_CONFIG_PROPERTY, StringUtils.collectionToCommaDelimitedString(clusterHostAndPorts));
+		if (timeout >= 0) {
+			map.put(REDIS_CLUSTER_TIMEOUT_CONFIG_PROPERTY, Long.valueOf(timeout));
+		}
+		if (redirects >= 0) {
+			map.put(REDIS_CLUSTER_MAX_REDIRECTS_CONFIG_PROPERTY, Integer.valueOf(redirects));
+		}
+
+		return map;
+	}
+}

src/main/java/org/springframework/data/redis/connection/RedisClusterConnection.java

@@ -0,0 +1,155 @@
+/*
+ * Copyright 2015 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.connection;
+
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+
+import org.springframework.data.redis.core.types.RedisClientInfo;
+
+/**
+ * {@link RedisClusterConnection} allows sending commands to dedicated nodes within the cluster.
+ * 
+ * @author Christoph Strobl
+ * @since 1.7
+ */
+public interface RedisClusterConnection extends RedisConnection, RedisClusterCommands {
+
+	/**
+	 * @param node must not be {@literal null}.
+	 * @return
+	 * @see RedisConnectionCommands#ping()
+	 */
+	String ping(RedisClusterNode node);
+
+	/**
+	 * @param node must not be {@literal null}.
+	 * @see RedisServerCommands#bgReWriteAof()
+	 */
+	void bgReWriteAof(RedisClusterNode node);
+
+	/**
+	 * @param node must not be {@literal null}.
+	 * @see RedisServerCommands#bgSave()
+	 */
+	void bgSave(RedisClusterNode node);
+
+	/**
+	 * @param node must not be {@literal null}.
+	 * @return
+	 * @see RedisServerCommands#lastSave()
+	 */
+	Long lastSave(RedisClusterNode node);
+
+	/**
+	 * @param node must not be {@literal null}.
+	 * @see RedisServerCommands#save()
+	 */
+	void save(RedisClusterNode node);
+
+	/**
+	 * @param node must not be {@literal null}.
+	 * @return
+	 * @see RedisServerCommands#dbSize()
+	 */
+	Long dbSize(RedisClusterNode node);
+
+	/**
+	 * @param node must not be {@literal null}.
+	 * @see RedisServerCommands#flushDb()
+	 */
+	void flushDb(RedisClusterNode node);
+
+	/**
+	 * @param node must not be {@literal null}.
+	 * @see RedisServerCommands#flushAll()
+	 */
+	void flushAll(RedisClusterNode node);
+
+	/**
+	 * @param node must not be {@literal null}.
+	 * @return
+	 * @see RedisServerCommands#info()
+	 */
+	Properties info(RedisClusterNode node);
+
+	/**
+	 * @param node must not be {@literal null}.
+	 * @param section
+	 * @return
+	 * @see RedisServerCommands#info(String)
+	 */
+	Properties info(RedisClusterNode node, String section);
+
+	/**
+	 * @param node must not be {@literal null}.
+	 * @param pattern must not be {@literal null}.
+	 * @return
+	 * @see RedisKeyCommands#keys(byte[])
+	 */
+	Set<byte[]> keys(RedisClusterNode node, byte[] pattern);
+
+	/**
+	 * @param node must not be {@literal null}.
+	 * @return
+	 * @see RedisKeyCommands#randomKey()
+	 */
+	byte[] randomKey(RedisClusterNode node);
+
+	/**
+	 * @param node must not be {@literal null}.
+	 * @see RedisServerCommands#shutdown()
+	 */
+	void shutdown(RedisClusterNode node);
+
+	/**
+	 * @param node must not be {@literal null}.
+	 * @param pattern
+	 * @return
+	 * @see RedisServerCommands#getConfig(String)
+	 */
+	List<String> getConfig(RedisClusterNode node, String pattern);
+
+	/**
+	 * @param node must not be {@literal null}.
+	 * @param param
+	 * @param value
+	 * @see RedisServerCommands#setConfig(String, String)
+	 */
+	void setConfig(RedisClusterNode node, String param, String value);
+
+	/**
+	 * @param node must not be {@literal null}.
+	 * @see RedisServerCommands#resetConfigStats()
+	 */
+	void resetConfigStats(RedisClusterNode node);
+
+	/**
+	 * @param node must not be {@literal null}.
+	 * @return
+	 * @see RedisServerCommands#time()
+	 */
+	Long time(RedisClusterNode node);
+
+	/**
+	 * @param node must not be {@literal null}.
+	 * @return
+	 * @see RedisServerCommands#getClientList()
+	 */
+	public List<RedisClientInfo> getClientList(RedisClusterNode node);
+
+}

src/main/java/org/springframework/data/redis/connection/RedisClusterNode.java

@@ -0,0 +1,342 @@
+/*
+ * Copyright 2015 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.connection;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.Set;
+
+import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
+
+/**
+ * Representation of a Redis server within the cluster.
+ * 
+ * @author Christoph Strobl
+ * @since 1.7
+ */
+public class RedisClusterNode extends RedisNode {
+
+	private SlotRange slotRange;
+	private LinkState linkState;
+	private Set<Flag> flags;
+
+	protected RedisClusterNode() {
+		super();
+	}
+
+	/**
+	 * Creates new {@link RedisClusterNode} with empty {@link SlotRange}.
+	 * 
+	 * @param host must not be {@literal null}.
+	 * @param port
+	 */
+	public RedisClusterNode(String host, int port) {
+		this(host, port, new SlotRange(Collections.<Integer> emptySet()));
+	}
+
+	/**
+	 * Creates new {@link RedisClusterNode} with given {@link SlotRange}.
+	 * 
+	 * @param host must not be {@literal null}.
+	 * @param port
+	 * @param slotRange can be {@literal null}.
+	 */
+	public RedisClusterNode(String host, int port, SlotRange slotRange) {
+
+		super(host, port);
+		this.slotRange = slotRange != null ? slotRange : new SlotRange(Collections.<Integer> emptySet());
+	}
+
+	/**
+	 * Get the served {@link SlotRange}.
+	 * 
+	 * @return never {@literal null}.
+	 */
+	public SlotRange getSlotRange() {
+		return slotRange;
+	}
+
+	/**
+	 * @param slot
+	 * @return true if slot is covered.
+	 */
+	public boolean servesSlot(int slot) {
+		return slotRange.contains(slot);
+	}
+
+	/**
+	 * @return
+	 */
+	public LinkState getLinkState() {
+		return linkState;
+	}
+
+	/**
+	 * @return true if node is connected to cluster.
+	 */
+	public boolean isConnected() {
+		return LinkState.CONNECTED.equals(linkState);
+	}
+
+	/**
+	 * @return never {@literal null}.
+	 */
+	public Set<Flag> getFlags() {
+		return flags == null ? Collections.<Flag> emptySet() : flags;
+	}
+
+	/**
+	 * @return true if node is marked as failing.
+	 */
+	public boolean isMarkedAsFail() {
+
+		if (!CollectionUtils.isEmpty(flags)) {
+			return flags.contains(Flag.FAIL) || flags.contains(Flag.PFAIL);
+		}
+		return false;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisNode#toString()
+	 */
+	@Override
+	public String toString() {
+		return super.toString();
+	}
+
+	/**
+	 * Get {@link RedisClusterNodeBuilder} for creating new {@link RedisClusterNode}.
+	 * 
+	 * @return never {@literal null}.
+	 */
+	public static RedisClusterNodeBuilder newRedisClusterNode() {
+		return new RedisClusterNodeBuilder();
+	}
+
+	/**
+	 * @author Christoph Strobl
+	 * @since 1.7
+	 */
+	public static class SlotRange {
+
+		private final Set<Integer> range;
+
+		/**
+		 * @param lowerBound must not be {@literal null}.
+		 * @param upperBound must not be {@literal null}.
+		 */
+		public SlotRange(Integer lowerBound, Integer upperBound) {
+
+			Assert.notNull(lowerBound, "LowerBound must not be null!");
+			Assert.notNull(upperBound, "UpperBound must not be null!");
+
+			this.range = new LinkedHashSet<Integer>();
+			for (int i = lowerBound; i <= upperBound; i++) {
+				this.range.add(i);
+			}
+		}
+
+		public SlotRange(Collection<Integer> range) {
+			this.range = CollectionUtils.isEmpty(range) ? Collections.<Integer> emptySet()
+					: new LinkedHashSet<Integer>(range);
+		}
+
+		@Override
+		public String toString() {
+			return range.toString();
+		}
+
+		/**
+		 * @param slot
+		 * @return true when slot is part of the range.
+		 */
+		public boolean contains(int slot) {
+			return range.contains(slot);
+		}
+
+		/**
+		 * @return
+		 */
+		public Set<Integer> getSlots() {
+			return Collections.unmodifiableSet(range);
+		}
+
+		public int[] getSlotsArray() {
+
+			int[] slots = new int[range.size()];
+			int pos = 0;
+
+			for (Integer value : range) {
+				slots[pos++] = value.intValue();
+			}
+
+			return slots;
+		}
+	}
+
+	/**
+	 * @author Christoph Strobl
+	 * @since 1.7
+	 */
+	public static enum LinkState {
+		CONNECTED, DISCONNECTED
+	}
+
+	/**
+	 * @author Christoph Strobl
+	 * @since 1.7
+	 */
+	public static enum Flag {
+
+		MYSELF("myself"), MASTER("master"), SLAVE("slave"), FAIL("fail"), PFAIL("fail?"), HANDSHAKE("handshake"), NOADDR(
+				"noaddr"), NOFLAGS("noflags");
+
+		private String raw;
+
+		Flag(String raw) {
+			this.raw = raw;
+		}
+
+		public String getRaw() {
+			return raw;
+		}
+
+	}
+
+	/**
+	 * Builder for creating new {@link RedisClusterNode}.
+	 * 
+	 * @author Christoph Strobl
+	 * @since 1.7
+	 */
+	public static class RedisClusterNodeBuilder extends RedisNodeBuilder {
+
+		Set<Flag> flags;
+		LinkState linkState;
+		SlotRange slotRange;
+
+		public RedisClusterNodeBuilder() {
+
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.redis.connection.RedisNode.RedisNodeBuilder#listeningAt(java.lang.String, int)
+		 */
+		@Override
+		public RedisClusterNodeBuilder listeningAt(String host, int port) {
+			super.listeningAt(host, port);
+			return this;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.redis.connection.RedisNode.RedisNodeBuilder#withName(java.lang.String)
+		 */
+		@Override
+		public RedisClusterNodeBuilder withName(String name) {
+			super.withName(name);
+			return this;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.redis.connection.RedisNode.RedisNodeBuilder#withId(java.lang.String)
+		 */
+		@Override
+		public RedisClusterNodeBuilder withId(String id) {
+			super.withId(id);
+			return this;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.redis.connection.RedisNode.RedisNodeBuilder#promotedAs(org.springframework.data.redis.connection.RedisNode.NodeType)
+		 */
+		@Override
+		public RedisClusterNodeBuilder promotedAs(NodeType nodeType) {
+			super.promotedAs(nodeType);
+			return this;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.redis.connection.RedisNode.RedisNodeBuilder#slaveOf(java.lang.String)
+		 */
+		public RedisClusterNodeBuilder slaveOf(String masterId) {
+			super.slaveOf(masterId);
+			return this;
+		}
+
+		/**
+		 * Set flags for node.
+		 * 
+		 * @param flags
+		 * @return
+		 */
+		public RedisClusterNodeBuilder withFlags(Set<Flag> flags) {
+
+			this.flags = flags;
+			return this;
+		}
+
+		/**
+		 * Set {@link SlotRange}.
+		 * 
+		 * @param range
+		 * @return
+		 */
+		public RedisClusterNodeBuilder serving(SlotRange range) {
+
+			this.slotRange = range;
+			return this;
+		}
+
+		/**
+		 * Set {@link LinkState}.
+		 * 
+		 * @param linkState
+		 * @return
+		 */
+		public RedisClusterNodeBuilder linkState(LinkState linkState) {
+			this.linkState = linkState;
+			return this;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.redis.connection.RedisNode.RedisNodeBuilder#build()
+		 */
+		@Override
+		public RedisClusterNode build() {
+
+			RedisNode base = super.build();
+
+			RedisClusterNode node = new RedisClusterNode(base.getHost(), base.getPort(), slotRange);
+			node.id = base.id;
+			node.type = base.type;
+			node.masterId = base.masterId;
+			node.name = base.name;
+			node.flags = flags;
+			node.linkState = linkState;
+			return node;
+		}
+	}
+
+}

src/main/java/org/springframework/data/redis/connection/RedisConnectionFactory.java

@@ -33,6 +33,15 @@ public interface RedisConnectionFactory extends PersistenceExceptionTranslator {
 	 */
 	RedisConnection getConnection();
 
+	/**
+	 * Provides a suitable connection for interacting with Redis Cluster.
+	 * 
+	 * @return
+	 * @throws
+	 * @since 1.7
+	 */
+	RedisClusterConnection getClusterConnection();
+
 	/**
 	 * Specifies if pipelined results should be converted to the expected data type. If false, results of
 	 * {@link RedisConnection#closePipeline()} and {RedisConnection#exec()} will be of the type returned by the underlying
@@ -45,6 +54,7 @@ public interface RedisConnectionFactory extends PersistenceExceptionTranslator {
 
 	/**
 	 * Provides a suitable connection for interacting with Redis Sentinel.
+	 * 
 	 * @return connection for interacting with Redis Sentinel.
 	 * @since 1.4
 	 */

src/main/java/org/springframework/data/redis/connection/RedisNode.java

@@ -21,14 +21,16 @@ import org.springframework.util.ObjectUtils;
 /**
  * @author Christoph Strobl
  * @author Thomas Darimont
- * 
  * @since 1.4
  */
 public class RedisNode implements NamedNode {
 
-	private String name;
-	private String host;
-	private int port;
+	String id;
+	String name;
+	String host;
+	int port;
+	NodeType type;
+	String masterId;
 
 	/**
 	 * Creates a new {@link RedisNode} with the given {@code host}, {@code port}.
@@ -37,9 +39,9 @@ public class RedisNode implements NamedNode {
 	 * @param port
 	 */
 	public RedisNode(String host, int port) {
-		
-		Assert.notNull(host,"host must not be null!");
-		
+
+		Assert.notNull(host, "host must not be null!");
+
 		this.host = host;
 		this.port = port;
 	}
@@ -67,6 +69,56 @@ public class RedisNode implements NamedNode {
 		this.name = name;
 	}
 
+	/**
+	 * @return
+	 * @since 1.7
+	 */
+	public String getMasterId() {
+		return masterId;
+	}
+
+	/**
+	 * @return
+	 * @since 1.7
+	 */
+	public String getId() {
+		return id;
+	}
+
+	/**
+	 * @return
+	 * @since 1.7
+	 */
+	public NodeType getType() {
+		return type;
+	}
+
+	/**
+	 * @return
+	 * @since 1.7
+	 */
+	public boolean isMaster() {
+		return ObjectUtils.nullSafeEquals(NodeType.MASTER, getType());
+	}
+
+	/**
+	 * @return
+	 * @since 1.7
+	 */
+	public boolean isSlave() {
+		return ObjectUtils.nullSafeEquals(NodeType.SLAVE, getType());
+	}
+
+	/**
+	 * Get {@link RedisNodeBuilder} for creating new {@link RedisNode}.
+	 * 
+	 * @return never {@literal null}.
+	 * @since 1.7
+	 */
+	public static RedisNodeBuilder newRedisNode() {
+		return new RedisNodeBuilder();
+	}
+
 	@Override
 	public String toString() {
 		return asString();
@@ -110,7 +162,16 @@ public class RedisNode implements NamedNode {
 
 	/**
 	 * @author Christoph Strobl
+	 * @since 1.7
+	 */
+	public enum NodeType {
+		MASTER, SLAVE
+	}
+
+	/**
+	 * Builder for creating new {@link RedisNode}.
 	 * 
+	 * @author Christoph Strobl
 	 * @since 1.4
 	 */
 	public static class RedisNodeBuilder {
@@ -121,11 +182,21 @@ public class RedisNode implements NamedNode {
 			node = new RedisNode();
 		}
 
+		/**
+		 * Define node name.
+		 */
 		public RedisNodeBuilder withName(String name) {
 			node.name = name;
 			return this;
 		}
 
+		/**
+		 * Set host and port of server.
+		 * 
+		 * @param host must not be {@literal null}.
+		 * @param port
+		 * @return
+		 */
 		public RedisNodeBuilder listeningAt(String host, int port) {
 
 			Assert.hasText(host, "Hostname must not be empty or null.");
@@ -134,6 +205,49 @@ public class RedisNode implements NamedNode {
 			return this;
 		}
 
+		/**
+		 * Set id of server.
+		 * 
+		 * @param id
+		 * @return
+		 */
+		public RedisNodeBuilder withId(String id) {
+
+			node.id = id;
+			return this;
+		}
+
+		/**
+		 * Set server role.
+		 * 
+		 * @param nodeType
+		 * @return
+		 * @since 1.7
+		 */
+		public RedisNodeBuilder promotedAs(NodeType type) {
+
+			node.type = type;
+			return this;
+		}
+
+		/**
+		 * Set the id of the master node.
+		 * 
+		 * @param masterId
+		 * @return
+		 * @since 1.7
+		 */
+		public RedisNodeBuilder slaveOf(String masterId) {
+
+			node.masterId = masterId;
+			return this;
+		}
+
+		/**
+		 * Get the {@link RedisNode}.
+		 * 
+		 * @return
+		 */
 		public RedisNode build() {
 			return this.node;
 		}

src/main/java/org/springframework/data/redis/connection/RedisServerCommands.java

@@ -33,6 +33,13 @@ public interface RedisServerCommands {
 		SAVE, NOSAVE;
 	}
 
+	/**
+	 * @since 1.7
+	 */
+	public enum MigrateOption {
+		COPY, REPLACE
+	}
+
 	/**
 	 * Start an {@literal Append Only File} rewrite process on server.
 	 * <p>
@@ -229,4 +236,23 @@ public interface RedisServerCommands {
 	 * @since 1.3
 	 */
 	void slaveOfNoOne();
+
+	/**
+	 * @param key must not be {@literal null}.
+	 * @param target must not be {@literal null}.
+	 * @param dbIndex
+	 * @param option can be {@literal null}. Defaulted to {@link MigrateOption#COPY}.
+	 * @since 1.7
+	 */
+	void migrate(byte[] key, RedisNode target, int dbIndex, MigrateOption option);
+
+	/**
+	 * @param key must not be {@literal null}.
+	 * @param target must not be {@literal null}.
+	 * @param dbIndex
+	 * @param option can be {@literal null}. Defaulted to {@link MigrateOption#COPY}.
+	 * @param timeout
+	 * @since 1.7
+	 */
+	void migrate(byte[] key, RedisNode target, int dbIndex, MigrateOption option, long timeout);
 }

src/main/java/org/springframework/data/redis/connection/convert/Converters.java

@@ -16,6 +16,10 @@
 package org.springframework.data.redis.connection.convert;
 
 import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
@@ -23,8 +27,16 @@ import java.util.Set;
 
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.data.redis.connection.DataType;
+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.RedisClusterNode.RedisClusterNodeBuilder;
+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.CollectionUtils;
 import org.springframework.util.NumberUtils;
+import org.springframework.util.StringUtils;
 
 /**
  * Common type converters
@@ -40,6 +52,93 @@ abstract public class Converters {
 	private static final Converter<Long, Boolean> LONG_TO_BOOLEAN = new LongToBooleanConverter();
 	private static final Converter<String, DataType> STRING_TO_DATA_TYPE = new StringToDataTypeConverter();
 	private static final Converter<Map<?, ?>, Properties> MAP_TO_PROPERTIES = MapToPropertiesConverter.INSTANCE;
+	private static final Converter<String, RedisClusterNode> STRING_TO_CLUSTER_NODE_CONVERTER;
+	private static final Map<String, Flag> flagLookupMap;
+
+	static {
+
+		flagLookupMap = new LinkedHashMap<String, RedisClusterNode.Flag>(Flag.values().length, 1);
+		for (Flag flag : Flag.values()) {
+			flagLookupMap.put(flag.getRaw(), flag);
+		}
+
+		STRING_TO_CLUSTER_NODE_CONVERTER = new Converter<String, RedisClusterNode>() {
+
+			static final int ID_INDEX = 0;
+			static final int HOST_PORT_INDEX = 1;
+			static final int FLAGS_INDEX = 2;
+			static final int MASTER_ID_INDEX = 3;
+			static final int LINK_STATE_INDEX = 7;
+			static final int SLOTS_INDEX = 8;
+
+			@Override
+			public RedisClusterNode convert(String source) {
+
+				String[] args = source.split(" ");
+				String[] hostAndPort = StringUtils.split(args[HOST_PORT_INDEX], ":");
+
+				SlotRange range = parseSlotRange(args);
+				Set<Flag> flags = parseFlags(args);
+
+				RedisClusterNodeBuilder nodeBuilder = RedisClusterNode.newRedisClusterNode()
+						.listeningAt(hostAndPort[0], Integer.valueOf(hostAndPort[1])) //
+						.withId(args[ID_INDEX]) //
+						.promotedAs(flags.contains(Flag.MASTER) ? NodeType.MASTER : NodeType.SLAVE) //
+						.serving(range) //
+						.withFlags(flags) //
+						.linkState(parseLinkState(args));
+
+				if (!args[MASTER_ID_INDEX].isEmpty() && !args[MASTER_ID_INDEX].startsWith("-")) {
+					nodeBuilder.slaveOf(args[MASTER_ID_INDEX]);
+				}
+
+				return nodeBuilder.build();
+			}
+
+			private Set<Flag> parseFlags(String[] args) {
+
+				String raw = args[FLAGS_INDEX];
+
+				Set<Flag> flags = new LinkedHashSet<RedisClusterNode.Flag>(8, 1);
+				if (StringUtils.hasText(raw)) {
+					for (String flag : raw.split(",")) {
+						flags.add(flagLookupMap.get(flag));
+					}
+				}
+				return flags;
+			}
+
+			private LinkState parseLinkState(String[] args) {
+
+				String raw = args[LINK_STATE_INDEX];
+
+				if (StringUtils.hasText(raw)) {
+					return LinkState.valueOf(raw.toUpperCase());
+				}
+				return LinkState.DISCONNECTED;
+			}
+
+			private SlotRange parseSlotRange(String[] args) {
+
+				SlotRange range = new SlotRange(Collections.<Integer> emptySet());
+				if (args.length > SLOTS_INDEX && !args[SLOTS_INDEX].startsWith("[")) {
+
+					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]));
+						}
+					} else {
+						range = new SlotRange(Integer.valueOf(raw), Integer.valueOf(raw));
+					}
+				}
+				return range;
+			}
+
+		};
+	}
 
 	public static Converter<String, Properties> stringToProps() {
 		return STRING_TO_PROPS;
@@ -73,6 +172,39 @@ abstract public class Converters {
 		return (source ? ONE : ZERO);
 	}
 
+	/**
+	 * Converts the result of a single line of {@code CLUSTER NODES} into a {@link RedisClusterNode}.
+	 * 
+	 * @param clusterNodesLine
+	 * @return
+	 * @since 1.7
+	 */
+	protected static RedisClusterNode toClusterNode(String clusterNodesLine) {
+		return STRING_TO_CLUSTER_NODE_CONVERTER.convert(clusterNodesLine);
+	}
+
+	/**
+	 * Converts the result of {@code CLUSTER NODES} into {@link RedisClusterNode}s.
+	 * 
+	 * @param clusterNodes
+	 * @return
+	 * @since 1.7
+	 */
+	public static Set<RedisClusterNode> toSetOfRedisClusterNodes(Collection<String> lines) {
+
+		if (CollectionUtils.isEmpty(lines)) {
+			return Collections.emptySet();
+		}
+
+		Set<RedisClusterNode> nodes = new LinkedHashSet<RedisClusterNode>(lines.size());
+
+		for (String line : lines) {
+			nodes.add(toClusterNode(line));
+		}
+
+		return nodes;
+	}
+
 	public static List<Object> toObjects(Set<Tuple> tuples) {
 		List<Object> tupleArgs = new ArrayList<Object>(tuples.size() * 2);
 		for (Tuple tuple : tuples) {

src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java

@@ -3538,4 +3538,41 @@ public class JedisConnection extends AbstractRedisConnection {
 		}
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisServerCommands#migrate(byte[], org.springframework.data.redis.connection.RedisNode, int, org.springframework.data.redis.connection.RedisServerCommands.MigrateOption)
+	 */
+	@Override
+	public void migrate(byte[] key, RedisNode target, int dbIndex, MigrateOption option) {
+		migrate(key, target, dbIndex, option, Long.MAX_VALUE);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisServerCommands#migrate(byte[], org.springframework.data.redis.connection.RedisNode, int, org.springframework.data.redis.connection.RedisServerCommands.MigrateOption, long)
+	 */
+	@Override
+	public void migrate(byte[] key, RedisNode target, int dbIndex, MigrateOption option, long timeout) {
+
+		final int timeoutToUse = timeout <= Integer.MAX_VALUE ? (int) timeout : Integer.MAX_VALUE;
+
+		try {
+			if (isPipelined()) {
+
+				pipeline(new JedisResult(pipeline.migrate(JedisConverters.toBytes(target.getHost()), target.getPort(), key,
+						dbIndex, timeoutToUse)));
+				return;
+			}
+			if (isQueueing()) {
+				transaction(new JedisResult(transaction.migrate(JedisConverters.toBytes(target.getHost()), target.getPort(),
+						key, dbIndex, timeoutToUse)));
+				return;
+			}
+			jedis.migrate(JedisConverters.toBytes(target.getHost()), target.getPort(), key, dbIndex, timeoutToUse);
+		} catch (Exception ex) {
+			throw convertJedisAccessException(ex);
+		}
+
+	}
+
 }

src/main/java/org/springframework/data/redis/connection/jedis/JedisConnectionFactory.java

@@ -18,18 +18,24 @@ package org.springframework.data.redis.connection.jedis;
 import java.lang.reflect.Method;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.LinkedHashSet;
 import java.util.Set;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.dao.DataAccessException;
+import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.dao.InvalidDataAccessResourceUsageException;
 import org.springframework.data.redis.ExceptionTranslationStrategy;
 import org.springframework.data.redis.PassThroughExceptionTranslationStrategy;
 import org.springframework.data.redis.RedisConnectionFailureException;
+import org.springframework.data.redis.connection.ClusterCommandExecutor;
+import org.springframework.data.redis.connection.RedisClusterConfiguration;
+import org.springframework.data.redis.connection.RedisClusterConnection;
 import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.connection.RedisNode;
@@ -40,7 +46,9 @@ import org.springframework.util.CollectionUtils;
 import org.springframework.util.ReflectionUtils;
 import org.springframework.util.StringUtils;
 
+import redis.clients.jedis.HostAndPort;
 import redis.clients.jedis.Jedis;
+import redis.clients.jedis.JedisCluster;
 import redis.clients.jedis.JedisPool;
 import redis.clients.jedis.JedisPoolConfig;
 import redis.clients.jedis.JedisSentinelPool;
@@ -93,6 +101,9 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean,
 	private int dbIndex = 0;
 	private boolean convertPipelineAndTxResults = true;
 	private RedisSentinelConfiguration sentinelConfig;
+	private RedisClusterConfiguration clusterConfig;
+	private JedisCluster cluster;
+	private ClusterCommandExecutor clusterCommandExecutor;
 
 	/**
 	 * Constructs a new <code>JedisConnectionFactory</code> instance with default settings (default connection pooling, no
@@ -116,7 +127,7 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean,
 	 * @param poolConfig pool configuration
 	 */
 	public JedisConnectionFactory(JedisPoolConfig poolConfig) {
-		this(null, poolConfig);
+		this((RedisSentinelConfiguration) null, poolConfig);
 	}
 
 	/**
@@ -143,6 +154,29 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean,
 		this.poolConfig = poolConfig != null ? poolConfig : new JedisPoolConfig();
 	}
 
+	/**
+	 * Constructs a new {@link JedisConnectionFactory} instance using the given {@link RedisClusterConfiguration} applied
+	 * to create a {@link JedisCluster}.
+	 * 
+	 * @param clusterConfig
+	 * @since 1.7
+	 */
+	public JedisConnectionFactory(RedisClusterConfiguration clusterConfig) {
+		this.clusterConfig = clusterConfig;
+	}
+
+	/**
+	 * Constructs a new {@link JedisConnectionFactory} instance using the given {@link RedisClusterConfiguration} applied
+	 * to create a {@link JedisCluster}.
+	 * 
+	 * @param clusterConfig
+	 * @since 1.7
+	 */
+	public JedisConnectionFactory(RedisClusterConfiguration clusterConfig, JedisPoolConfig poolConfig) {
+		this.clusterConfig = clusterConfig;
+		this.poolConfig = poolConfig;
+	}
+
 	/**
 	 * Returns a Jedis instance to be used as a Redis connection. The instance can be newly created or retrieved from a
 	 * pool.
@@ -151,6 +185,7 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean,
 	 */
 	protected Jedis fetchJedisConnector() {
 		try {
+
 			if (usePool && pool != null) {
 				return pool.getResource();
 			}
@@ -191,9 +226,13 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean,
 			}
 		}
 
-		if (usePool) {
+		if (usePool && clusterConfig == null) {
 			this.pool = createPool();
 		}
+
+		if (clusterConfig != null) {
+			this.cluster = createCluster();
+		}
 	}
 
 	private Pool<Jedis> createPool() {
@@ -228,6 +267,40 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean,
 				getTimeoutFrom(getShardInfo()), getShardInfo().getPassword());
 	}
 
+	private JedisCluster createCluster() {
+
+		JedisCluster cluster = createCluster(this.clusterConfig, this.poolConfig);
+		this.clusterCommandExecutor = new ClusterCommandExecutor(new JedisClusterConnection.JedisClusterTopologyProvider(
+				cluster), new JedisClusterConnection.JedisClusterNodeResourceProvider(cluster), EXCEPTION_TRANSLATION);
+		return cluster;
+	}
+
+	/**
+	 * Creates {@link JedisCluster} for given {@link RedisClusterConfiguration} and {@link GenericObjectPoolConfig}.
+	 * 
+	 * @param clusterConfig must not be {@literal null}.
+	 * @param poolConfig can be {@literal null}.
+	 * @return
+	 * @since 1.7
+	 */
+	protected JedisCluster createCluster(RedisClusterConfiguration clusterConfig, GenericObjectPoolConfig poolConfig) {
+
+		Assert.notNull("Cluster configuration must not be null!");
+
+		Set<HostAndPort> hostAndPort = new HashSet<HostAndPort>();
+		for (RedisNode node : clusterConfig.getClusterNodes()) {
+			hostAndPort.add(new HostAndPort(node.getHost(), node.getPort()));
+		}
+
+		int timeout = clusterConfig.getClusterTimeout() != null ? clusterConfig.getClusterTimeout().intValue() : 1;
+		int redirects = clusterConfig.getMaxRedirects() != null ? clusterConfig.getMaxRedirects().intValue() : 5;
+
+		if (poolConfig != null) {
+			return new JedisCluster(hostAndPort, timeout, redirects, poolConfig);
+		}
+		return new JedisCluster(hostAndPort, timeout, redirects, poolConfig);
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.beans.factory.DisposableBean#destroy()
@@ -241,13 +314,30 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean,
 			}
 			pool = null;
 		}
+		if (cluster != null) {
+			try {
+				cluster.close();
+			} catch (Exception ex) {
+				log.warn("Cannot properly close Jedis cluster", ex);
+			}
+			try {
+				clusterCommandExecutor.destroy();
+			} catch (Exception ex) {
+				log.warn("Cannot properly close cluster command executor", ex);
+			}
+		}
 	}
 
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisConnectionFactory#getConnection()
 	 */
-	public JedisConnection getConnection() {
+	public RedisConnection getConnection() {
+
+		if (cluster != null) {
+			return getClusterConnection();
+		}
+
 		Jedis jedis = fetchJedisConnector();
 		JedisConnection connection = (usePool ? new JedisConnection(jedis, pool, dbIndex) : new JedisConnection(jedis,
 				null, dbIndex));
@@ -255,6 +345,19 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean,
 		return postProcessConnection(connection);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisConnectionFactory#getClusterConnection()
+	 */
+	@Override
+	public RedisClusterConnection getClusterConnection() {
+
+		if (cluster == null) {
+			throw new InvalidDataAccessApiUsageException("Cluster is not configured!");
+		}
+		return new JedisClusterConnection(cluster, clusterCommandExecutor);
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.dao.support.PersistenceExceptionTranslator#translateExceptionIfPossible(java.lang.RuntimeException)

src/main/java/org/springframework/data/redis/connection/jedis/JedisConverters.java

@@ -17,6 +17,7 @@ package org.springframework.data.redis.connection.jedis;
 
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
@@ -25,6 +26,7 @@ import java.util.Set;
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.dao.DataAccessException;
 import org.springframework.data.redis.connection.DefaultTuple;
+import org.springframework.data.redis.connection.RedisClusterNode;
 import org.springframework.data.redis.connection.RedisListCommands.Position;
 import org.springframework.data.redis.connection.RedisServer;
 import org.springframework.data.redis.connection.RedisStringCommands.BitOperation;
@@ -67,14 +69,15 @@ abstract public class JedisConverters extends Converters {
 	private static final Converter<String[], List<RedisClientInfo>> STRING_TO_CLIENT_INFO_CONVERTER = new StringToRedisClientInfoConverter();
 	private static final Converter<redis.clients.jedis.Tuple, Tuple> TUPLE_CONVERTER;
 	private static final ListConverter<redis.clients.jedis.Tuple, Tuple> TUPLE_LIST_TO_TUPLE_LIST_CONVERTER;
+	private static final Converter<Object, RedisClusterNode> OBJECT_TO_CLUSTER_NODE_CONVERTER;
 
 	public static final byte[] PLUS_BYTES;
 	public static final byte[] MINUS_BYTES;
 	public static final byte[] POSITIVE_INFINITY_BYTES;
 	public static final byte[] NEGATIVE_INFINITY_BYTES;
-	
+
 	static {
-		
+
 		STRING_TO_BYTES = new Converter<String, byte[]>() {
 			public byte[] convert(String source) {
 				return source == null ? null : SafeEncoder.encode(source);
@@ -91,11 +94,24 @@ abstract public class JedisConverters extends Converters {
 		};
 		TUPLE_SET_TO_TUPLE_SET = new SetConverter<redis.clients.jedis.Tuple, Tuple>(TUPLE_CONVERTER);
 		TUPLE_LIST_TO_TUPLE_LIST_CONVERTER = new ListConverter<redis.clients.jedis.Tuple, Tuple>(TUPLE_CONVERTER);
-		
 		PLUS_BYTES = toBytes("+");
 		MINUS_BYTES = toBytes("-");
 		POSITIVE_INFINITY_BYTES = toBytes("+inf");
 		NEGATIVE_INFINITY_BYTES = toBytes("-inf");
+
+		OBJECT_TO_CLUSTER_NODE_CONVERTER = new Converter<Object, RedisClusterNode>() {
+
+			@Override
+			public RedisClusterNode convert(Object infos) {
+
+				List<Object> values = (List<Object>) infos;
+				RedisClusterNode.SlotRange range = new RedisClusterNode.SlotRange(((Number) values.get(0)).intValue(),
+						((Number) values.get(1)).intValue());
+				List<Object> nodeInfo = (List<Object>) values.get(2);
+				return new RedisClusterNode(JedisConverters.toString((byte[]) nodeInfo.get(0)),
+						((Number) nodeInfo.get(1)).intValue(), range);
+			}
+		};
 	}
 
 	public static Converter<String, byte[]> stringToBytes() {
@@ -169,6 +185,15 @@ abstract public class JedisConverters extends Converters {
 		return source == null ? null : SafeEncoder.encode(source);
 	}
 
+	/**
+	 * @param source
+	 * @return
+	 * @since 1.7
+	 */
+	public static RedisClusterNode toNode(Object source) {
+		return OBJECT_TO_CLUSTER_NODE_CONVERTER.convert(source);
+	}
+
 	/**
 	 * @param source
 	 * @return
@@ -200,6 +225,21 @@ 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/jedis/JedisExceptionConverter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2014 the original author or authors.
+ * Copyright 2013-2015 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.
@@ -21,26 +21,46 @@ import java.net.UnknownHostException;
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.dao.DataAccessException;
 import org.springframework.dao.InvalidDataAccessApiUsageException;
+import org.springframework.data.redis.ClusterRedirectException;
 import org.springframework.data.redis.RedisConnectionFailureException;
+import org.springframework.data.redis.TooManyClusterRedirectionsException;
 
+import redis.clients.jedis.exceptions.JedisClusterMaxRedirectionsException;
 import redis.clients.jedis.exceptions.JedisConnectionException;
 import redis.clients.jedis.exceptions.JedisDataException;
 import redis.clients.jedis.exceptions.JedisException;
+import redis.clients.jedis.exceptions.JedisRedirectionException;
 
 /**
  * Converts Exceptions thrown from Jedis to {@link DataAccessException}s
  * 
  * @author Jennifer Hickey
  * @author Thomas Darimont
+ * @author Christoph Strobl
  */
 public class JedisExceptionConverter implements Converter<Exception, DataAccessException> {
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.core.convert.converter.Converter#convert(java.lang.Object)
+	 */
 	public DataAccessException convert(Exception ex) {
 
 		if (ex instanceof DataAccessException) {
 			return (DataAccessException) ex;
 		}
 		if (ex instanceof JedisDataException) {
+
+			if (ex instanceof JedisRedirectionException) {
+				JedisRedirectionException re = (JedisRedirectionException) ex;
+				return new ClusterRedirectException(re.getSlot(), re.getTargetNode().getHost(), re.getTargetNode().getPort(),
+						ex);
+			}
+
+			if (ex instanceof JedisClusterMaxRedirectionsException) {
+				return new TooManyClusterRedirectionsException(ex.getMessage(), ex);
+			}
+
 			return new InvalidDataAccessApiUsageException(ex.getMessage(), ex);
 		}
 		if (ex instanceof JedisConnectionException) {

src/main/java/org/springframework/data/redis/connection/jredis/JredisConnection.java

@@ -41,6 +41,7 @@ import org.springframework.data.redis.connection.AbstractRedisConnection;
 import org.springframework.data.redis.connection.DataType;
 import org.springframework.data.redis.connection.MessageListener;
 import org.springframework.data.redis.connection.Pool;
+import org.springframework.data.redis.connection.RedisNode;
 import org.springframework.data.redis.connection.ReturnType;
 import org.springframework.data.redis.connection.SortParameters;
 import org.springframework.data.redis.connection.Subscription;
@@ -1370,38 +1371,84 @@ public class JredisConnection extends AbstractRedisConnection {
 		throw new UnsupportedOperationException("ZRANGEBYLEX is no supported for jredis.");
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisZSetCommands#zRevRangeByScore(byte[], org.springframework.data.redis.connection.RedisZSetCommands.Range)
+	 */
 	@Override
 	public Set<byte[]> zRevRangeByScore(byte[] key, Range range) {
 		throw new UnsupportedOperationException();
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisZSetCommands#zRevRangeByScore(byte[], org.springframework.data.redis.connection.RedisZSetCommands.Range, org.springframework.data.redis.connection.RedisZSetCommands.Limit)
+	 */
 	@Override
 	public Set<byte[]> zRevRangeByScore(byte[] key, Range range, Limit limit) {
 		throw new UnsupportedOperationException();
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisZSetCommands#zRevRangeByScoreWithScores(byte[], org.springframework.data.redis.connection.RedisZSetCommands.Range, org.springframework.data.redis.connection.RedisZSetCommands.Limit)
+	 */
 	@Override
 	public Set<Tuple> zRevRangeByScoreWithScores(byte[] key, Range range, Limit limit) {
 		throw new UnsupportedOperationException();
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisZSetCommands#zRemRangeByScore(byte[], org.springframework.data.redis.connection.RedisZSetCommands.Range)
+	 */
 	@Override
 	public Long zRemRangeByScore(byte[] key, Range range) {
 		throw new UnsupportedOperationException();
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisZSetCommands#zRangeByScore(byte[], org.springframework.data.redis.connection.RedisZSetCommands.Range)
+	 */
 	@Override
 	public Set<byte[]> zRangeByScore(byte[] key, Range range) {
 		throw new UnsupportedOperationException();
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisZSetCommands#zRangeByScore(byte[], org.springframework.data.redis.connection.RedisZSetCommands.Range, org.springframework.data.redis.connection.RedisZSetCommands.Limit)
+	 */
 	@Override
 	public Set<byte[]> zRangeByScore(byte[] key, Range range, Limit limit) {
 		throw new UnsupportedOperationException();
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisZSetCommands#zRevRangeByScoreWithScores(byte[], org.springframework.data.redis.connection.RedisZSetCommands.Range)
+	 */
 	@Override
 	public Set<Tuple> zRevRangeByScoreWithScores(byte[] key, Range range) {
 		throw new UnsupportedOperationException();
 	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisServerCommands#migrate(byte[], org.springframework.data.redis.connection.RedisNode, int, org.springframework.data.redis.connection.RedisServerCommands.MigrateOption)
+	 */
+	@Override
+	public void migrate(byte[] key, RedisNode target, int dbIndex, MigrateOption option) {
+		throw new UnsupportedOperationException();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisServerCommands#migrate(byte[], org.springframework.data.redis.connection.RedisNode, int, org.springframework.data.redis.connection.RedisServerCommands.MigrateOption, long)
+	 */
+	@Override
+	public void migrate(byte[] key, RedisNode target, int dbIndex, MigrateOption option, long timeout) {
+		throw new UnsupportedOperationException();
+	}
 }

src/main/java/org/springframework/data/redis/connection/jredis/JredisConnectionFactory.java

@@ -26,6 +26,7 @@ import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.dao.DataAccessException;
 import org.springframework.data.redis.connection.Pool;
+import org.springframework.data.redis.connection.RedisClusterConnection;
 import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.connection.RedisSentinelConnection;
@@ -107,6 +108,15 @@ public class JredisConnectionFactory implements InitializingBean, DisposableBean
 		return postProcessConnection(connection);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisConnectionFactory#getClusterConnection()
+	 */
+	@Override
+	public RedisClusterConnection getClusterConnection() {
+		throw new UnsupportedOperationException("Jredis does not support Redis Cluster.");
+	}
+
 	/**
 	 * Post process a newly retrieved connection. Useful for decorating or executing initialization commands on a new
 	 * connection. This implementation simply returns the connection.

src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java

@@ -149,7 +149,7 @@ public class LettuceConnection extends AbstractRedisConnection {
 	private boolean isPipelined = false;
 	private List<LettuceResult> ppline;
 	private Queue<FutureResult<?>> txResults = new LinkedList<FutureResult<?>>();
-	private RedisClient client;
+	private AbstractRedisClient client;
 	private volatile LettuceSubscription subscription;
 	private LettucePool pool;
 	/** flag indicating whether the connection needs to be dropped or not */
@@ -310,7 +310,7 @@ public class LettuceConnection extends AbstractRedisConnection {
 	 * @since 1.7
 	 */
 	public LettuceConnection(com.lambdaworks.redis.RedisAsyncConnection<byte[], byte[]> sharedConnection, long timeout,
-			RedisClient client, LettucePool pool, int defaultDbIndex) {
+			AbstractRedisClient client, LettucePool pool, int defaultDbIndex) {
 
 		this.asyncSharedConn = sharedConnection;
 		this.timeout = timeout;
@@ -3398,7 +3398,7 @@ public class LettuceConnection extends AbstractRedisConnection {
 	private RedisPubSubConnection<byte[], byte[]> switchToPubSub() {
 		close();
 		// open a pubsub one
-		return client.connectPubSub(CODEC);
+		return ((RedisClient) client).connectPubSub(CODEC);
 	}
 
 	private void pipeline(LettuceResult result) {
@@ -3423,7 +3423,7 @@ public class LettuceConnection extends AbstractRedisConnection {
 		return getAsyncDedicatedConnection();
 	}
 
-	private com.lambdaworks.redis.RedisConnection<byte[], byte[]> getConnection() {
+	protected com.lambdaworks.redis.RedisConnection<byte[], byte[]> getConnection() {
 		if (isQueueing()) {
 			return getDedicatedConnection();
 		}
@@ -3433,19 +3433,28 @@ public class LettuceConnection extends AbstractRedisConnection {
 		return getDedicatedConnection();
 	}
 
-	private RedisAsyncConnection<byte[], byte[]> getAsyncDedicatedConnection() {
+	protected RedisAsyncConnection<byte[], byte[]> getAsyncDedicatedConnection() {
 		if (asyncDedicatedConn == null) {
-			if (this.pool != null) {
-				this.asyncDedicatedConn = pool.getResource();
-			} else {
 
-				this.asyncDedicatedConn = client.connectAsync(CODEC);
+			asyncDedicatedConn = doGetAsyncDedicatedConnection();
+
+			if (this.pool == null) {
 				this.asyncDedicatedConn.select(dbIndex);
 			}
+
 		}
 		return asyncDedicatedConn;
 	}
 
+	protected RedisAsyncConnection<byte[], byte[]> doGetAsyncDedicatedConnection() {
+
+		if (this.pool != null) {
+			return pool.getResource();
+		} else {
+			return ((RedisClient) client).connectAsync(CODEC);
+		}
+	}
+
 	private com.lambdaworks.redis.RedisConnection<byte[], byte[]> getDedicatedConnection() {
 		if (dedicatedConn == null) {
 			this.dedicatedConn = syncConnection(getAsyncDedicatedConnection());
@@ -3581,7 +3590,7 @@ public class LettuceConnection extends AbstractRedisConnection {
 
 		RedisConnection<String, String> connection = null;
 		try {
-			connection = client.connect(getRedisURI(node));
+			connection = ((RedisClient) client).connect(getRedisURI(node));
 			return connection.ping().equalsIgnoreCase("pong");
 		} catch (Exception e) {
 			return false;
@@ -3602,7 +3611,8 @@ public class LettuceConnection extends AbstractRedisConnection {
 	 */
 	@Override
 	protected RedisSentinelConnection getSentinelConnection(RedisNode sentinel) {
-		RedisSentinelAsyncConnection<String, String> connection = client.connectSentinelAsync(getRedisURI(sentinel));
+		RedisSentinelAsyncConnection<String, String> connection = ((RedisClient) client)
+				.connectSentinelAsync(getRedisURI(sentinel));
 		return new LettuceSentinelConnection(connection);
 	}
 
@@ -4025,4 +4035,37 @@ public class LettuceConnection extends AbstractRedisConnection {
 		}
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisServerCommands#migrate(byte[], org.springframework.data.redis.connection.RedisNode, int, org.springframework.data.redis.connection.RedisServerCommands.MigrateOption)
+	 */
+	@Override
+	public void migrate(byte[] key, RedisNode target, int dbIndex, MigrateOption option) {
+		migrate(key, target, dbIndex, option, Long.MAX_VALUE);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisServerCommands#migrate(byte[], org.springframework.data.redis.connection.RedisNode, int, org.springframework.data.redis.connection.RedisServerCommands.MigrateOption, long)
+	 */
+	@Override
+	public void migrate(byte[] key, RedisNode target, int dbIndex, MigrateOption option, long timeout) {
+
+		try {
+			if (isPipelined()) {
+				pipeline(new LettuceResult(getAsyncConnection().migrate(target.getHost(), target.getPort(), key, dbIndex,
+						timeout)));
+				return;
+			}
+			if (isQueueing()) {
+				transaction(new LettuceTxResult(getConnection().migrate(target.getHost(), target.getPort(), key, dbIndex,
+						timeout)));
+				return;
+			}
+			getConnection().migrate(target.getHost(), target.getPort(), key, dbIndex, timeout);
+		} catch (Exception ex) {
+			throw convertLettuceAccessException(ex);
+		}
+	}
+
 }

src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java

@@ -16,6 +16,8 @@
 
 package org.springframework.data.redis.connection.lettuce;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.commons.logging.Log;
@@ -23,22 +25,30 @@ import org.apache.commons.logging.LogFactory;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.dao.DataAccessException;
+import org.springframework.dao.InvalidDataAccessApiUsageException;
+import org.springframework.dao.InvalidDataAccessResourceUsageException;
 import org.springframework.data.redis.ExceptionTranslationStrategy;
 import org.springframework.data.redis.PassThroughExceptionTranslationStrategy;
 import org.springframework.data.redis.RedisConnectionFailureException;
+import org.springframework.data.redis.connection.ClusterCommandExecutor;
 import org.springframework.data.redis.connection.Pool;
+import org.springframework.data.redis.connection.RedisClusterConfiguration;
+import org.springframework.data.redis.connection.RedisClusterConnection;
 import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.connection.RedisNode;
 import org.springframework.data.redis.connection.RedisSentinelConfiguration;
 import org.springframework.data.redis.connection.RedisSentinelConnection;
 import org.springframework.util.Assert;
 
+import com.lambdaworks.redis.AbstractRedisClient;
 import com.lambdaworks.redis.LettuceFutures;
 import com.lambdaworks.redis.RedisAsyncConnection;
 import com.lambdaworks.redis.RedisClient;
 import com.lambdaworks.redis.RedisException;
 import com.lambdaworks.redis.RedisFuture;
 import com.lambdaworks.redis.RedisURI;
+import com.lambdaworks.redis.cluster.RedisClusterClient;
 
 /**
  * Connection factory creating <a href="http://github.com/mp911de/lettuce">Lettuce</a>-based connections.
@@ -68,7 +78,7 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea
 
 	private String hostName = "localhost";
 	private int port = 6379;
-	private RedisClient client;
+	private AbstractRedisClient client;
 	private long timeout = TimeUnit.MILLISECONDS.convert(60, TimeUnit.SECONDS);
 	private long shutdownTimeout = TimeUnit.MILLISECONDS.convert(2, TimeUnit.SECONDS);
 	private boolean validateConnection = false;
@@ -81,6 +91,8 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea
 	private String password;
 	private boolean convertPipelineAndTxResults = true;
 	private RedisSentinelConfiguration sentinelConfiguration;
+	private RedisClusterConfiguration clusterConfiguration;
+	private ClusterCommandExecutor clusterCommandExecutor;
 
 	/**
 	 * Constructs a new <code>LettuceConnectionFactory</code> instance with default settings.
@@ -105,17 +117,46 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea
 		this.sentinelConfiguration = sentinelConfiguration;
 	}
 
+	/**
+	 * Constructs a new {@link LettuceConnectionFactory} instance using the given {@link RedisClusterConfiguration}
+	 * applied to create a {@link RedisClusterClient}.
+	 * 
+	 * @param clusterConfig
+	 * @since 1.7
+	 */
+	public LettuceConnectionFactory(RedisClusterConfiguration clusterConfig) {
+		this.clusterConfiguration = clusterConfig;
+	}
+
 	public LettuceConnectionFactory(LettucePool pool) {
 		this.pool = pool;
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
+	 */
 	public void afterPropertiesSet() {
 		this.client = createRedisClient();
+
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.beans.factory.DisposableBean#destroy()
+	 */
 	public void destroy() {
 		resetConnection();
 		client.shutdown(shutdownTimeout, shutdownTimeout, TimeUnit.MILLISECONDS);
+
+		if (clusterCommandExecutor != null) {
+
+			try {
+				clusterCommandExecutor.destroy();
+			} catch (Exception ex) {
+				log.warn("Cannot properly close cluster command executor", ex);
+			}
+		}
 	}
 
 	/*
@@ -124,11 +165,29 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea
 	 */
 	public RedisConnection getConnection() {
 
+		if (isClusterAware()) {
+			return getClusterConnection();
+		}
+
 		LettuceConnection connection = new LettuceConnection(getSharedConnection(), timeout, client, pool, dbIndex);
 		connection.setConvertPipelineAndTxResults(convertPipelineAndTxResults);
 		return connection;
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisConnectionFactory#getClusterConnection()
+	 */
+	@Override
+	public RedisClusterConnection getClusterConnection() {
+
+		if (!isClusterAware()) {
+			throw new InvalidDataAccessApiUsageException("Cluster is not configured!");
+		}
+
+		return new LettuceClusterConnection((RedisClusterClient) client, clusterCommandExecutor);
+	}
+
 	public void initConnection() {
 		synchronized (this.connectionMonitor) {
 			if (this.connection != null) {
@@ -377,9 +436,16 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea
 
 	protected RedisAsyncConnection<byte[], byte[]> createLettuceConnector() {
 		try {
-			RedisAsyncConnection<byte[], byte[]> connection = client.connectAsync(LettuceConnection.CODEC);
-			if (dbIndex > 0) {
-				connection.select(dbIndex);
+
+			RedisAsyncConnection connection = null;
+			if (client instanceof RedisClient) {
+				connection = ((RedisClient) client).connectAsync(LettuceConnection.CODEC);
+				if (dbIndex > 0) {
+					connection.select(dbIndex);
+				}
+			} else {
+				connection = (RedisAsyncConnection<byte[], byte[]>) ((RedisClusterClient) client)
+						.connectClusterAsync(LettuceConnection.CODEC);
 			}
 			return connection;
 		} catch (RedisException e) {
@@ -387,13 +453,29 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea
 		}
 	}
 
-	private RedisClient createRedisClient() {
+	private AbstractRedisClient createRedisClient() {
 
 		if (isRedisSentinelAware()) {
 			RedisURI redisURI = getSentinelRedisURI();
 			return new RedisClient(redisURI);
 		}
 
+		if (isClusterAware()) {
+
+			List<RedisURI> initialUris = new ArrayList<RedisURI>();
+			for (RedisNode node : this.clusterConfiguration.getClusterNodes()) {
+				initialUris.add(new RedisURI(node.getHost(), node.getPort(), this.timeout, TimeUnit.MILLISECONDS));
+			}
+
+			RedisClusterClient clusterClient = new RedisClusterClient(initialUris);
+
+			this.clusterCommandExecutor = new ClusterCommandExecutor(
+					new LettuceClusterConnection.LettuceClusterTopologyProvider(clusterClient),
+					new LettuceClusterConnection.LettuceClusterNodeResourceProvider(clusterClient), EXCEPTION_TRANSLATION);
+
+			return clusterClient;
+		}
+
 		if (pool != null) {
 			return pool.getClient();
 		}
@@ -419,8 +501,18 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea
 		return sentinelConfiguration != null;
 	}
 
+	/**
+	 * @return since 1.7
+	 */
+	public boolean isClusterAware() {
+		return clusterConfiguration != null;
+	}
+
 	@Override
 	public RedisSentinelConnection getSentinelConnection() {
-		return new LettuceSentinelConnection(client.connectSentinelAsync());
+		if (!(client instanceof RedisClient)) {
+			throw new InvalidDataAccessResourceUsageException("Unable to connect to senitels using " + client.getClass());
+		}
+		return new LettuceSentinelConnection(((RedisClient) client).connectSentinelAsync());
 	}
 }

src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConverters.java

@@ -30,8 +30,13 @@ import java.util.Set;
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.dao.DataAccessException;
 import org.springframework.data.redis.connection.DefaultTuple;
+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.RedisClusterNode.SlotRange;
 import org.springframework.data.redis.connection.RedisListCommands.Position;
 import org.springframework.data.redis.connection.RedisNode;
+import org.springframework.data.redis.connection.RedisNode.NodeType;
 import org.springframework.data.redis.connection.RedisSentinelConfiguration;
 import org.springframework.data.redis.connection.RedisServer;
 import org.springframework.data.redis.connection.RedisZSetCommands.Range.Boundary;
@@ -52,6 +57,8 @@ import com.lambdaworks.redis.RedisURI;
 import com.lambdaworks.redis.ScoredValue;
 import com.lambdaworks.redis.ScriptOutputType;
 import com.lambdaworks.redis.SortArgs;
+import com.lambdaworks.redis.cluster.models.partitions.Partitions;
+import com.lambdaworks.redis.cluster.models.partitions.RedisClusterNode.NodeFlag;
 import com.lambdaworks.redis.protocol.LettuceCharsets;
 
 /**
@@ -78,6 +85,8 @@ abstract public class LettuceConverters extends Converters {
 	private static final Converter<List<byte[]>, Map<byte[], byte[]>> BYTES_LIST_TO_MAP;
 	private static final Converter<List<byte[]>, List<Tuple>> BYTES_LIST_TO_TUPLE_LIST_CONVERTER;
 	private static final Converter<String[], List<RedisClientInfo>> STRING_TO_LIST_OF_CLIENT_INFO = new StringToRedisClientInfoConverter();
+	private static final Converter<Partitions, List<RedisClusterNode>> PARTITIONS_TO_CLUSTER_NODES;
+	private static Converter<com.lambdaworks.redis.cluster.models.partitions.RedisClusterNode, RedisClusterNode> CLUSTER_NODE_TO_CLUSTER_NODE_CONVERTER;
 
 	public static final byte[] PLUS_BYTES;
 	public static final byte[] MINUS_BYTES;
@@ -199,6 +208,74 @@ abstract public class LettuceConverters extends Converters {
 			}
 		};
 
+		PARTITIONS_TO_CLUSTER_NODES = new Converter<Partitions, List<RedisClusterNode>>() {
+
+			@Override
+			public List<RedisClusterNode> convert(Partitions source) {
+
+				if (source == null) {
+					return Collections.emptyList();
+				}
+				List<RedisClusterNode> nodes = new ArrayList<RedisClusterNode>();
+				for (com.lambdaworks.redis.cluster.models.partitions.RedisClusterNode node : source.getPartitions()) {
+					nodes.add(CLUSTER_NODE_TO_CLUSTER_NODE_CONVERTER.convert(node));
+				}
+
+				return nodes;
+			};
+
+		};
+
+		CLUSTER_NODE_TO_CLUSTER_NODE_CONVERTER = new Converter<com.lambdaworks.redis.cluster.models.partitions.RedisClusterNode, RedisClusterNode>() {
+
+			@Override
+			public RedisClusterNode convert(com.lambdaworks.redis.cluster.models.partitions.RedisClusterNode source) {
+
+				Set<Flag> flags = parseFlags(source.getFlags());
+
+				return RedisClusterNode.newRedisClusterNode().listeningAt(source.getUri().getHost(), source.getUri().getPort())
+						.withId(source.getNodeId()).promotedAs(flags.contains(Flag.MASTER) ? NodeType.MASTER : NodeType.SLAVE)
+						.serving(new SlotRange(source.getSlots())).withFlags(flags)
+						.linkState(source.isConnected() ? LinkState.CONNECTED : LinkState.DISCONNECTED)
+						.slaveOf(source.getSlaveOf()).build();
+			}
+
+			private Set<Flag> parseFlags(Set<NodeFlag> source) {
+
+				Set<Flag> flags = new LinkedHashSet<Flag>(source != null ? source.size() : 8, 1);
+				for (NodeFlag flag : source) {
+					switch (flag) {
+						case NOFLAGS:
+							flags.add(Flag.NOFLAGS);
+							break;
+						case EVENTUAL_FAIL:
+							flags.add(Flag.PFAIL);
+							break;
+						case FAIL:
+							flags.add(Flag.FAIL);
+							break;
+						case HANDSHAKE:
+							flags.add(Flag.HANDSHAKE);
+							break;
+						case MASTER:
+							flags.add(Flag.MASTER);
+							break;
+						case MYSELF:
+							flags.add(Flag.MYSELF);
+							break;
+						case NOADDR:
+							flags.add(Flag.NOADDR);
+							break;
+						case SLAVE:
+							flags.add(Flag.SLAVE);
+							break;
+					}
+				}
+				return flags;
+			}
+
+		};
+
 		PLUS_BYTES = toBytes("+");
 		MINUS_BYTES = toBytes("-");
 		POSITIVE_INFINITY_BYTES = toBytes("+inf");
@@ -523,4 +600,18 @@ abstract public class LettuceConverters extends Converters {
 		return toString(buffer.array());
 	}
 
+	public static List<RedisClusterNode> partitionsToClusterNodes(Partitions partitions) {
+		return PARTITIONS_TO_CLUSTER_NODES.convert(partitions);
+	}
+
+	/**
+	 * @param source
+	 * @return
+	 * @since 1.7
+	 */
+	public static RedisClusterNode toRedisClusterNode(
+			com.lambdaworks.redis.cluster.models.partitions.RedisClusterNode source) {
+		return CLUSTER_NODE_TO_CLUSTER_NODE_CONVERTER.convert(source);
+	}
+
 }

src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java

@@ -38,6 +38,7 @@ import org.springframework.data.redis.connection.AbstractRedisConnection;
 import org.springframework.data.redis.connection.DataType;
 import org.springframework.data.redis.connection.FutureResult;
 import org.springframework.data.redis.connection.MessageListener;
+import org.springframework.data.redis.connection.RedisNode;
 import org.springframework.data.redis.connection.RedisPipelineException;
 import org.springframework.data.redis.connection.RedisSubscribedConnectionException;
 import org.springframework.data.redis.connection.ReturnType;
@@ -2623,4 +2624,22 @@ public class SrpConnection extends AbstractRedisConnection {
 		throw new UnsupportedOperationException("ZRANGEBYLEX is no supported for srp.");
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisServerCommands#migrate(byte[], org.springframework.data.redis.connection.RedisNode, int, org.springframework.data.redis.connection.RedisServerCommands.MigrateOption)
+	 */
+	@Override
+	public void migrate(byte[] key, RedisNode target, int dbIndex, MigrateOption option) {
+		throw new UnsupportedOperationException();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisServerCommands#migrate(byte[], org.springframework.data.redis.connection.RedisNode, int, org.springframework.data.redis.connection.RedisServerCommands.MigrateOption, long)
+	 */
+	@Override
+	public void migrate(byte[] key, RedisNode target, int dbIndex, MigrateOption option, long timeout) {
+		throw new UnsupportedOperationException();
+	}
+
 }

src/main/java/org/springframework/data/redis/connection/srp/SrpConnectionFactory.java

@@ -24,6 +24,7 @@ import org.springframework.beans.factory.InitializingBean;
 import org.springframework.dao.DataAccessException;
 import org.springframework.data.redis.ExceptionTranslationStrategy;
 import org.springframework.data.redis.PassThroughExceptionTranslationStrategy;
+import org.springframework.data.redis.connection.RedisClusterConnection;
 import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.connection.RedisSentinelConnection;
@@ -83,6 +84,15 @@ public class SrpConnectionFactory implements InitializingBean, DisposableBean, R
 		return connection;
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisConnectionFactory#getClusterConnection()
+	 */
+	@Override
+	public RedisClusterConnection getClusterConnection() {
+		throw new UnsupportedOperationException("Srp does not support Redis Cluster.");
+	}
+
 	public DataAccessException translateExceptionIfPossible(RuntimeException ex) {
 		return EXCEPTION_TRANSLATION.translate(ex);
 	}

src/main/java/org/springframework/data/redis/connection/util/ByteArraySet.java

@@ -0,0 +1,141 @@
+/*
+ * Copyright 2015 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.connection.util;
+
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.LinkedHashSet;
+import java.util.Set;
+
+public class ByteArraySet implements Set<ByteArrayWrapper> {
+
+	LinkedHashSet<ByteArrayWrapper> delegate;
+
+	public ByteArraySet() {
+		this.delegate = new LinkedHashSet<ByteArrayWrapper>();
+	}
+
+	public ByteArraySet(Collection<byte[]> values) {
+		this();
+		addAll(values);
+	}
+
+	public int size() {
+		return delegate.size();
+	}
+
+	public boolean contains(Object o) {
+
+		if (o instanceof byte[]) {
+			return delegate.contains(new ByteArrayWrapper((byte[]) o));
+		}
+		return delegate.contains(o);
+	}
+
+	public boolean add(ByteArrayWrapper e) {
+		return delegate.add(e);
+	}
+
+	public boolean add(byte[] e) {
+		return delegate.add(new ByteArrayWrapper(e));
+	}
+
+	public boolean containsAll(Collection<?> c) {
+
+		for (Object o : c) {
+			if (o instanceof byte[]) {
+				if (!contains(new ByteArrayWrapper((byte[]) o))) {
+					return false;
+				}
+			} else {
+				if (!contains(o)) {
+					return false;
+				}
+			}
+		}
+		return true;
+	}
+
+	public boolean addAll(Collection<? extends ByteArrayWrapper> c) {
+		return delegate.addAll(c);
+	}
+
+	public boolean addAll(Iterable<byte[]> c) {
+
+		for (byte[] o : c) {
+			add(o);
+		}
+		return true;
+	}
+
+	@Override
+	public boolean isEmpty() {
+		return delegate.isEmpty();
+	}
+
+	@Override
+	public Iterator<ByteArrayWrapper> iterator() {
+		return delegate.iterator();
+	}
+
+	@Override
+	public Object[] toArray() {
+		return delegate.toArray();
+	}
+
+	@Override
+	public <T> T[] toArray(T[] a) {
+		return delegate.toArray(a);
+	}
+
+	@Override
+	public boolean remove(Object o) {
+
+		if (o instanceof byte[]) {
+			delegate.remove(new ByteArrayWrapper((byte[]) o));
+		}
+		return delegate.remove(o);
+	}
+
+	@Override
+	public boolean retainAll(Collection<?> c) {
+		return delegate.retainAll(c);
+	}
+
+	@Override
+	public boolean removeAll(Collection<?> c) {
+
+		for (Object o : c) {
+			remove(o);
+		}
+		return true;
+	}
+
+	@Override
+	public void clear() {
+		delegate.clear();
+	}
+
+	public Set<byte[]> asRawSet() {
+
+		Set<byte[]> result = new LinkedHashSet<byte[]>();
+		for (ByteArrayWrapper wrapper : delegate) {
+			result.add(wrapper.getArray());
+		}
+		return result;
+	}
+
+}

src/main/java/org/springframework/data/redis/core/AbstractOperations.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2014 the original author or authors.
+ * Copyright 2011-2015 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.
@@ -16,6 +16,7 @@
 package org.springframework.data.redis.core;
 
 import java.util.Collection;
+import java.util.Collections;
 import java.util.LinkedHashMap;
 import java.util.LinkedHashSet;
 import java.util.List;
@@ -29,6 +30,7 @@ import org.springframework.data.redis.core.ZSetOperations.TypedTuple;
 import org.springframework.data.redis.serializer.RedisSerializer;
 import org.springframework.data.redis.serializer.SerializationUtils;
 import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
 
 /**
  * Internal base class used by various RedisTemplate XXXOperations implementations.
@@ -288,6 +290,23 @@ abstract class AbstractOperations<K, V> {
 		return (K) keySerializer().deserialize(value);
 	}
 
+	/**
+	 * @param keys
+	 * @return
+	 * @since 1.7
+	 */
+	Set<K> deserializeKeys(Set<byte[]> keys) {
+
+		if (CollectionUtils.isEmpty(keys)) {
+			return Collections.emptySet();
+		}
+		Set<K> result = new LinkedHashSet<K>(keys.size());
+		for (byte[] key : keys) {
+			result.add(deserializeKey(key));
+		}
+		return result;
+	}
+
 	@SuppressWarnings("unchecked")
 	V deserializeValue(byte[] value) {
 		if (valueSerializer() == null) {

src/main/java/org/springframework/data/redis/core/ClusterOperations.java

@@ -0,0 +1,145 @@
+/*
+ * Copyright 2015 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.core;
+
+import java.util.Collection;
+import java.util.Set;
+
+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.
+ * 
+ * @author Christoph Strobl
+ * @since 1.7
+ */
+public interface ClusterOperations<K, V> {
+
+	/**
+	 * Get all keys located at given node.
+	 * 
+	 * @param node must not be {@literal null}.
+	 * @param pattern
+	 * @return never {@literal null}.
+	 * @see RedisConnection#keys(byte[])
+	 */
+	Set<K> keys(RedisClusterNode node, K pattern);
+
+	/**
+	 * Ping the given node;
+	 * 
+	 * @param node must not be {@literal null}.
+	 * @return
+	 * @see RedisConnection#ping()
+	 */
+	String ping(RedisClusterNode node);
+
+	/**
+	 * Get a random key from the range served by the given node.
+	 * 
+	 * @param node must not be {@literal null}.
+	 * @return
+	 * @see RedisConnection#randomKey()
+	 */
+	K randomKey(RedisClusterNode node);
+
+	/**
+	 * Add slots to given node;
+	 * 
+	 * @param node must not be {@literal null}.
+	 * @param slots must not be {@literal null}.
+	 */
+	void addSlots(RedisClusterNode node, int... slots);
+
+	/**
+	 * Add slots in {@link SlotRange} to given node.
+	 * 
+	 * @param node must not be {@literal null}.
+	 * @param range must not be {@literal null}.
+	 */
+	void addSlots(RedisClusterNode node, SlotRange range);
+
+	/**
+	 * tart an {@literal Append Only File} rewrite process on given node.
+	 * 
+	 * @param node must not be {@literal null}.
+	 * @see RedisConnection#bgReWriteAof()
+	 */
+	void bgReWriteAof(RedisClusterNode node);
+
+	/**
+	 * Start background saving of db on given node.
+	 * 
+	 * @param node must not be {@literal null}.
+	 * @see RedisConnection#bgSave()
+	 */
+	void bgSave(RedisClusterNode node);
+
+	/**
+	 * Add the node to cluster.
+	 * 
+	 * @param node must not be {@literal null}.
+	 */
+	void meet(RedisClusterNode node);
+
+	/**
+	 * Remove the node from the cluster.
+	 * 
+	 * @param node must not be {@literal null}.
+	 */
+	void forget(RedisClusterNode node);
+
+	/**
+	 * Flush db on node.
+	 * 
+	 * @param node must not be {@literal null}.
+	 * @see RedisConnection#flushDb()
+	 */
+	void flushDb(RedisClusterNode node);
+
+	/**
+	 * @param node must not be {@literal null}.
+	 * @return
+	 */
+	Collection<RedisClusterNode> getSlaves(RedisClusterNode node);
+
+	/**
+	 * Synchronous save current db snapshot on server.
+	 * 
+	 * @param node must not be {@literal null}.
+	 * @see RedisConnection#save()
+	 */
+	void save(RedisClusterNode node);
+
+	/**
+	 * Shutdown given node.
+	 * 
+	 * @param node must not be {@literal null}.
+	 * @see RedisConnection#shutdown()
+	 */
+	void shutdown(RedisClusterNode node);
+
+	/**
+	 * Move slot assignment from one source to target node and copy keys associated with the slot.
+	 * 
+	 * @param source must not be {@literal null}.
+	 * @param slot
+	 * @param target must not be {@literal null}.
+	 */
+	void reshard(RedisClusterNode source, int slot, RedisClusterNode target);
+}

src/main/java/org/springframework/data/redis/core/DefaultClusterOperations.java

@@ -0,0 +1,336 @@
+/*
+ * Copyright 2015 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.core;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Set;
+
+import org.springframework.dao.DataAccessException;
+import org.springframework.data.redis.connection.RedisClusterCommands.AddSlots;
+import org.springframework.data.redis.connection.RedisClusterConnection;
+import org.springframework.data.redis.connection.RedisClusterNode;
+import org.springframework.data.redis.connection.RedisClusterNode.SlotRange;
+import org.springframework.data.redis.connection.RedisServerCommands.MigrateOption;
+import org.springframework.util.Assert;
+
+/**
+ * Default {@link ClusterOperations} implementation.
+ * 
+ * @author Christoph Strobl
+ * @since 1.7
+ * @param <K>
+ * @param <V>
+ */
+public class DefaultClusterOperations<K, V> extends AbstractOperations<K, V> implements ClusterOperations<K, V> {
+
+	private final RedisTemplate<K, V> template;
+
+	/**
+	 * Creates new {@link DefaultClusterOperations} delegating to the given {@link RedisTemplate}.
+	 * 
+	 * @param template must not be {@literal null}.
+	 */
+	public DefaultClusterOperations(RedisTemplate<K, V> template) {
+
+		super(template);
+		this.template = template;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.RedisClusterOperations#keys(org.springframework.data.redis.connection.RedisNode, byte[])
+	 */
+	@Override
+	public Set<K> keys(final RedisClusterNode node, final K pattern) {
+
+		Assert.notNull(node, "ClusterNode must not be null.");
+
+		return execute(new RedisClusterCallback<Set<K>>() {
+
+			@Override
+			public Set<K> doInRedis(RedisClusterConnection connection) throws DataAccessException {
+				return deserializeKeys(connection.keys(node, rawKey(pattern)));
+			}
+		});
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.RedisClusterOperations#randomKey(org.springframework.data.redis.connection.RedisNode)
+	 */
+	@Override
+	public K randomKey(final RedisClusterNode node) {
+
+		Assert.notNull(node, "ClusterNode must not be null.");
+
+		return execute(new RedisClusterCallback<K>() {
+
+			@Override
+			public K doInRedis(RedisClusterConnection connection) throws DataAccessException {
+				return deserializeKey(connection.randomKey(node));
+			}
+		});
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.RedisClusterOperations#ping(org.springframework.data.redis.connection.RedisNode)
+	 */
+	@Override
+	public String ping(final RedisClusterNode node) {
+
+		Assert.notNull(node, "ClusterNode must not be null.");
+
+		return execute(new RedisClusterCallback<String>() {
+
+			@Override
+			public String doInRedis(RedisClusterConnection connection) throws DataAccessException {
+				return connection.ping(node);
+			}
+		});
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.RedisClusterOperations#addSlots(org.springframework.data.redis.connection.RedisClusterNode, int[])
+	 */
+	@Override
+	public void addSlots(final RedisClusterNode node, final int... slots) {
+
+		Assert.notNull(node, "ClusterNode must not be null.");
+
+		execute(new RedisClusterCallback<Void>() {
+
+			@Override
+			public Void doInRedis(RedisClusterConnection connection) throws DataAccessException {
+				connection.clusterAddSlots(node, slots);
+				return null;
+			}
+		});
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.RedisClusterOperations#addSlots(org.springframework.data.redis.connection.RedisClusterNode, org.springframework.data.redis.connection.RedisClusterNode.SlotRange)
+	 */
+	@Override
+	public void addSlots(RedisClusterNode node, SlotRange range) {
+
+		Assert.notNull(node, "ClusterNode must not be null.");
+		Assert.notNull(range, "Range must not be null.");
+
+		addSlots(node, range.getSlotsArray());
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.RedisClusterOperations#bgReWriteAof(org.springframework.data.redis.connection.RedisClusterNode)
+	 */
+	@Override
+	public void bgReWriteAof(final RedisClusterNode node) {
+
+		Assert.notNull(node, "ClusterNode must not be null.");
+
+		execute(new RedisClusterCallback<Void>() {
+
+			@Override
+			public Void doInRedis(RedisClusterConnection connection) throws DataAccessException {
+				connection.bgReWriteAof(node);
+				return null;
+			}
+		});
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.RedisClusterOperations#bgSave(org.springframework.data.redis.connection.RedisClusterNode)
+	 */
+	@Override
+	public void bgSave(final RedisClusterNode node) {
+
+		Assert.notNull(node, "ClusterNode must not be null.");
+
+		execute(new RedisClusterCallback<Void>() {
+
+			@Override
+			public Void doInRedis(RedisClusterConnection connection) throws DataAccessException {
+				connection.bgSave(node);
+				return null;
+			}
+		});
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.RedisClusterOperations#meet(org.springframework.data.redis.connection.RedisClusterNode)
+	 */
+	@Override
+	public void meet(final RedisClusterNode node) {
+
+		Assert.notNull(node, "ClusterNode must not be null.");
+
+		execute(new RedisClusterCallback<Void>() {
+
+			@Override
+			public Void doInRedis(RedisClusterConnection connection) throws DataAccessException {
+				connection.clusterMeet(node);
+				return null;
+			}
+		});
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.RedisClusterOperations#forget(org.springframework.data.redis.connection.RedisClusterNode)
+	 */
+	@Override
+	public void forget(final RedisClusterNode node) {
+
+		Assert.notNull(node, "ClusterNode must not be null.");
+
+		execute(new RedisClusterCallback<Void>() {
+
+			@Override
+			public Void doInRedis(RedisClusterConnection connection) throws DataAccessException {
+				connection.clusterForget(node);
+				return null;
+			}
+		});
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.RedisClusterOperations#flushDb(org.springframework.data.redis.connection.RedisClusterNode)
+	 */
+	@Override
+	public void flushDb(final RedisClusterNode node) {
+
+		Assert.notNull(node, "ClusterNode must not be null.");
+
+		execute(new RedisClusterCallback<Void>() {
+
+			@Override
+			public Void doInRedis(RedisClusterConnection connection) throws DataAccessException {
+				connection.flushDb(node);
+				return null;
+			}
+		});
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.RedisClusterOperations#getSlaves(org.springframework.data.redis.connection.RedisClusterNode)
+	 */
+	@Override
+	public Collection<RedisClusterNode> getSlaves(final RedisClusterNode node) {
+
+		Assert.notNull(node, "ClusterNode must not be null.");
+
+		return execute(new RedisClusterCallback<Collection<RedisClusterNode>>() {
+
+			@Override
+			public Collection<RedisClusterNode> doInRedis(RedisClusterConnection connection) throws DataAccessException {
+				return connection.clusterGetSlaves(node);
+			}
+		});
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.RedisClusterOperations#save(org.springframework.data.redis.connection.RedisClusterNode)
+	 */
+	@Override
+	public void save(final RedisClusterNode node) {
+
+		Assert.notNull(node, "ClusterNode must not be null.");
+
+		execute(new RedisClusterCallback<Void>() {
+
+			@Override
+			public Void doInRedis(RedisClusterConnection connection) throws DataAccessException {
+				connection.save(node);
+				return null;
+			}
+		});
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.RedisClusterOperations#shutdown(org.springframework.data.redis.connection.RedisClusterNode)
+	 */
+	@Override
+	public void shutdown(final RedisClusterNode node) {
+
+		Assert.notNull(node, "ClusterNode must not be null.");
+
+		execute(new RedisClusterCallback<Void>() {
+
+			@Override
+			public Void doInRedis(RedisClusterConnection connection) throws DataAccessException {
+				connection.shutdown(node);
+				return null;
+			}
+		});
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.ClusterOperations#reshard(org.springframework.data.redis.connection.RedisClusterNode, int, org.springframework.data.redis.connection.RedisClusterNode)
+	 */
+	@Override
+	public void reshard(final RedisClusterNode source, final int slot, final RedisClusterNode target) {
+
+		Assert.notNull(source, "Source node must not be null.");
+		Assert.notNull(target, "Target node must not be null.");
+
+		execute(new RedisClusterCallback<Void>() {
+
+			@Override
+			public Void doInRedis(RedisClusterConnection connection) throws DataAccessException {
+
+				connection.clusterSetSlot(target, slot, AddSlots.IMPORTING);
+				connection.clusterSetSlot(source, slot, AddSlots.MIGRATING);
+				List<byte[]> keys = connection.clusterGetKeysInSlot(slot, Integer.MAX_VALUE);
+
+				for (byte[] key : keys) {
+					connection.migrate(key, source, 0, MigrateOption.COPY);
+				}
+				connection.clusterSetSlot(target, slot, AddSlots.NODE);
+				return null;
+			}
+		});
+	}
+
+	/**
+	 * Executed wrapped command upon {@link RedisClusterConnection}.
+	 * 
+	 * @param callback
+	 * @return
+	 */
+	public <T> T execute(RedisClusterCallback<T> callback) {
+
+		Assert.notNull(callback, "ClusterCallback must not be null!");
+
+		RedisClusterConnection connection = template.getConnectionFactory().getClusterConnection();
+
+		try {
+			return callback.doInRedis(connection);
+		} finally {
+			connection.close();
+		}
+	}
+}

src/main/java/org/springframework/data/redis/core/RedisClusterCallback.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright 2015 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.core;
+
+import org.springframework.dao.DataAccessException;
+import org.springframework.data.redis.connection.RedisClusterConnection;
+
+/**
+ * Callback interface for low level operations executed against a clustered Redis environment.
+ * 
+ * @author Christoph Strobl
+ * @since 1.7
+ * @param <T>
+ */
+public interface RedisClusterCallback<T> {
+
+	/**
+	 * Gets called by {@link RedisClusterTemplate} with an active Redis connection. Does not need to care about activating
+	 * or closing the connection or handling exceptions.
+	 * 
+	 * @param connection never {@literal null}.
+	 * @return
+	 * @throws DataAccessException
+	 */
+	T doInRedis(RedisClusterConnection connection) throws DataAccessException;
+}

src/main/java/org/springframework/data/redis/core/RedisOperations.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2014 the original author or authors.
+ * Copyright 2011-2015 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.
@@ -283,6 +283,14 @@ public interface RedisOperations<K, V> {
 	 */
 	<HK, HV> BoundHashOperations<K, HK, HV> boundHashOps(K key);
 
+	/**
+	 * Returns the cluster specific operations interface.
+	 * 
+	 * @return never {@literal null}.
+	 * @since 1.7
+	 */
+	ClusterOperations<K, V> opsForCluster();
+
 	List<V> sort(SortQuery<K> query);
 
 	<T> List<T> sort(SortQuery<K> query, RedisSerializer<T> resultSerializer);

src/main/java/org/springframework/data/redis/core/RedisTemplate.java

@@ -1016,6 +1016,14 @@ public class RedisTemplate<K, V> extends RedisAccessor implements RedisOperation
 		return new DefaultHashOperations<K, HK, HV>(this);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.RedisOperations#opsForCluster()
+	 */
+	public ClusterOperations<K, V> opsForCluster() {
+		return new DefaultClusterOperations<K, V>(this);
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.RedisOperations#killClient(java.lang.Object)