diff --git a/pom.xml b/pom.xml index fd9b692e71..3b91194d41 100644 --- a/pom.xml +++ b/pom.xml @@ -214,6 +214,12 @@ ${resilience4j.version} true + + org.powermock + powermock-module-junit4 + 2.0.2 + test + io.github.resilience4j resilience4j-circuitbreaker @@ -391,7 +397,7 @@ @{failsafeSuffixArgLine} ${JVM_OPTS} - + **/*IntegrationTest.java **/*IntegrationTests.java @@ -494,6 +500,7 @@ **/Health*.java src/main/java/redis/clients/jedis/MultiClusterClientConfig.java src/main/java/redis/clients/jedis/HostAndPort.java + src/main/java/redis/clients/jedis/util/ReadOnlyCommands.java diff --git a/src/main/java/redis/clients/jedis/JedisClientConfig.java b/src/main/java/redis/clients/jedis/JedisClientConfig.java index ce7fd82de4..5548b36b64 100644 --- a/src/main/java/redis/clients/jedis/JedisClientConfig.java +++ b/src/main/java/redis/clients/jedis/JedisClientConfig.java @@ -6,9 +6,9 @@ import javax.net.ssl.SSLSocketFactory; import redis.clients.jedis.authentication.AuthXManager; +import redis.clients.jedis.util.ReadOnlyCommands; public interface JedisClientConfig { - default RedisProtocol getRedisProtocol() { return null; } diff --git a/src/main/java/redis/clients/jedis/JedisSentineled.java b/src/main/java/redis/clients/jedis/JedisSentineled.java index 26f208a03b..147ac351b5 100644 --- a/src/main/java/redis/clients/jedis/JedisSentineled.java +++ b/src/main/java/redis/clients/jedis/JedisSentineled.java @@ -7,6 +7,7 @@ import redis.clients.jedis.csc.CacheConfig; import redis.clients.jedis.csc.CacheFactory; import redis.clients.jedis.providers.SentineledConnectionProvider; +import redis.clients.jedis.util.ReadOnlyCommands; public class JedisSentineled extends UnifiedJedis { @@ -37,6 +38,21 @@ public JedisSentineled(String masterName, final JedisClientConfig masterClientCo masterClientConfig.getRedisProtocol()); } + public JedisSentineled(String masterName, final JedisClientConfig masterClientConfig, + final GenericObjectPoolConfig poolConfig, + Set sentinels, final JedisClientConfig sentinelClientConfig, ReadFrom readFrom) { + super(new SentineledConnectionProvider(masterName, masterClientConfig, poolConfig, sentinels, sentinelClientConfig, readFrom), + masterClientConfig.getRedisProtocol()); + } + + public JedisSentineled(String masterName, final JedisClientConfig masterClientConfig, + final GenericObjectPoolConfig poolConfig, + Set sentinels, final JedisClientConfig sentinelClientConfig, ReadFrom readFrom, + ReadOnlyCommands.ReadOnlyPredicate readOnlyPredicate) { + super(new SentineledConnectionProvider(masterName, masterClientConfig, poolConfig, sentinels, sentinelClientConfig, readFrom, readOnlyPredicate), + masterClientConfig.getRedisProtocol()); + } + @Experimental public JedisSentineled(String masterName, final JedisClientConfig masterClientConfig, Cache clientSideCache, final GenericObjectPoolConfig poolConfig, diff --git a/src/main/java/redis/clients/jedis/ReadFrom.java b/src/main/java/redis/clients/jedis/ReadFrom.java new file mode 100644 index 0000000000..5ac339009b --- /dev/null +++ b/src/main/java/redis/clients/jedis/ReadFrom.java @@ -0,0 +1,12 @@ +package redis.clients.jedis; + +public enum ReadFrom { + // read from the upstream only. + UPSTREAM, + // read from the replica only. + REPLICA, + // read preferred from the upstream and fall back to a replica if the upstream is not available. + UPSTREAM_PREFERRED, + // read preferred from replica and fall back to upstream if no replica is not available. + REPLICA_PREFERRED +} diff --git a/src/main/java/redis/clients/jedis/providers/SentineledConnectionProvider.java b/src/main/java/redis/clients/jedis/providers/SentineledConnectionProvider.java index dedf34fb69..e04a23b050 100644 --- a/src/main/java/redis/clients/jedis/providers/SentineledConnectionProvider.java +++ b/src/main/java/redis/clients/jedis/providers/SentineledConnectionProvider.java @@ -3,6 +3,7 @@ import java.util.ArrayList; import java.util.Collection; import java.util.List; +import java.util.Map; import java.util.Set; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.locks.Lock; @@ -19,13 +20,24 @@ import redis.clients.jedis.Jedis; import redis.clients.jedis.JedisClientConfig; import redis.clients.jedis.JedisPubSub; +import redis.clients.jedis.ReadFrom; import redis.clients.jedis.annots.Experimental; import redis.clients.jedis.csc.Cache; import redis.clients.jedis.exceptions.JedisConnectionException; import redis.clients.jedis.exceptions.JedisException; import redis.clients.jedis.util.IOUtils; +import redis.clients.jedis.util.ReadOnlyCommands; public class SentineledConnectionProvider implements ConnectionProvider { + class PoolInfo { + public String host; + public ConnectionPool pool; + + public PoolInfo(String host, ConnectionPool pool) { + this.host = host; + this.pool = pool; + } + } private static final Logger LOG = LoggerFactory.getLogger(SentineledConnectionProvider.class); @@ -49,8 +61,18 @@ public class SentineledConnectionProvider implements ConnectionProvider { private final long subscribeRetryWaitTimeMillis; + private final ReadFrom readFrom; + + private ReadOnlyCommands.ReadOnlyPredicate READ_ONLY_COMMANDS; + private final Lock initPoolLock = new ReentrantLock(true); + private final List slavePools = new ArrayList<>(); + + private final Lock slavePoolsLock = new ReentrantLock(true); + + private int poolIndex; + public SentineledConnectionProvider(String masterName, final JedisClientConfig masterClientConfig, Set sentinels, final JedisClientConfig sentinelClientConfig) { this(masterName, masterClientConfig, null, null, sentinels, sentinelClientConfig); @@ -69,26 +91,41 @@ public SentineledConnectionProvider(String masterName, final JedisClientConfig m DEFAULT_SUBSCRIBE_RETRY_WAIT_TIME_MILLIS); } + public SentineledConnectionProvider(String masterName, final JedisClientConfig masterClientConfig, + final GenericObjectPoolConfig poolConfig, + Set sentinels, final JedisClientConfig sentinelClientConfig, ReadFrom readFrom) { + this(masterName, masterClientConfig, null, poolConfig, sentinels, sentinelClientConfig, + DEFAULT_SUBSCRIBE_RETRY_WAIT_TIME_MILLIS, readFrom, ReadOnlyCommands.asPredicate()); + } + + public SentineledConnectionProvider(String masterName, final JedisClientConfig masterClientConfig, + final GenericObjectPoolConfig poolConfig, + Set sentinels, final JedisClientConfig sentinelClientConfig, ReadFrom readFrom, + ReadOnlyCommands.ReadOnlyPredicate readOnlyPredicate) { + this(masterName, masterClientConfig, null, poolConfig, sentinels, sentinelClientConfig, + DEFAULT_SUBSCRIBE_RETRY_WAIT_TIME_MILLIS, readFrom, readOnlyPredicate); + } + @Experimental public SentineledConnectionProvider(String masterName, final JedisClientConfig masterClientConfig, Cache clientSideCache, final GenericObjectPoolConfig poolConfig, Set sentinels, final JedisClientConfig sentinelClientConfig) { this(masterName, masterClientConfig, clientSideCache, poolConfig, sentinels, sentinelClientConfig, - DEFAULT_SUBSCRIBE_RETRY_WAIT_TIME_MILLIS); + DEFAULT_SUBSCRIBE_RETRY_WAIT_TIME_MILLIS, ReadFrom.UPSTREAM, ReadOnlyCommands.asPredicate()); } public SentineledConnectionProvider(String masterName, final JedisClientConfig masterClientConfig, final GenericObjectPoolConfig poolConfig, Set sentinels, final JedisClientConfig sentinelClientConfig, final long subscribeRetryWaitTimeMillis) { - this(masterName, masterClientConfig, null, poolConfig, sentinels, sentinelClientConfig, subscribeRetryWaitTimeMillis); + this(masterName, masterClientConfig, null, poolConfig, sentinels, sentinelClientConfig, subscribeRetryWaitTimeMillis, ReadFrom.UPSTREAM, ReadOnlyCommands.asPredicate()); } @Experimental public SentineledConnectionProvider(String masterName, final JedisClientConfig masterClientConfig, Cache clientSideCache, final GenericObjectPoolConfig poolConfig, Set sentinels, final JedisClientConfig sentinelClientConfig, - final long subscribeRetryWaitTimeMillis) { + final long subscribeRetryWaitTimeMillis, ReadFrom readFrom, ReadOnlyCommands.ReadOnlyPredicate readOnlyPredicate) { this.masterName = masterName; this.masterClientConfig = masterClientConfig; @@ -97,11 +134,49 @@ public SentineledConnectionProvider(String masterName, final JedisClientConfig m this.sentinelClientConfig = sentinelClientConfig; this.subscribeRetryWaitTimeMillis = subscribeRetryWaitTimeMillis; + this.readFrom = readFrom; + this.READ_ONLY_COMMANDS = readOnlyPredicate; HostAndPort master = initSentinels(sentinels); initMaster(master); } + private Connection getSlaveResource() { + int startIdx; + slavePoolsLock.lock(); + try { + poolIndex++; + if (poolIndex >= slavePools.size()) { + poolIndex = 0; + } + startIdx = poolIndex; + } finally { + slavePoolsLock.unlock(); + } + return _getSlaveResource(startIdx, 0); + } + + private Connection _getSlaveResource(int idx, int cnt) { + PoolInfo poolInfo; + slavePoolsLock.lock(); + try { + if (cnt >= slavePools.size()) { + return null; + } + poolInfo = slavePools.get(idx % slavePools.size()); + } finally { + slavePoolsLock.unlock(); + } + + try { + Connection jedis = poolInfo.pool.getResource(); + return jedis; + } catch (Exception e) { + LOG.error("get connection fail:", e); + return _getSlaveResource(idx + 1, cnt + 1); + } + } + @Override public Connection getConnection() { return pool.getResource(); @@ -109,7 +184,43 @@ public Connection getConnection() { @Override public Connection getConnection(CommandArguments args) { - return pool.getResource(); + boolean isReadCommand = READ_ONLY_COMMANDS.isReadOnly(args); + if (!isReadCommand) { + return pool.getResource(); + } + + Connection conn; + switch (readFrom) { + case REPLICA: + conn = getSlaveResource(); + if (conn == null) { + throw new JedisException("all replica is invalid"); + } + return conn; + case UPSTREAM_PREFERRED: + try { + conn = pool.getResource(); + if (conn != null) { + return conn; + } + } catch (Exception e) { + LOG.error("get master connection error", e); + } + + conn = getSlaveResource(); + if (conn == null) { + throw new JedisException("all redis instance is invalid"); + } + return conn; + case REPLICA_PREFERRED: + conn = getSlaveResource(); + if (conn != null) { + return conn; + } + return pool.getResource(); + default: + return pool.getResource(); + } } @Override @@ -117,6 +228,10 @@ public void close() { sentinelListeners.forEach(SentinelListener::shutdown); pool.close(); + + for (PoolInfo slavePool : slavePools) { + slavePool.pool.close(); + } } public HostAndPort getCurrentMaster() { @@ -167,6 +282,88 @@ private ConnectionPool createNodePool(HostAndPort master) { } } + private void initSlaves(List slaves) { + List removedSlavePools = new ArrayList<>(); + slavePoolsLock.lock(); + try { + for (int i = slavePools.size()-1; i >= 0; i--) { + PoolInfo poolInfo = slavePools.get(i); + boolean found = false; + for (HostAndPort slave : slaves) { + String host = slave.toString(); + if (poolInfo.host.equals(host)) { + found = true; + break; + } + } + if (!found) { + removedSlavePools.add(slavePools.remove(i)); + } + } + + for (HostAndPort slave : slaves) { + addSlave(slave); + } + } finally { + slavePoolsLock.unlock(); + if (!removedSlavePools.isEmpty() && clientSideCache != null) { + clientSideCache.flush(); + } + + for (PoolInfo removedSlavePool : removedSlavePools) { + removedSlavePool.pool.destroy(); + } + } + } + + private static boolean isHealthy(String flags) { + for (String flag : flags.split(",")) { + switch (flag.trim()) { + case "s_down": + case "o_down": + case "disconnected": + return false; + } + } + return true; + } + + private void addSlave(HostAndPort slave) { + String newSlaveHost = slave.toString(); + slavePoolsLock.lock(); + try { + for (int i = 0; i < this.slavePools.size(); i++) { + PoolInfo poolInfo = this.slavePools.get(i); + if (poolInfo.host.equals(newSlaveHost)) { + return; + } + } + slavePools.add(new PoolInfo(newSlaveHost, createNodePool(slave))); + } finally { + slavePoolsLock.unlock(); + } + } + + private void removeSlave(HostAndPort slave) { + String newSlaveHost = slave.toString(); + PoolInfo removed = null; + slavePoolsLock.lock(); + try { + for (int i = 0; i < this.slavePools.size(); i++) { + PoolInfo poolInfo = this.slavePools.get(i); + if (poolInfo.host.equals(newSlaveHost)) { + removed = slavePools.remove(i); + break; + } + } + } finally { + slavePoolsLock.unlock(); + } + if (removed != null) { + removed.pool.destroy(); + } + } + private HostAndPort initSentinels(Set sentinels) { HostAndPort master = null; @@ -262,6 +459,24 @@ public void run() { sentinelJedis = new Jedis(node, sentinelClientConfig); + List> slaveInfos = sentinelJedis.sentinelSlaves(masterName); + + List slaves = new ArrayList<>(); + + for (int i = 0; i < slaveInfos.size(); i++) { + Map slaveInfo = slaveInfos.get(i); + String flags = slaveInfo.get("flags"); + if (flags == null || !isHealthy(flags)) { + continue; + } + String ip = slaveInfo.get("ip"); + int port = Integer.parseInt(slaveInfo.get("port")); + HostAndPort slave = new HostAndPort(ip, port); + slaves.add(slave); + } + + initSlaves(slaves); + // code for active refresh List masterAddr = sentinelJedis.sentinelGetMasterAddrByName(masterName); if (masterAddr == null || masterAddr.size() != 2) { @@ -273,26 +488,63 @@ public void run() { sentinelJedis.subscribe(new JedisPubSub() { @Override public void onMessage(String channel, String message) { - LOG.debug("Sentinel {} published: {}.", node, message); - - String[] switchMasterMsg = message.split(" "); - - if (switchMasterMsg.length > 3) { - - if (masterName.equals(switchMasterMsg[0])) { - initMaster(toHostAndPort(switchMasterMsg[3], switchMasterMsg[4])); - } else { - LOG.debug( - "Ignoring message on +switch-master for master {}. Our master is {}.", - switchMasterMsg[0], masterName); - } - - } else { - LOG.error("Invalid message received on sentinel {} on channel +switch-master: {}.", - node, message); + LOG.debug("Sentinel {} with channel {} published: {}.", node, channel, message); + + String[] switchMsg = message.split(" "); + String slaveIp; + int slavePort; + switch (channel) { + case "+switch-master": + if (switchMsg.length > 3) { + if (masterName.equals(switchMsg[0])) { + initMaster(toHostAndPort(switchMsg[3], switchMsg[4])); + } else { + LOG.debug( + "Ignoring message on +switch-master for master {}. Our master is {}.", + switchMsg[0], masterName); + } + } else { + LOG.error("Invalid message received on sentinel {} on channel +switch-master: {}.", + node, message); + } + break; + case "+sdown": + if (switchMsg[0].equals("master")) { + return; + } + if (!masterName.equals(switchMsg[5])) { + return; + } + slaveIp = switchMsg[2]; + slavePort = Integer.parseInt(switchMsg[3]); + removeSlave(new HostAndPort(slaveIp, slavePort)); + break; + case "-sdown": + if (switchMsg.length < 6) { + return; + } + if (!masterName.equals(switchMsg[5])) { + return; + } + slaveIp = switchMsg[2]; + slavePort = Integer.parseInt(switchMsg[3]); + addSlave(new HostAndPort(slaveIp, slavePort)); + break; + case "+slave": + if (!masterName.equals(switchMsg[5])) { + return; + } + slaveIp = switchMsg[2]; + slavePort = Integer.parseInt(switchMsg[3]); + addSlave(new HostAndPort(slaveIp, slavePort)); + + String masterIp = switchMsg[6]; + int masterPort = Integer.parseInt(switchMsg[7]); + removeSlave(new HostAndPort(masterIp, masterPort)); + break; } } - }, "+switch-master"); + }, "+switch-master", "+sdown", "-sdown", "+slave"); } catch (JedisException e) { diff --git a/src/main/java/redis/clients/jedis/util/ReadOnlyCommands.java b/src/main/java/redis/clients/jedis/util/ReadOnlyCommands.java new file mode 100644 index 0000000000..7b488e1541 --- /dev/null +++ b/src/main/java/redis/clients/jedis/util/ReadOnlyCommands.java @@ -0,0 +1,105 @@ +package redis.clients.jedis.util; + +import redis.clients.jedis.CommandArguments; +import redis.clients.jedis.Protocol.Command; +import redis.clients.jedis.bloom.RedisBloomProtocol.BloomFilterCommand; +import redis.clients.jedis.bloom.RedisBloomProtocol.CountMinSketchCommand; +import redis.clients.jedis.bloom.RedisBloomProtocol.CuckooFilterCommand; +import redis.clients.jedis.bloom.RedisBloomProtocol.TDigestCommand; +import redis.clients.jedis.bloom.RedisBloomProtocol.TopKCommand; +import redis.clients.jedis.commands.ProtocolCommand; +import redis.clients.jedis.json.JsonProtocol.JsonCommand; +import redis.clients.jedis.search.SearchProtocol.SearchCommand; +import redis.clients.jedis.timeseries.TimeSeriesProtocol.TimeSeriesCommand; + +import java.util.Arrays; +import java.util.HashSet; +import java.util.Set; + +public class ReadOnlyCommands { + + private static final ReadOnlyPredicate PREDICATE = command -> isReadOnlyCommand(command); + + private static final Set READ_ONLY_COMMANDS = new HashSet( + Arrays.asList( + // string + Command.PING, Command.AUTH, Command.HELLO, Command.GET, Command.EXISTS, Command.TYPE, + Command.KEYS, Command.RANDOMKEY, Command.DUMP, Command.DBSIZE, Command.SELECT, Command.ECHO, + Command.EXPIRETIME, Command.PEXPIRETIME, Command.TTL, Command.PTTL, Command.SORT_RO, + Command.INFO, Command.MONITOR, Command.LCS, Command.MGET, Command.STRLEN, Command.SUBSTR, + // bit + Command.GETBIT, Command.BITPOS, Command.GETRANGE, Command.BITCOUNT, Command.BITFIELD_RO, + // hash + Command.HGET, Command.HMGET, Command.HEXISTS, Command.HLEN, Command.HKEYS, Command.HVALS, + Command.HGETALL, Command.HSTRLEN, Command.HTTL, Command.HPTTL, Command.HEXPIRETIME, + Command.HPEXPIRETIME, Command.HRANDFIELD, + // list + Command.LLEN, Command.LRANGE, Command.LINDEX, Command.LPOS, + // set + Command.SMEMBERS, Command.SCARD, Command.SRANDMEMBER, Command.SINTER, Command.SUNION, + Command.SDIFF, Command.SISMEMBER, Command.SMISMEMBER, Command.SINTERCARD, + // zset + Command.ZDIFF, Command.ZRANGE, Command.ZRANK, Command.ZREVRANK, Command.ZREVRANGE, + Command.ZRANDMEMBER, Command.ZCARD, Command.ZSCORE, Command.ZCOUNT, Command.ZUNION, + Command.ZINTER, Command.ZRANGEBYSCORE, Command.ZREVRANGEBYSCORE, Command.ZLEXCOUNT, + Command.ZRANGEBYLEX, Command.ZREVRANGEBYLEX, Command.ZMSCORE, Command.ZINTERCARD, + // geo + Command.GEODIST, Command.GEOHASH, Command.GEOPOS, Command.GEORADIUS_RO, + Command.GEORADIUSBYMEMBER_RO, + // hyper log + Command.PFCOUNT, + // stream + Command.XLEN, Command.XRANGE, Command.XREVRANGE, Command.XREAD, Command.XREADGROUP, + Command.XPENDING, Command.XINFO, + // program + Command.FCALL_RO, + // vector set + Command.LASTSAVE, Command.ROLE, Command.OBJECT, Command.TIME, Command.SCAN, Command.HSCAN, + Command.SSCAN, Command.ZSCAN, Command.LOLWUT, Command.VSIM, Command.VDIM, Command.VCARD, + Command.VEMB, Command.VLINKS, Command.VRANDMEMBER, Command.VGETATTR, Command.VINFO, + // BloomFilterCommand + BloomFilterCommand.EXISTS, BloomFilterCommand.MEXISTS, BloomFilterCommand.CARD, + BloomFilterCommand.INFO, + // CuckooFilterCommand + CuckooFilterCommand.EXISTS, CuckooFilterCommand.MEXISTS, CuckooFilterCommand.COUNT, + CuckooFilterCommand.INFO, + // CountMinSketchCommand + CountMinSketchCommand.QUERY, CountMinSketchCommand.INFO, + // TopKCommand + TopKCommand.QUERY, TopKCommand.LIST, TopKCommand.INFO, + // TDigestCommand + TDigestCommand.INFO, TDigestCommand.CDF, TDigestCommand.QUANTILE, TDigestCommand.MIN, + TDigestCommand.MAX, TDigestCommand.TRIMMED_MEAN, TDigestCommand.RANK, + TDigestCommand.REVRANK, TDigestCommand.BYRANK, TDigestCommand.BYREVRANK, + // JsonCommand + JsonCommand.GET, JsonCommand.MGET, JsonCommand.TYPE, JsonCommand.STRLEN, + JsonCommand.ARRINDEX, JsonCommand.ARRLEN, JsonCommand.OBJKEYS, JsonCommand.OBJLEN, + JsonCommand.DEBUG, JsonCommand.RESP, + // SearchCommand + SearchCommand.INFO, SearchCommand.SEARCH, SearchCommand.EXPLAIN, SearchCommand.EXPLAINCLI, + SearchCommand.AGGREGATE, SearchCommand.CURSOR, SearchCommand.SYNDUMP, SearchCommand.SUGGET, + SearchCommand.SUGLEN, SearchCommand.DICTDUMP, SearchCommand.SPELLCHECK, + SearchCommand.TAGVALS, SearchCommand.PROFILE, SearchCommand._LIST, + // TimeSeriesCommand + TimeSeriesCommand.RANGE, TimeSeriesCommand.REVRANGE, TimeSeriesCommand.MRANGE, + TimeSeriesCommand.MREVRANGE, TimeSeriesCommand.INFO, TimeSeriesCommand.GET, + TimeSeriesCommand.MGET, TimeSeriesCommand.QUERYINDEX)); + + public static ReadOnlyPredicate asPredicate() { + return PREDICATE; + } + + public static boolean isReadOnlyCommand(CommandArguments args) { + return READ_ONLY_COMMANDS.contains(args.getCommand()); + } + + @FunctionalInterface + public interface ReadOnlyPredicate { + + /** + * @param command the input command. + * @return {@code true} if the input argument matches the predicate, otherwise {@code false} + */ + boolean isReadOnly(CommandArguments command); + } +} diff --git a/src/test/java/redis/clients/jedis/SentineledConnectionProviderTest.java b/src/test/java/redis/clients/jedis/SentineledConnectionProviderTest.java index e75b893fce..ace96dcbd9 100644 --- a/src/test/java/redis/clients/jedis/SentineledConnectionProviderTest.java +++ b/src/test/java/redis/clients/jedis/SentineledConnectionProviderTest.java @@ -1,5 +1,6 @@ package redis.clients.jedis; +import java.util.ArrayList; import java.util.HashSet; import java.util.Set; @@ -7,11 +8,16 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.junit.runner.RunWith; +import org.powermock.core.classloader.annotations.PrepareForTest; +import org.powermock.modules.junit4.PowerMockRunner; +import org.powermock.reflect.Whitebox; import org.junit.jupiter.api.Tag; import redis.clients.jedis.exceptions.JedisConnectionException; import redis.clients.jedis.exceptions.JedisException; import redis.clients.jedis.providers.SentineledConnectionProvider; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertThrows; @@ -21,6 +27,8 @@ * @see JedisSentinelPoolTest */ @Tag("integration") +@RunWith(PowerMockRunner.class) +@PrepareForTest({SentineledConnectionProvider.class}) public class SentineledConnectionProviderTest { private static final String MASTER_NAME = "mymaster"; @@ -30,6 +38,8 @@ public class SentineledConnectionProviderTest { protected Set sentinels = new HashSet<>(); + protected String password = "foobared"; + @BeforeEach public void setUp() throws Exception { sentinels.clear(); @@ -43,7 +53,7 @@ public void repeatedSentinelPoolInitialization() { for (int i = 0; i < 20; ++i) { try (SentineledConnectionProvider provider = new SentineledConnectionProvider(MASTER_NAME, - DefaultJedisClientConfig.builder().timeoutMillis(1000).password("foobared").database(2).build(), + DefaultJedisClientConfig.builder().timeoutMillis(1000).password(password).database(2).build(), sentinels, DefaultJedisClientConfig.builder().build())) { provider.getConnection().close(); @@ -78,7 +88,7 @@ public void checkCloseableConnections() throws Exception { GenericObjectPoolConfig config = new GenericObjectPoolConfig<>(); try (JedisSentineled jedis = new JedisSentineled(MASTER_NAME, - DefaultJedisClientConfig.builder().timeoutMillis(1000).password("foobared").database(2).build(), + DefaultJedisClientConfig.builder().timeoutMillis(1000).password(password).database(2).build(), config, sentinels, DefaultJedisClientConfig.builder().build())) { assertSame(SentineledConnectionProvider.class, jedis.provider.getClass()); jedis.set("foo", "bar"); @@ -93,7 +103,7 @@ public void checkResourceIsCloseable() { config.setBlockWhenExhausted(false); try (JedisSentineled jedis = new JedisSentineled(MASTER_NAME, - DefaultJedisClientConfig.builder().timeoutMillis(1000).password("foobared").database(2).build(), + DefaultJedisClientConfig.builder().timeoutMillis(1000).password(password).database(2).build(), config, sentinels, DefaultJedisClientConfig.builder().build())) { Connection conn = jedis.provider.getConnection(); @@ -115,7 +125,7 @@ public void checkResourceIsCloseable() { @Test public void testResetInvalidPassword() { DefaultRedisCredentialsProvider credentialsProvider - = new DefaultRedisCredentialsProvider(new DefaultRedisCredentials(null, "foobared")); + = new DefaultRedisCredentialsProvider(new DefaultRedisCredentials(null, password)); try (JedisSentineled jedis = new JedisSentineled(MASTER_NAME, DefaultJedisClientConfig.builder() .timeoutMillis(2000).credentialsProvider(credentialsProvider).database(2) @@ -156,7 +166,7 @@ public void testResetValidPassword() { fail("Should not get resource from pool"); } catch (JedisException e) { } - credentialsProvider.setCredentials(new DefaultRedisCredentials(null, "foobared")); + credentialsProvider.setCredentials(new DefaultRedisCredentials(null, password)); try (Connection conn2 = jedis.provider.getConnection()) { new Jedis(conn2).set("foo", "bar"); @@ -164,4 +174,74 @@ public void testResetValidPassword() { } } } + + @Test + public void testReadWriteSeparation() throws InterruptedException { + DefaultRedisCredentialsProvider credentialsProvider + = new DefaultRedisCredentialsProvider(new DefaultRedisCredentials(null, password)); + + try (JedisSentineled jedis = new JedisSentineled(MASTER_NAME, DefaultJedisClientConfig.builder() + .timeoutMillis(2000).credentialsProvider(credentialsProvider).database(2) + .clientName("my_shiny_client_name").build(), new ConnectionPoolConfig(), + sentinels, DefaultJedisClientConfig.builder().build())) { + + jedis.set("foo", "bar"); + Thread.sleep(1000); + assertEquals("bar", jedis.get("foo")); + } + } + + @Test + public void testReadFromREPLICAAndNoSlave() throws InterruptedException { + DefaultRedisCredentialsProvider credentialsProvider + = new DefaultRedisCredentialsProvider(new DefaultRedisCredentials(null, password)); + + try (JedisSentineled jedis = new JedisSentineled(MASTER_NAME, DefaultJedisClientConfig.builder() + .timeoutMillis(2000).credentialsProvider(credentialsProvider).database(2) + .clientName("my_shiny_client_name").build(), new ConnectionPoolConfig(), + sentinels, DefaultJedisClientConfig.builder().build(), ReadFrom.REPLICA)) { + + Thread.sleep(1000); + Whitebox.setInternalState(jedis.provider, "slavePools", new ArrayList<>()); + jedis.set("foo", "bar"); + Thread.sleep(1000); + assertThrows(JedisException.class, () -> jedis.get("foo")); + } + } + + @Test + public void testFallbackTOMasterWhenNOSlave() throws InterruptedException { + DefaultRedisCredentialsProvider credentialsProvider + = new DefaultRedisCredentialsProvider(new DefaultRedisCredentials(null, password)); + + try (JedisSentineled jedis = new JedisSentineled(MASTER_NAME, DefaultJedisClientConfig.builder() + .timeoutMillis(2000).credentialsProvider(credentialsProvider).database(2) + .clientName("my_shiny_client_name").build(), new ConnectionPoolConfig(), + sentinels, DefaultJedisClientConfig.builder().build(), ReadFrom.REPLICA_PREFERRED)) { + + Thread.sleep(1000); + Whitebox.setInternalState(jedis.provider, "slavePools", new ArrayList<>()); + jedis.set("foo", "bar"); + Thread.sleep(1000); + assertDoesNotThrow(() -> jedis.get("foo")); + } + } + + @Test + public void testAllWriteCommandsWhenNOSlave() throws InterruptedException { + DefaultRedisCredentialsProvider credentialsProvider + = new DefaultRedisCredentialsProvider(new DefaultRedisCredentials(null, password)); + + try (JedisSentineled jedis = new JedisSentineled(MASTER_NAME, DefaultJedisClientConfig.builder() + .timeoutMillis(2000).credentialsProvider(credentialsProvider).database(2) + .clientName("my_shiny_client_name").build(), new ConnectionPoolConfig(), + sentinels, DefaultJedisClientConfig.builder().build(), ReadFrom.REPLICA_PREFERRED, command -> false)) { + + Thread.sleep(1000); + Whitebox.setInternalState(jedis.provider, "slavePools", new ArrayList<>()); + jedis.set("foo", "bar"); + Thread.sleep(1000); + assertDoesNotThrow(() -> jedis.get("foo")); + } + } } diff --git a/src/test/java/redis/clients/jedis/csc/UnifiedJedisClientSideCacheTestBase.java b/src/test/java/redis/clients/jedis/csc/UnifiedJedisClientSideCacheTestBase.java index fa4043799e..87e7940471 100644 --- a/src/test/java/redis/clients/jedis/csc/UnifiedJedisClientSideCacheTestBase.java +++ b/src/test/java/redis/clients/jedis/csc/UnifiedJedisClientSideCacheTestBase.java @@ -145,7 +145,7 @@ public void immutableCacheEntriesTest() { } @Test - public void invalidationTest() { + public void invalidationTest() throws InterruptedException { try (UnifiedJedis jedis = createCachedJedis(CacheConfig.builder().build())) { Cache cache = jedis.getCache(); jedis.set("{csc}1", "one"); @@ -161,6 +161,7 @@ public void invalidationTest() { assertEquals(0, cache.getStats().getInvalidationCount()); jedis.set("{csc}1", "new-one"); + Thread.sleep(1000); List reply2 = jedis.mget("{csc}1", "{csc}2", "{csc}3"); assertEquals(Arrays.asList("new-one", "two", "three"), reply2);