Use Lettuce functionality for Cluster commands where possible.

We now remove our own code in favor of Lettuce's advanced cluster support to leverage asynchronous functionality in pipelining.

Document pipelining restrictions regarding Redis Cluster.

Original Pull Request: #2889

Author: mp911de

Diff for: src/main/antora/modules/ROOT/pages/redis/cluster.adoc

@@ -129,3 +129,6 @@ clusterOps.shutdown(NODE_7379);                                              <1>
 
 <1> Shut down node at 7379 and cross fingers there is a replica in place that can take over.
 ====
+
+NOTE: Redis Cluster pipelining is currently only supported throug the Lettuce driver except for the following commands when using cross-slot keys: `rename`, `renameNX`, `sort`, `bLPop`, `bRPop`, `rPopLPush`, `bRPopLPush`, `info`, `sMove`, `sInter`, `sInterStore`, `sUnion`, `sUnionStore`, `sDiff`, `sDiffStore`.
+Same-slot keys are fully supported.

Diff for: src/main/antora/modules/ROOT/pages/redis/pipelining.adoc

@@ -20,7 +20,9 @@ List<Object> results = stringRedisTemplate.executePipelined(
 });
 ----
 
-The preceding example runs a bulk right pop of items from a queue in a pipeline. The `results` `List` contains all of the popped items. `RedisTemplate` uses its value, hash key, and hash value serializers to deserialize all results before returning, so the returned items in the preceding example are Strings. There are additional `executePipelined` methods that let you pass a custom serializer for pipelined results.
+The preceding example runs a bulk right pop of items from a queue in a pipeline.
+The `results` `List` contains all the popped items. `RedisTemplate` uses its value, hash key, and hash value serializers to deserialize all results before returning, so the returned items in the preceding example are Strings.
+There are additional `executePipelined` methods that let you pass a custom serializer for pipelined results.
 
 Note that the value returned from the `RedisCallback` is required to be `null`, as this value is discarded in favor of returning the results of the pipelined commands.
 
@@ -35,3 +37,7 @@ factory.setPipeliningFlushPolicy(PipeliningFlushPolicy.buffered(3)); <1>
 ----
 <1> Buffer locally and flush after every 3rd command.
 ====
+
+NOTE: Pipelining is limited to Redis Standalone.
+Redis Cluster is currently only supported through the Lettuce driver except for the following commands when using cross-slot keys: `rename`, `renameNX`, `sort`, `bLPop`, `bRPop`, `rPopLPush`, `bRPopLPush`, `info`, `sMove`, `sInter`, `sInterStore`, `sUnion`, `sUnionStore`, `sDiff`, `sDiffStore`.
+Same-slot keys are fully supported.

Diff for: src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterKeyCommands.java

@@ -18,11 +18,8 @@
 import io.lettuce.core.KeyScanCursor;
 import io.lettuce.core.ScanArgs;
 
-import java.util.Collection;
-import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
-import java.util.concurrent.ThreadLocalRandom;
 
 import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.redis.connection.ClusterSlotHashUtil;
@@ -50,47 +47,6 @@ class LettuceClusterKeyCommands extends LettuceKeyCommands {
 		this.connection = connection;
 	}
 
-	@Override
-	public byte[] randomKey() {
-
-		List<RedisClusterNode> nodes = connection.clusterGetNodes();
-		Set<RedisClusterNode> inspectedNodes = new HashSet<>(nodes.size());
-
-		do {
-
-			RedisClusterNode node = nodes.get(ThreadLocalRandom.current().nextInt(nodes.size()));
-
-			while (inspectedNodes.contains(node)) {
-				node = nodes.get(ThreadLocalRandom.current().nextInt(nodes.size()));
-			}
-			inspectedNodes.add(node);
-			byte[] key = randomKey(node);
-
-			if (key != null && key.length > 0) {
-				return key;
-			}
-		} while (nodes.size() != inspectedNodes.size());
-
-		return null;
-	}
-
-	@Override
-	public Set<byte[]> keys(byte[] pattern) {
-
-		Assert.notNull(pattern, "Pattern must not be null");
-
-		Collection<List<byte[]>> keysPerNode = connection.getClusterCommandExecutor()
-				.executeCommandOnAllNodes((LettuceClusterCommandCallback<List<byte[]>>) connection -> connection.keys(pattern))
-				.resultsAsList();
-
-		Set<byte[]> keys = new HashSet<>();
-
-		for (List<byte[]> keySet : keysPerNode) {
-			keys.addAll(keySet);
-		}
-		return keys;
-	}
-
 	@Override
 	public void rename(byte[] oldKey, byte[] newKey) {
 

Diff for: src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterServerCommands.java

@@ -18,7 +18,6 @@
 import io.lettuce.core.api.sync.RedisServerCommands;
 
 import java.util.ArrayList;
-import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
@@ -34,7 +33,6 @@
 import org.springframework.data.redis.connection.lettuce.LettuceClusterConnection.LettuceClusterCommandCallback;
 import org.springframework.data.redis.core.types.RedisClientInfo;
 import org.springframework.util.Assert;
-import org.springframework.util.CollectionUtils;
 
 /**
  * @author Mark Paluch
@@ -71,37 +69,11 @@ public void save(RedisClusterNode node) {
 		executeCommandOnSingleNode(RedisServerCommands::save, node);
 	}
 
-	@Override
-	public Long dbSize() {
-
-		Collection<Long> dbSizes = executeCommandOnAllNodes(RedisServerCommands::dbsize).resultsAsList();
-
-		if (CollectionUtils.isEmpty(dbSizes)) {
-			return 0L;
-		}
-
-		Long size = 0L;
-		for (Long value : dbSizes) {
-			size += value;
-		}
-		return size;
-	}
-
 	@Override
 	public Long dbSize(RedisClusterNode node) {
 		return executeCommandOnSingleNode(RedisServerCommands::dbsize, node).getValue();
 	}
 
-	@Override
-	public void flushDb() {
-		executeCommandOnAllNodes(RedisServerCommands::flushdb);
-	}
-
-	@Override
-	public void flushDb(FlushOption option) {
-		executeCommandOnAllNodes(it -> it.flushdb(LettuceConverters.toFlushMode(option)));
-	}
-
 	@Override
 	public void flushDb(RedisClusterNode node) {
 		executeCommandOnSingleNode(RedisServerCommands::flushdb, node);
@@ -112,16 +84,6 @@ public void flushDb(RedisClusterNode node, FlushOption option) {
 		executeCommandOnSingleNode(it -> it.flushdb(LettuceConverters.toFlushMode(option)), node);
 	}
 
-	@Override
-	public void flushAll() {
-		executeCommandOnAllNodes(RedisServerCommands::flushall);
-	}
-
-	@Override
-	public void flushAll(FlushOption option) {
-		executeCommandOnAllNodes(it -> it.flushall(LettuceConverters.toFlushMode(option)));
-	}
-
 	@Override
 	public void flushAll(RedisClusterNode node) {
 		executeCommandOnSingleNode(RedisServerCommands::flushall, node);

Diff for: src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterStringCommands.java

@@ -15,11 +15,6 @@
  */
 package org.springframework.data.redis.connection.lettuce;
 
-import java.util.Map;
-
-import org.springframework.data.redis.connection.ClusterSlotHashUtil;
-import org.springframework.util.Assert;
-
 /**
  * @author Christoph Strobl
  * @author Mark Paluch
@@ -31,21 +26,4 @@ class LettuceClusterStringCommands extends LettuceStringCommands {
 		super(connection);
 	}
 
-	@Override
-	public Boolean mSetNX(Map<byte[], byte[]> tuples) {
-
-		Assert.notNull(tuples, "Tuples must not be null");
-
-		if (ClusterSlotHashUtil.isSameSlotForAllKeys(tuples.keySet().toArray(new byte[tuples.keySet().size()][]))) {
-			return super.mSetNX(tuples);
-		}
-
-		boolean result = true;
-		for (Map.Entry<byte[], byte[]> entry : tuples.entrySet()) {
-			if (!setNX(entry.getKey(), entry.getValue()) && result) {
-				result = false;
-			}
-		}
-		return result;
-	}
 }

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

@@ -49,7 +49,9 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Queue;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
@@ -102,8 +104,8 @@
  */
 public class LettuceConnection extends AbstractRedisConnection {
 
-	private static final ExceptionTranslationStrategy EXCEPTION_TRANSLATION =
-			new FallbackExceptionTranslationStrategy(LettuceExceptionConverter.INSTANCE);
+	private static final ExceptionTranslationStrategy EXCEPTION_TRANSLATION = new FallbackExceptionTranslationStrategy(
+			LettuceExceptionConverter.INSTANCE);
 
 	static final RedisCodec<byte[], byte[]> CODEC = ByteArrayCodec.INSTANCE;
 
@@ -189,8 +191,8 @@ public LettuceConnection(@Nullable StatefulRedisConnection<byte[], byte[]> share
 	/**
 	 * Creates a new {@link LettuceConnection}.
 	 *
-	 * @param sharedConnection A native connection that is shared with other {@link LettuceConnection}s.
-	 * Should not be used for transactions or blocking operations.
+	 * @param sharedConnection A native connection that is shared with other {@link LettuceConnection}s. Should not be
+	 *          used for transactions or blocking operations.
 	 * @param timeout The connection timeout (in milliseconds)
 	 * @param client The {@link RedisClient} to use when making pub/sub connections.
 	 * @param defaultDbIndex The db index to use along with {@link RedisClient} when establishing a dedicated connection.
@@ -209,8 +211,8 @@ public LettuceConnection(@Nullable StatefulRedisConnection<byte[], byte[]> share
 	/**
 	 * Creates a new {@link LettuceConnection}.
 	 *
-	 * @param sharedConnection A native connection that is shared with other {@link LettuceConnection}s.
-	 * Should not be used for transactions or blocking operations.
+	 * @param sharedConnection A native connection that is shared with other {@link LettuceConnection}s. Should not be
+	 *          used for transactions or blocking operations.
 	 * @param connectionProvider connection provider to obtain and release native connections.
 	 * @param timeout The connection timeout (in milliseconds)
 	 * @param defaultDbIndex The db index to use along with {@link RedisClient} when establishing a dedicated connection.
@@ -225,8 +227,8 @@ public LettuceConnection(@Nullable StatefulRedisConnection<byte[], byte[]> share
 	/**
 	 * Creates a new {@link LettuceConnection}.
 	 *
-	 * @param sharedConnection A native connection that is shared with other {@link LettuceConnection}s.
-	 * Should not be used for transactions or blocking operations.
+	 * @param sharedConnection A native connection that is shared with other {@link LettuceConnection}s. Should not be
+	 *          used for transactions or blocking operations.
 	 * @param connectionProvider connection provider to obtain and release native connections.
 	 * @param timeout The connection timeout (in milliseconds)
 	 * @param defaultDbIndex The db index to use along with {@link RedisClient} when establishing a dedicated connection.
@@ -453,24 +455,19 @@ private LettuceInvoker doInvoke(RedisClusterAsyncCommands<byte[], byte[]> connec
 
 	<T, R> LettuceResult<T, R> newLettuceResult(Future<T> resultHolder, Converter<T, R> converter) {
 
-		return LettuceResultBuilder.<T, R>forResponse(resultHolder)
-				.mappedWith(converter)
-				.convertPipelineAndTxResults(this.convertPipelineAndTxResults)
-				.build();
+		return LettuceResultBuilder.<T, R> forResponse(resultHolder).mappedWith(converter)
+				.convertPipelineAndTxResults(this.convertPipelineAndTxResults).build();
 	}
 
 	<T, R> LettuceResult<T, R> newLettuceResult(Future<T> resultHolder, Converter<T, R> converter,
 			Supplier<R> defaultValue) {
 
-		return LettuceResultBuilder.<T, R>forResponse(resultHolder)
-				.mappedWith(converter)
-				.convertPipelineAndTxResults(this.convertPipelineAndTxResults)
-				.defaultNullTo(defaultValue)
-				.build();
+		return LettuceResultBuilder.<T, R> forResponse(resultHolder).mappedWith(converter)
+				.convertPipelineAndTxResults(this.convertPipelineAndTxResults).defaultNullTo(defaultValue).build();
 	}
 
 	<T, R> LettuceResult<T, R> newLettuceStatusResult(Future<T> resultHolder) {
-		return LettuceResultBuilder.<T, R>forResponse(resultHolder).buildStatusResult();
+		return LettuceResultBuilder.<T, R> forResponse(resultHolder).buildStatusResult();
 	}
 
 	void pipeline(LettuceResult<?, ?> result) {
@@ -583,7 +580,7 @@ public List<Object> closePipeline() {
 		pipeliningFlushState = null;
 		isPipelined = false;
 
-		List<io.lettuce.core.protocol.RedisCommand<?, ?, ?>> futures = new ArrayList<>(ppline.size());
+		List<CompletableFuture<?>> futures = new ArrayList<>(ppline.size());
 
 		for (LettuceResult<?, ?> result : ppline) {
 			futures.add(result.getResultHolder());
@@ -600,10 +597,24 @@ public List<Object> closePipeline() {
 			if (done) {
 				for (LettuceResult<?, ?> result : ppline) {
 
-					if (result.getResultHolder().getOutput().hasError()) {
+					CompletableFuture<?> resultHolder = result.getResultHolder();
+					if (resultHolder.isCompletedExceptionally()) {
+
+						String message;
+						if (resultHolder instanceof io.lettuce.core.protocol.RedisCommand<?, ?, ?> rc) {
+							message = rc.getOutput().getError();
+						} else {
+							try {
+								resultHolder.get();
+								message = "";
+							} catch (InterruptedException ignore) {
+								message = "";
+							} catch (ExecutionException e) {
+								message = e.getCause().getMessage();
+							}
+						}
 
-						Exception exception = new InvalidDataAccessApiUsageException(result.getResultHolder()
-								.getOutput().getError());
+						Exception exception = new InvalidDataAccessApiUsageException(message);
 
 						// remember only the first error
 						if (problem == null) {
@@ -684,8 +695,8 @@ public List<Object> exec() {
 				LettuceTransactionResultConverter resultConverter = new LettuceTransactionResultConverter(
 						new LinkedList<>(txResults), exceptionConverter);
 
-				pipeline(newLettuceResult(exec, source ->
-						resultConverter.convert(LettuceConverters.transactionResultUnwrapper().convert(source))));
+				pipeline(newLettuceResult(exec,
+						source -> resultConverter.convert(LettuceConverters.transactionResultUnwrapper().convert(source))));
 
 				return null;
 			}
@@ -837,8 +848,7 @@ <T> T failsafeReadScanValues(List<?> source, @SuppressWarnings("rawtypes") @Null
 
 		try {
 			return (T) (converter != null ? converter.convert(source) : source);
-		} catch (IndexOutOfBoundsException ignore) {
- 		}
+		} catch (IndexOutOfBoundsException ignore) {}
 
 		return null;
 	}