From 3e9cedfdcb7b3c81c9911bdb882b45fa717b1ce0 Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Mon, 20 May 2019 16:51:48 +0200 Subject: [PATCH] DATAREDIS-973 - Fix database selection using Lettuce. We now correctly set the requested database using RedisURI so the driver performs the database selection. We no longer need to call select ourselves. For shared connections (pooled and single shared connection) we now no longer call select. This can leave connections with a different database associated. Original Pull Request: #451 --- .../lettuce/LettuceConnectionFactory.java | 15 +-- .../LettuceConnectionFactoryTests.java | 94 +++++++++++++++---- .../LettuceSentinelIntegrationTests.java | 46 +++++++-- 3 files changed, 120 insertions(+), 35 deletions(-) diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java index 26a74ccd5..ff2c24444 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java @@ -437,7 +437,7 @@ public class LettuceConnectionFactory synchronized (this.connectionMonitor) { if (this.connection == null) { - this.connection = new SharedConnection<>(connectionProvider, false); + this.connection = new SharedConnection<>(connectionProvider); } return this.connection; @@ -449,7 +449,7 @@ public class LettuceConnectionFactory synchronized (this.connectionMonitor) { if (this.reactiveConnection == null) { - this.reactiveConnection = new SharedConnection<>(reactiveConnectionProvider, true); + this.reactiveConnection = new SharedConnection<>(reactiveConnectionProvider); } return this.reactiveConnection; @@ -1013,6 +1013,7 @@ public class LettuceConnectionFactory getRedisPassword().toOptional().ifPresent(builder::withPassword); clientConfiguration.getClientName().ifPresent(builder::withClientName); + builder.withDatabase(getDatabase()); builder.withSsl(clientConfiguration.isUseSsl()); builder.withVerifyPeer(clientConfiguration.isVerifyPeer()); builder.withStartTls(clientConfiguration.isStartTls()); @@ -1026,6 +1027,7 @@ public class LettuceConnectionFactory RedisURI.Builder builder = RedisURI.Builder.socket(socketPath); getRedisPassword().toOptional().ifPresent(builder::withPassword); + builder.withDatabase(getDatabase()); builder.withTimeout(clientConfiguration.getCommandTimeout()); return builder.build(); @@ -1062,7 +1064,6 @@ public class LettuceConnectionFactory class SharedConnection { private final LettuceConnectionProvider connectionProvider; - private final boolean shareNativeClusterConnection; /** Synchronization monitor for the shared Connection */ private final Object connectionMonitor = new Object(); @@ -1100,13 +1101,7 @@ public class LettuceConnectionFactory private StatefulConnection getNativeConnection() { try { - - StatefulConnection connection = connectionProvider.getConnection(StatefulConnection.class); - if (connection instanceof StatefulRedisConnection && getDatabase() > 0) { - ((StatefulRedisConnection) connection).sync().select(getDatabase()); - } - - return connection; + return connectionProvider.getConnection(StatefulConnection.class); } catch (RedisException e) { throw new RedisConnectionFailureException("Unable to connect to Redis", e); } diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java index 4fa86d99d..6985f0bc2 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java @@ -25,9 +25,11 @@ import io.lettuce.core.ReadFrom; import io.lettuce.core.RedisException; import io.lettuce.core.api.async.RedisAsyncCommands; import io.lettuce.core.api.reactive.BaseRedisReactiveCommands; +import reactor.test.StepVerifier; import java.io.File; import java.time.Duration; +import java.util.function.Consumer; import org.apache.commons.pool2.impl.GenericObjectPoolConfig; import org.junit.After; @@ -35,14 +37,15 @@ import org.junit.AfterClass; import org.junit.Before; import org.junit.Ignore; import org.junit.Test; + import org.springframework.data.redis.ConnectionFactoryTracker; import org.springframework.data.redis.RedisConnectionFailureException; import org.springframework.data.redis.RedisSystemException; import org.springframework.data.redis.SettingsUtils; import org.springframework.data.redis.connection.DefaultStringRedisConnection; import org.springframework.data.redis.connection.RedisConnection; -import org.springframework.data.redis.connection.RedisStaticMasterReplicaConfiguration; import org.springframework.data.redis.connection.RedisStandaloneConfiguration; +import org.springframework.data.redis.connection.RedisStaticMasterReplicaConfiguration; import org.springframework.data.redis.connection.StringRedisConnection; /** @@ -131,30 +134,88 @@ public class LettuceConnectionFactoryTests { assertSame(connection.getNativeConnection(), conn2.getNativeConnection()); } - @Test + @Test // DATAREDIS-973 public void testSelectDb() { - LettuceConnectionFactory factory2 = new LettuceConnectionFactory(SettingsUtils.getHost(), SettingsUtils.getPort()); - factory2.setClientResources(LettuceTestClientResources.getSharedClientResources()); - factory2.setShutdownTimeout(0); - factory2.setDatabase(1); - factory2.afterPropertiesSet(); + // put an item in database 0 + connection.set("sometestkey", "sometestvalue"); + + LettuceConnectionFactory sharingConnectionFactory = newConnectionFactory(cf -> {}); + runSelectDbTest(sharingConnectionFactory); + + LettuceConnectionFactory nonSharingConnectionFactory = newConnectionFactory( + cf -> cf.setShareNativeConnection(false)); + runSelectDbTest(nonSharingConnectionFactory); + } + + private void runSelectDbTest(LettuceConnectionFactory factory2) { ConnectionFactoryTracker.add(factory2); - StringRedisConnection connection2 = new DefaultStringRedisConnection(factory2.getConnection()); - connection2.flushDb(); + StringRedisConnection separateDatabase = new DefaultStringRedisConnection(factory2.getConnection()); + separateDatabase.flushDb(); + // put an item in database 0 connection.set("sometestkey", "sometestvalue"); + try { // there should still be nothing in database 1 - assertEquals(Long.valueOf(0), connection2.dbSize()); + assertEquals(Long.valueOf(0), separateDatabase.dbSize()); } finally { - connection2.close(); + separateDatabase.close(); factory2.destroy(); } } + @Test // DATAREDIS-973 + public void testSelectDbReactive() { + + LettuceConnectionFactory sharingConnectionFactory = newConnectionFactory(cf -> {}); + runSelectDbReactiveTest(sharingConnectionFactory); + + LettuceConnectionFactory nonSharingConnectionFactory = newConnectionFactory( + cf -> cf.setShareNativeConnection(false)); + runSelectDbTest(nonSharingConnectionFactory); + } + + private void runSelectDbReactiveTest(LettuceConnectionFactory factory2) { + + ConnectionFactoryTracker.add(factory2); + + LettuceReactiveRedisConnection separateDatabase = factory2.getReactiveConnection(); + + separateDatabase.serverCommands().flushDb() // + .as(StepVerifier::create) // + .expectNextCount(1) // + .verifyComplete(); + + // put an item in database 0 + connection.set("sometestkey", "sometestvalue"); + + try { + // there should still be nothing in database 1 + separateDatabase.serverCommands().dbSize() // + .as(StepVerifier::create).expectNext(0L) // + .verifyComplete(); + } finally { + separateDatabase.close(); + factory2.destroy(); + } + } + + private static LettuceConnectionFactory newConnectionFactory(Consumer customizer) { + + LettuceConnectionFactory connectionFactory = new LettuceConnectionFactory(SettingsUtils.getHost(), + SettingsUtils.getPort()); + connectionFactory.setClientResources(LettuceTestClientResources.getSharedClientResources()); + connectionFactory.setShutdownTimeout(0); + connectionFactory.setDatabase(1); + customizer.accept(connectionFactory); + connectionFactory.afterPropertiesSet(); + + return connectionFactory; + } + @SuppressWarnings("unchecked") @Test public void testDisableSharedConnection() throws Exception { @@ -394,11 +455,10 @@ public class LettuceConnectionFactoryTests { LettuceClientConfiguration configuration = LettuceTestClientConfiguration.builder().readFrom(ReadFrom.SLAVE) .build(); - RedisStaticMasterReplicaConfiguration elastiCache = new RedisStaticMasterReplicaConfiguration(SettingsUtils.getHost()) - .node(SettingsUtils.getHost(), SettingsUtils.getPort() + 1); + RedisStaticMasterReplicaConfiguration elastiCache = new RedisStaticMasterReplicaConfiguration( + SettingsUtils.getHost()).node(SettingsUtils.getHost(), SettingsUtils.getPort() + 1); - LettuceConnectionFactory factory = new LettuceConnectionFactory(elastiCache, - configuration); + LettuceConnectionFactory factory = new LettuceConnectionFactory(elastiCache, configuration); factory.afterPropertiesSet(); RedisConnection connection = factory.getConnection(); @@ -422,8 +482,8 @@ public class LettuceConnectionFactoryTests { LettuceClientConfiguration configuration = LettuceTestClientConfiguration.builder().readFrom(ReadFrom.MASTER) .build(); - RedisStaticMasterReplicaConfiguration elastiCache = new RedisStaticMasterReplicaConfiguration(SettingsUtils.getHost(), - SettingsUtils.getPort() + 1); + RedisStaticMasterReplicaConfiguration elastiCache = new RedisStaticMasterReplicaConfiguration( + SettingsUtils.getHost(), SettingsUtils.getPort() + 1); LettuceConnectionFactory factory = new LettuceConnectionFactory(elastiCache, configuration); factory.afterPropertiesSet(); diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelIntegrationTests.java index 989aa6000..e7474aeb3 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelIntegrationTests.java @@ -21,7 +21,9 @@ import static org.hamcrest.core.IsEqual.equalTo; import static org.junit.Assert.*; import io.lettuce.core.ReadFrom; +import reactor.test.StepVerifier; +import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.Collection; import java.util.List; @@ -34,6 +36,7 @@ import org.junit.Test; import org.junit.runner.RunWith; import org.junit.runners.Parameterized; import org.junit.runners.Parameterized.Parameters; + import org.springframework.data.redis.ConnectionFactoryTracker; import org.springframework.data.redis.connection.AbstractConnectionIntegrationTests; import org.springframework.data.redis.connection.DefaultStringRedisConnection; @@ -65,7 +68,7 @@ public class LettuceSentinelIntegrationTests extends AbstractConnectionIntegrati static { SENTINEL_CONFIG = new RedisSentinelConfiguration() // - .master(MASTER_NAME).sentinel(SENTINEL_0).sentinel(SENTINEL_1); + .master(MASTER_NAME).sentinel(SENTINEL_0).sentinel(SENTINEL_1); SENTINEL_CONFIG.setDatabase(5); } @@ -129,12 +132,12 @@ public class LettuceSentinelIntegrationTests extends AbstractConnectionIntegrati assertThat(servers.get(0).getName(), is(MASTER_NAME)); } - @Test // DATAREDIS-842 + @Test // DATAREDIS-842, DATAREDIS-973 public void shouldUseSpecifiedDatabase() { RedisConnection connection = connectionFactory.getConnection(); - connection.flushDb(); + connection.flushAll(); connection.set("foo".getBytes(), "bar".getBytes()); connection.close(); @@ -142,17 +145,45 @@ public class LettuceSentinelIntegrationTests extends AbstractConnectionIntegrati connectionFactory.setClientResources(LettuceTestClientResources.getSharedClientResources()); connectionFactory.setShutdownTimeout(0); connectionFactory.setShareNativeConnection(false); + connectionFactory.setDatabase(5); connectionFactory.afterPropertiesSet(); RedisConnection directConnection = connectionFactory.getConnection(); - assertThat(directConnection.exists("foo".getBytes()), is(false)); - directConnection.select(5); - assertThat(directConnection.exists("foo".getBytes()), is(true)); + directConnection.select(0); + + assertThat(directConnection.exists("foo".getBytes()), is(false)); directConnection.close(); connectionFactory.destroy(); } + @Test // DATAREDIS-973 + public void reactiveShouldUseSpecifiedDatabase() { + + RedisConnection connection = connectionFactory.getConnection(); + + connection.flushAll(); + connection.set("foo".getBytes(), "bar".getBytes()); + connection.close(); + + LettuceConnectionFactory connectionFactory = new LettuceConnectionFactory(); + connectionFactory.setClientResources(LettuceTestClientResources.getSharedClientResources()); + connectionFactory.setShutdownTimeout(0); + connectionFactory.setShareNativeConnection(false); + connectionFactory.setDatabase(5); + connectionFactory.afterPropertiesSet(); + + LettuceReactiveRedisConnection reactiveConnection = connectionFactory.getReactiveConnection(); + + reactiveConnection.keyCommands().exists(ByteBuffer.wrap("foo".getBytes())) // + .as(StepVerifier::create) // + .expectNext(true) // + .verifyComplete(); + + reactiveConnection.close(); + connectionFactory.destroy(); + } + @Test // DATAREDIS-348 public void shouldReadSlavesOfMastersCorrectly() { @@ -248,8 +279,7 @@ public class LettuceSentinelIntegrationTests extends AbstractConnectionIntegrati LettuceClientConfiguration configuration = LettuceTestClientConfiguration.builder().readFrom(ReadFrom.SLAVE) .build(); - LettuceConnectionFactory factory = new LettuceConnectionFactory(SENTINEL_CONFIG, - configuration); + LettuceConnectionFactory factory = new LettuceConnectionFactory(SENTINEL_CONFIG, configuration); factory.afterPropertiesSet(); RedisConnection connection = factory.getConnection();