From 6f28b530b0f77eeede0e0a6b514afb42eade51da Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Thu, 4 Apr 2024 09:23:23 +0200 Subject: [PATCH] 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 --- .../modules/ROOT/pages/redis/cluster.adoc | 3 + .../modules/ROOT/pages/redis/pipelining.adoc | 8 ++- .../lettuce/LettuceClusterKeyCommands.java | 44 ------------- .../lettuce/LettuceClusterServerCommands.java | 38 ------------ .../lettuce/LettuceClusterStringCommands.java | 22 ------- .../connection/lettuce/LettuceConnection.java | 62 +++++++++++-------- .../LettuceClusterConnectionTests.java | 10 ++- .../LettuceClusterConnectionUnitTests.java | 50 +-------------- .../core/RedisTemplateIntegrationTests.java | 23 ++++--- 9 files changed, 70 insertions(+), 190 deletions(-) diff --git a/src/main/antora/modules/ROOT/pages/redis/cluster.adoc b/src/main/antora/modules/ROOT/pages/redis/cluster.adoc index 0ad017e14..6b35ae5d3 100644 --- a/src/main/antora/modules/ROOT/pages/redis/cluster.adoc +++ b/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 --git a/src/main/antora/modules/ROOT/pages/redis/pipelining.adoc b/src/main/antora/modules/ROOT/pages/redis/pipelining.adoc index 8a4d70665..7e5ddd054 100644 --- a/src/main/antora/modules/ROOT/pages/redis/pipelining.adoc +++ b/src/main/antora/modules/ROOT/pages/redis/pipelining.adoc @@ -20,7 +20,9 @@ List 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 --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterKeyCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterKeyCommands.java index dc4fbc54d..ae3d3b2e4 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterKeyCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterKeyCommands.java @@ -18,11 +18,8 @@ package org.springframework.data.redis.connection.lettuce; 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 nodes = connection.clusterGetNodes(); - Set 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 keys(byte[] pattern) { - - Assert.notNull(pattern, "Pattern must not be null"); - - Collection> keysPerNode = connection.getClusterCommandExecutor() - .executeCommandOnAllNodes((LettuceClusterCommandCallback>) connection -> connection.keys(pattern)) - .resultsAsList(); - - Set keys = new HashSet<>(); - - for (List keySet : keysPerNode) { - keys.addAll(keySet); - } - return keys; - } - @Override public void rename(byte[] oldKey, byte[] newKey) { diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterServerCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterServerCommands.java index bab772c47..6d2e66cda 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterServerCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterServerCommands.java @@ -18,7 +18,6 @@ package org.springframework.data.redis.connection.lettuce; 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.convert.Converters; 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 @@ class LettuceClusterServerCommands extends LettuceServerCommands implements Redi executeCommandOnSingleNode(RedisServerCommands::save, node); } - @Override - public Long dbSize() { - - Collection 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 @@ class LettuceClusterServerCommands extends LettuceServerCommands implements Redi 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 --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterStringCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterStringCommands.java index c1be6421a..4b6ae60f0 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterStringCommands.java +++ b/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 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 entry : tuples.entrySet()) { - if (!setNX(entry.getKey(), entry.getValue()) && result) { - result = false; - } - } - return result; - } } diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java index 93f78151b..c3ecbde73 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java @@ -49,7 +49,9 @@ import java.util.LinkedList; 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 @@ import org.springframework.util.ObjectUtils; */ 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 CODEC = ByteArrayCodec.INSTANCE; @@ -189,8 +191,8 @@ public class LettuceConnection extends AbstractRedisConnection { /** * 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 class LettuceConnection extends AbstractRedisConnection { /** * 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 class LettuceConnection extends AbstractRedisConnection { /** * 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 @@ public class LettuceConnection extends AbstractRedisConnection { LettuceResult newLettuceResult(Future resultHolder, Converter converter) { - return LettuceResultBuilder.forResponse(resultHolder) - .mappedWith(converter) - .convertPipelineAndTxResults(this.convertPipelineAndTxResults) - .build(); + return LettuceResultBuilder. forResponse(resultHolder).mappedWith(converter) + .convertPipelineAndTxResults(this.convertPipelineAndTxResults).build(); } LettuceResult newLettuceResult(Future resultHolder, Converter converter, Supplier defaultValue) { - return LettuceResultBuilder.forResponse(resultHolder) - .mappedWith(converter) - .convertPipelineAndTxResults(this.convertPipelineAndTxResults) - .defaultNullTo(defaultValue) - .build(); + return LettuceResultBuilder. forResponse(resultHolder).mappedWith(converter) + .convertPipelineAndTxResults(this.convertPipelineAndTxResults).defaultNullTo(defaultValue).build(); } LettuceResult newLettuceStatusResult(Future resultHolder) { - return LettuceResultBuilder.forResponse(resultHolder).buildStatusResult(); + return LettuceResultBuilder. forResponse(resultHolder).buildStatusResult(); } void pipeline(LettuceResult result) { @@ -583,7 +580,7 @@ public class LettuceConnection extends AbstractRedisConnection { pipeliningFlushState = null; isPipelined = false; - List> futures = new ArrayList<>(ppline.size()); + List> futures = new ArrayList<>(ppline.size()); for (LettuceResult result : ppline) { futures.add(result.getResultHolder()); @@ -600,10 +597,24 @@ public class LettuceConnection extends AbstractRedisConnection { if (done) { for (LettuceResult result : ppline) { - if (result.getResultHolder().getOutput().hasError()) { + CompletableFuture resultHolder = result.getResultHolder(); + if (resultHolder.isCompletedExceptionally()) { - Exception exception = new InvalidDataAccessApiUsageException(result.getResultHolder() - .getOutput().getError()); + 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(message); // remember only the first error if (problem == null) { @@ -684,8 +695,8 @@ public class LettuceConnection extends AbstractRedisConnection { 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 @@ public class LettuceConnection extends AbstractRedisConnection { try { return (T) (converter != null ? converter.convert(source) : source); - } catch (IndexOutOfBoundsException ignore) { - } + } catch (IndexOutOfBoundsException ignore) {} return null; } diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionTests.java index 67c9b066b..5ecba1547 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionTests.java @@ -1620,7 +1620,15 @@ public class LettuceClusterConnectionTests implements ClusterConnectionTests { nativeConnection.set(KEY_1, VALUE_1); nativeConnection.set(KEY_2, VALUE_2); - assertThat(clusterConnection.randomKey()).isNotNull(); + for (int i = 0; i < 20; i++) { + + byte[] k = clusterConnection.randomKey(); + if (k == null) { + continue; + } + + assertThat(k).isIn(KEY_1_BYTES, KEY_2_BYTES); + } } @Test // DATAREDIS-315 diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionUnitTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionUnitTests.java index 8f1803d17..25fd46d78 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionUnitTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionUnitTests.java @@ -18,7 +18,6 @@ package org.springframework.data.redis.connection.lettuce; import static org.assertj.core.api.Assertions.*; import static org.mockito.Mockito.*; import static org.springframework.data.redis.connection.ClusterTestVariables.*; -import static org.springframework.data.redis.test.util.MockitoUtils.*; import io.lettuce.core.RedisFuture; import io.lettuce.core.RedisURI; @@ -46,6 +45,7 @@ import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; import org.mockito.junit.jupiter.MockitoSettings; import org.mockito.quality.Strictness; + import org.springframework.data.redis.connection.ClusterCommandExecutor; import org.springframework.data.redis.connection.ClusterNodeResourceProvider; import org.springframework.data.redis.connection.ClusterTopologyProvider; @@ -193,22 +193,6 @@ class LettuceClusterConnectionUnitTests { assertThat(connection.isClosed()).isTrue(); } - @Test // DATAREDIS-315 - void keysShouldBeRunOnAllClusterNodes() { - - when(clusterConnection1Mock.keys(any(byte[].class))).thenReturn(Collections. emptyList()); - when(clusterConnection2Mock.keys(any(byte[].class))).thenReturn(Collections. emptyList()); - when(clusterConnection3Mock.keys(any(byte[].class))).thenReturn(Collections. emptyList()); - - byte[] pattern = LettuceConverters.toBytes("*"); - - connection.keys(pattern); - - verify(clusterConnection1Mock, times(1)).keys(pattern); - verify(clusterConnection2Mock, times(1)).keys(pattern); - verify(clusterConnection3Mock, times(1)).keys(pattern); - } - @Test // DATAREDIS-315 void keysShouldOnlyBeRunOnDedicatedNodeWhenPinned() { @@ -223,38 +207,6 @@ class LettuceClusterConnectionUnitTests { verify(clusterConnection3Mock, never()).keys(pattern); } - @Test // DATAREDIS-315 - void randomKeyShouldReturnAnyKeyFromRandomNode() { - - when(clusterConnection1Mock.randomkey()).thenReturn(KEY_1_BYTES); - when(clusterConnection2Mock.randomkey()).thenReturn(KEY_2_BYTES); - when(clusterConnection3Mock.randomkey()).thenReturn(KEY_3_BYTES); - - assertThat(connection.randomKey()).isIn(KEY_1_BYTES, KEY_2_BYTES, KEY_3_BYTES); - verifyInvocationsAcross("randomkey", times(1), clusterConnection1Mock, clusterConnection2Mock, - clusterConnection3Mock); - } - - @Test // DATAREDIS-315 - void randomKeyShouldReturnKeyWhenAvailableOnAnyNode() { - - when(clusterConnection3Mock.randomkey()).thenReturn(KEY_3_BYTES); - - for (int i = 0; i < 100; i++) { - assertThat(connection.randomKey()).isEqualTo(KEY_3_BYTES); - } - } - - @Test // DATAREDIS-315 - void randomKeyShouldReturnNullWhenNoKeysPresentOnAllNodes() { - - when(clusterConnection1Mock.randomkey()).thenReturn(null); - when(clusterConnection2Mock.randomkey()).thenReturn(null); - when(clusterConnection3Mock.randomkey()).thenReturn(null); - - assertThat(connection.randomKey()).isNull(); - } - @Test // DATAREDIS-315 void clusterSetSlotImportingShouldBeExecutedCorrectly() { diff --git a/src/test/java/org/springframework/data/redis/core/RedisTemplateIntegrationTests.java b/src/test/java/org/springframework/data/redis/core/RedisTemplateIntegrationTests.java index 93fc2fff4..bbc18c6f0 100644 --- a/src/test/java/org/springframework/data/redis/core/RedisTemplateIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/core/RedisTemplateIntegrationTests.java @@ -361,8 +361,7 @@ public class RedisTemplateIntegrationTests { try { // Await EXEC completion as it's executed on a dedicated connection. Thread.sleep(100); - } catch (InterruptedException ignore) { - } + } catch (InterruptedException ignore) {} operations.opsForValue().set(key1, value1); operations.opsForValue().get(key1); @@ -673,7 +672,16 @@ public class RedisTemplateIntegrationTests { K key1 = keyFactory.instance(); V value1 = valueFactory.instance(); redisTemplate.opsForValue().set(key1, value1); - assertThat(redisTemplate.randomKey()).isEqualTo(key1); + + for (int i = 0; i < 20; i++) { + + K k = redisTemplate.randomKey(); + if (k == null) { + continue; + } + + assertThat(k).isEqualTo(key1); + } } @ParameterizedRedisTest @@ -723,8 +731,7 @@ public class RedisTemplateIntegrationTests { th.start(); try { th.join(); - } catch (InterruptedException ignore) { - } + } catch (InterruptedException ignore) {} operations.multi(); operations.opsForValue().set(key1, value3); @@ -756,8 +763,7 @@ public class RedisTemplateIntegrationTests { th.start(); try { th.join(); - } catch (InterruptedException ignore) { - } + } catch (InterruptedException ignore) {} operations.unwatch(); operations.multi(); @@ -794,8 +800,7 @@ public class RedisTemplateIntegrationTests { th.start(); try { th.join(); - } catch (InterruptedException ignore) { - } + } catch (InterruptedException ignore) {} operations.multi(); operations.opsForValue().set(key1, value3);