Spring/spring-data-redis
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Use Lettuce functionality for Cluster commands where possible.

Browse Source 
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
This commit is contained in:
Mark Paluch
2024-04-04 09:23:23 +02:00
committed by Christoph Strobl
parent d785b5f870
commit 6f28b530b0
9 changed files with 70 additions and 190 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
src/main/antora/modules/ROOT/pages/redis/cluster.adoc
View File

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

8
src/main/antora/modules/ROOT/pages/redis/pipelining.adoc
View File

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

44
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterKeyCommands.java
View File

@@ -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<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) {

38
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterServerCommands.java
View File

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

22
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterStringCommands.java
View File

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

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

@@ -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<byte[], byte[]> 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 {
<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 class LettuceConnection extends AbstractRedisConnection {
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 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;
}

10
src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionTests.java
View File

@@ -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

50
src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionUnitTests.java
View File

@@ -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.<byte[]> emptyList());
when(clusterConnection2Mock.keys(any(byte[].class))).thenReturn(Collections.<byte[]> emptyList());
when(clusterConnection3Mock.keys(any(byte[].class))).thenReturn(Collections.<byte[]> 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() {

23
src/test/java/org/springframework/data/redis/core/RedisTemplateIntegrationTests.java
View File

@@ -361,8 +361,7 @@ public class RedisTemplateIntegrationTests<K, V> {
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, V> {
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<K, V> {
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<K, V> {
th.start();
try {
th.join();
} catch (InterruptedException ignore) {
}
} catch (InterruptedException ignore) {}
operations.unwatch();
operations.multi();
@@ -794,8 +800,7 @@ public class RedisTemplateIntegrationTests<K, V> {
th.start();
try {
th.join();
} catch (InterruptedException ignore) {
}
} catch (InterruptedException ignore) {}
operations.multi();
operations.opsForValue().set(key1, value3);