diff --git a/src/main/java/org/springframework/data/redis/cache/RedisCache.java b/src/main/java/org/springframework/data/redis/cache/RedisCache.java
index 6707121d9..c8c5f5296 100644
--- a/src/main/java/org/springframework/data/redis/cache/RedisCache.java
+++ b/src/main/java/org/springframework/data/redis/cache/RedisCache.java
@@ -28,6 +28,7 @@ import org.springframework.cache.Cache;
 import org.springframework.cache.support.SimpleValueWrapper;
 import org.springframework.dao.DataAccessException;
 import org.springframework.data.redis.RedisSystemException;
+import org.springframework.data.redis.connection.DecoratedRedisConnection;
 import org.springframework.data.redis.connection.RedisClusterConnection;
 import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.ReturnType;
@@ -39,10 +40,11 @@ import org.springframework.util.ClassUtils;
 
 /**
  * Cache implementation on top of Redis.
- * 
+ *
  * @author Costin Leau
  * @author Christoph Strobl
  * @author Thomas Darimont
+ * @author Mark Paluch
  */
 @SuppressWarnings("unchecked")
 public class RedisCache implements Cache {
@@ -54,7 +56,7 @@ public class RedisCache implements Cache {
 
 	/**
 	 * Constructs a new <code>RedisCache</code> instance.
-	 * 
+	 *
 	 * @param name cache name
 	 * @param prefix
 	 * @param redisOperations
@@ -624,7 +626,16 @@ public class RedisCache implements Cache {
 			byte[] prefixToUse = Arrays.copyOf(metadata.getKeyPrefix(), metadata.getKeyPrefix().length + WILD_CARD.length);
 			System.arraycopy(WILD_CARD, 0, prefixToUse, metadata.getKeyPrefix().length, WILD_CARD.length);
 
-			connection.eval(REMOVE_KEYS_BY_PATTERN_LUA, ReturnType.INTEGER, 0, prefixToUse);
+			if(isClusterConnection(connection)) {
+
+				// load keys to the client because currently Redis Cluster connections do not allow eval of lua scripts.
+				Set<byte[]> keys = connection.keys(prefixToUse);
+				if (!keys.isEmpty()) {
+					connection.del(keys.toArray(new byte[keys.size()][]));
+				}
+			} else {
+				connection.eval(REMOVE_KEYS_BY_PATTERN_LUA, ReturnType.INTEGER, 0, prefixToUse);
+			}
 
 			return null;
 		}
@@ -765,8 +776,9 @@ public class RedisCache implements Cache {
 
 					return value;
 				} catch (RuntimeException e) {
-
-					connection.discard();
+					if (!isClusterConnection(connection)) {
+						connection.discard();
+					}
 					throw e;
 				}
 			} finally {
@@ -810,6 +822,11 @@ public class RedisCache implements Cache {
 	}
 
 	private static boolean isClusterConnection(RedisConnection connection) {
+
+		while (connection instanceof DecoratedRedisConnection) {
+			connection = ((DecoratedRedisConnection) connection).getDelegate();
+		}
+
 		return connection instanceof RedisClusterConnection;
 	}
 
diff --git a/src/main/java/org/springframework/data/redis/connection/DecoratedRedisConnection.java b/src/main/java/org/springframework/data/redis/connection/DecoratedRedisConnection.java
new file mode 100644
index 000000000..8f9969cc3
--- /dev/null
+++ b/src/main/java/org/springframework/data/redis/connection/DecoratedRedisConnection.java
@@ -0,0 +1,34 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * 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;
+
+/**
+ * Specifies that the connection decorates another {@link RedisConnection}.
+ *
+ * @author Mark Paluch
+ * @since 1.7
+ */
+public interface DecoratedRedisConnection {
+
+	/**
+	 * Gets the underlying {@link RedisConnection}.
+	 *
+	 * @return never {@literal null}.
+	 */
+	RedisConnection getDelegate();
+
+}
diff --git a/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java b/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java
index 5cdf6ee9e..91b6e92f8 100644
--- a/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java
+++ b/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java
@@ -49,8 +49,9 @@ import org.springframework.util.Assert;
  * @author Jennifer Hickey
  * @author Christoph Strobl
  * @author Thomas Darimont
+ * @author Mark Paluch
  */
-public class DefaultStringRedisConnection implements StringRedisConnection {
+public class DefaultStringRedisConnection implements StringRedisConnection, DecoratedRedisConnection {
 
 	private static final byte[][] EMPTY_2D_BYTE_ARRAY = new byte[0][];
 
@@ -2724,4 +2725,13 @@ public class DefaultStringRedisConnection implements StringRedisConnection {
 		delegate.migrate(key, target, dbIndex, option, timeout);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.DecoratedRedisConnection#getDelegate()
+	 */
+	@Override
+	public RedisConnection getDelegate() {
+		return delegate;
+	}
+
 }