diff --git a/src/main/java/redis/clients/jedis/JedisSentineled.java b/src/main/java/redis/clients/jedis/JedisSentineled.java new file mode 100644 index 0000000000..6d449e8d76 --- /dev/null +++ b/src/main/java/redis/clients/jedis/JedisSentineled.java @@ -0,0 +1,57 @@ +package redis.clients.jedis; + +import java.util.Set; +import org.apache.commons.pool2.impl.GenericObjectPoolConfig; +import redis.clients.jedis.providers.SentineledConnectionProvider; + +public class JedisSentineled extends UnifiedJedis { + + /** + * This constructor is here for easier transition from {@link JedisSentinelPool#JedisSentinelPool( + * java.lang.String, java.util.Set, redis.clients.jedis.JedisClientConfig, redis.clients.jedis.JedisClientConfig)}. + * + * @deprecated Use {@link #JedisSentineled(java.lang.String, redis.clients.jedis.JedisClientConfig, + * java.util.Set, redis.clients.jedis.JedisClientConfig)}. + */ + @Deprecated + // Legacy + public JedisSentineled(String masterName, Set sentinels, + final JedisClientConfig masterClientConfig, final JedisClientConfig sentinelClientConfig) { + this(masterName, masterClientConfig, sentinels, sentinelClientConfig); + } + + public JedisSentineled(String masterName, final JedisClientConfig masterClientConfig, + Set sentinels, final JedisClientConfig sentinelClientConfig) { + this(new SentineledConnectionProvider(masterName, masterClientConfig, sentinels, sentinelClientConfig)); + } + + /** + * This constructor is here for easier transition from {@link JedisSentinelPool#JedisSentinelPool( + * java.lang.String, java.util.Set, org.apache.commons.pool2.impl.GenericObjectPoolConfig, + * redis.clients.jedis.JedisClientConfig, redis.clients.jedis.JedisClientConfig)}. + * + * @deprecated Use {@link #JedisSentineled(java.lang.String, redis.clients.jedis.JedisClientConfig, + * org.apache.commons.pool2.impl.GenericObjectPoolConfig, java.util.Set, redis.clients.jedis.JedisClientConfig)}. + */ + @Deprecated + // Legacy + public JedisSentineled(String masterName, Set sentinels, + final GenericObjectPoolConfig poolConfig, final JedisClientConfig masterClientConfig, + final JedisClientConfig sentinelClientConfig) { + this(masterName, masterClientConfig, poolConfig, sentinels, sentinelClientConfig); + } + + public JedisSentineled(String masterName, final JedisClientConfig masterClientConfig, + final GenericObjectPoolConfig poolConfig, + Set sentinels, final JedisClientConfig sentinelClientConfig) { + this(new SentineledConnectionProvider(masterName, masterClientConfig, poolConfig, sentinels, sentinelClientConfig)); + } + + public JedisSentineled(SentineledConnectionProvider sentineledConnectionProvider) { + super(sentineledConnectionProvider); + } + + public HostAndPort getCurrentMaster() { + return ((SentineledConnectionProvider) provider).getCurrentMaster(); + } +} diff --git a/src/main/java/redis/clients/jedis/providers/SentineledConnectionProvider.java b/src/main/java/redis/clients/jedis/providers/SentineledConnectionProvider.java new file mode 100644 index 0000000000..5058f07179 --- /dev/null +++ b/src/main/java/redis/clients/jedis/providers/SentineledConnectionProvider.java @@ -0,0 +1,280 @@ +package redis.clients.jedis.providers; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.Set; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.apache.commons.pool2.impl.GenericObjectPoolConfig; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import redis.clients.jedis.CommandArguments; +import redis.clients.jedis.Connection; +import redis.clients.jedis.ConnectionPool; +import redis.clients.jedis.HostAndPort; +import redis.clients.jedis.Jedis; +import redis.clients.jedis.JedisClientConfig; +import redis.clients.jedis.JedisPubSub; +import redis.clients.jedis.exceptions.JedisConnectionException; +import redis.clients.jedis.exceptions.JedisException; +import redis.clients.jedis.util.IOUtils; + +public class SentineledConnectionProvider implements ConnectionProvider { + + private static final Logger LOG = LoggerFactory.getLogger(SentineledConnectionProvider.class); + + protected static final long DEFAULT_SUBSCRIBE_RETRY_WAIT_TIME_MILLIS = 5000; + + private volatile HostAndPort currentMaster; + + private volatile ConnectionPool pool; + + private final String masterName; + + private final JedisClientConfig masterClientConfig; + + private final GenericObjectPoolConfig masterPoolConfig; + + protected final Collection sentinelListeners = new ArrayList<>(); + + private final JedisClientConfig sentinelClientConfig; + + private final long subscribeRetryWaitTimeMillis; + + private final Object initPoolLock = new Object(); + + public SentineledConnectionProvider(String masterName, final JedisClientConfig masterClientConfig, + Set sentinels, final JedisClientConfig sentinelClientConfig) { + this(masterName, masterClientConfig, /*poolConfig*/ null, sentinels, sentinelClientConfig); + } + + public SentineledConnectionProvider(String masterName, final JedisClientConfig masterClientConfig, + final GenericObjectPoolConfig poolConfig, + Set sentinels, final JedisClientConfig sentinelClientConfig) { + this(masterName, masterClientConfig, poolConfig, sentinels, sentinelClientConfig, + DEFAULT_SUBSCRIBE_RETRY_WAIT_TIME_MILLIS); + } + + public SentineledConnectionProvider(String masterName, final JedisClientConfig masterClientConfig, + final GenericObjectPoolConfig poolConfig, + Set sentinels, final JedisClientConfig sentinelClientConfig, + final long subscribeRetryWaitTimeMillis) { + + this.masterName = masterName; + this.masterClientConfig = masterClientConfig; + this.masterPoolConfig = poolConfig; + + this.sentinelClientConfig = sentinelClientConfig; + this.subscribeRetryWaitTimeMillis = subscribeRetryWaitTimeMillis; + + HostAndPort master = initSentinels(sentinels); + initMaster(master); + } + + @Override + public Connection getConnection() { + return pool.getResource(); + } + + @Override + public Connection getConnection(CommandArguments args) { + return pool.getResource(); + } + + @Override + public void close() { + sentinelListeners.forEach(SentinelListener::shutdown); + + pool.close(); + } + + public HostAndPort getCurrentMaster() { + return currentMaster; + } + + private void initMaster(HostAndPort master) { + synchronized (initPoolLock) { + if (!master.equals(currentMaster)) { + currentMaster = master; + + ConnectionPool newPool = masterPoolConfig != null + ? new ConnectionPool(currentMaster, masterClientConfig, masterPoolConfig) + : new ConnectionPool(currentMaster, masterClientConfig); + + ConnectionPool existingPool = pool; + pool = newPool; + LOG.info("Created connection pool to master at {}.", master); + + if (existingPool != null) { + // although we clear the pool, we still have to check the returned object in getResource, + // this call only clears idle instances, not borrowed instances + // existingPool.clear(); // necessary?? + existingPool.close(); + } + } + } + } + + private HostAndPort initSentinels(Set sentinels) { + + HostAndPort master = null; + boolean sentinelAvailable = false; + + LOG.debug("Trying to find master from available sentinels..."); + + for (HostAndPort sentinel : sentinels) { + + LOG.debug("Connecting to Sentinel {}...", sentinel); + + try (Jedis jedis = new Jedis(sentinel, sentinelClientConfig)) { + + List masterAddr = jedis.sentinelGetMasterAddrByName(masterName); + + // connected to sentinel... + sentinelAvailable = true; + + if (masterAddr == null || masterAddr.size() != 2) { + LOG.warn("Sentinel {} is not monitoring master {}.", sentinel, masterName); + continue; + } + + master = toHostAndPort(masterAddr); + LOG.debug("Redis master reported at {}.", master); + break; + } catch (JedisException e) { + // resolves #1036, it should handle JedisException there's another chance + // of raising JedisDataException + LOG.warn("Could not get master address from {}.", sentinel, e); + } + } + + if (master == null) { + if (sentinelAvailable) { + // can connect to sentinel, but master name seems to not monitored + throw new JedisException( + "Can connect to sentinel, but " + masterName + " seems to be not monitored."); + } else { + throw new JedisConnectionException( + "All sentinels down, cannot determine where " + masterName + " is running."); + } + } + + LOG.info("Redis master running at {}. Starting sentinel listeners...", master); + + for (HostAndPort sentinel : sentinels) { + + SentinelListener listener = new SentinelListener(sentinel); + // whether SentinelListener threads are alive or not, process can be stopped + listener.setDaemon(true); + sentinelListeners.add(listener); + listener.start(); + } + + return master; + } + + /** + * Must be of size 2. + */ + private static HostAndPort toHostAndPort(List masterAddr) { + return toHostAndPort(masterAddr.get(0), masterAddr.get(1)); + } + + private static HostAndPort toHostAndPort(String hostStr, String portStr) { + return new HostAndPort(hostStr, Integer.parseInt(portStr)); + } + + protected class SentinelListener extends Thread { + + protected final HostAndPort node; + protected volatile Jedis sentinelJedis; + protected AtomicBoolean running = new AtomicBoolean(false); + + public SentinelListener(HostAndPort node) { + super(String.format("%s-SentinelListener-[%s]", masterName, node.toString())); + this.node = node; + } + + @Override + public void run() { + + running.set(true); + + while (running.get()) { + + try { + // double check that it is not being shutdown + if (!running.get()) { + break; + } + + sentinelJedis = new Jedis(node, sentinelClientConfig); + + // code for active refresh + List masterAddr = sentinelJedis.sentinelGetMasterAddrByName(masterName); + if (masterAddr == null || masterAddr.size() != 2) { + LOG.warn("Can not get master {} address. Sentinel: {}.", masterName, node); + } else { + initMaster(toHostAndPort(masterAddr)); + } + + 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); + } + } + }, "+switch-master"); + + } catch (JedisException e) { + + if (running.get()) { + LOG.error("Lost connection to sentinel {}. Sleeping {}ms and retrying.", node, + subscribeRetryWaitTimeMillis, e); + try { + Thread.sleep(subscribeRetryWaitTimeMillis); + } catch (InterruptedException se) { + LOG.error("Sleep interrupted.", se); + } + } else { + LOG.debug("Unsubscribing from sentinel {}.", node); + } + } finally { + IOUtils.closeQuietly(sentinelJedis); + } + } + } + + // must not throw exception + public void shutdown() { + try { + LOG.debug("Shutting down listener on {}.", node); + running.set(false); + // This isn't good, the Jedis object is not thread safe + if (sentinelJedis != null) { + sentinelJedis.close(); + } + } catch (RuntimeException e) { + LOG.error("Error while shutting down.", e); + } + } + } +} diff --git a/src/test/java/redis/clients/jedis/JedisSentinelPoolTest.java b/src/test/java/redis/clients/jedis/JedisSentinelPoolTest.java index 9174fd724a..1d52d14784 100644 --- a/src/test/java/redis/clients/jedis/JedisSentinelPoolTest.java +++ b/src/test/java/redis/clients/jedis/JedisSentinelPoolTest.java @@ -9,7 +9,6 @@ import java.util.Set; import org.apache.commons.pool2.impl.GenericObjectPoolConfig; -import org.junit.After; import org.junit.Before; import org.junit.Test; @@ -20,16 +19,13 @@ public class JedisSentinelPoolTest { private static final String MASTER_NAME = "mymaster"; - protected static HostAndPort master = HostAndPorts.getRedisServers().get(2); - protected static HostAndPort slave1 = HostAndPorts.getRedisServers().get(3); + //protected static HostAndPort master = HostAndPorts.getRedisServers().get(2); + //protected static HostAndPort slave1 = HostAndPorts.getRedisServers().get(3); - protected static HostAndPort sentinel1 = HostAndPorts.getSentinelServers().get(1); - protected static HostAndPort sentinel2 = HostAndPorts.getSentinelServers().get(3); + protected static final HostAndPort sentinel1 = HostAndPorts.getSentinelServers().get(1); + protected static final HostAndPort sentinel2 = HostAndPorts.getSentinelServers().get(3); - protected static Jedis sentinelJedis1; - protected static Jedis sentinelJedis2; - - protected Set sentinels = new HashSet<>(); + protected final Set sentinels = new HashSet<>(); @Before public void setUp() throws Exception { @@ -37,15 +33,6 @@ public void setUp() throws Exception { sentinels.add(sentinel1.toString()); sentinels.add(sentinel2.toString()); - - sentinelJedis1 = new Jedis(sentinel1); - sentinelJedis2 = new Jedis(sentinel2); - } - - @After - public void tearDown() throws Exception { - sentinelJedis1.close(); - sentinelJedis2.close(); } @Test @@ -196,58 +183,4 @@ public void testResetValidPassword() { } } } -// -// @Test -// public void ensureSafeTwiceFailover() throws InterruptedException { -// JedisSentinelPool pool = new JedisSentinelPool(MASTER_NAME, sentinels, -// new GenericObjectPoolConfig(), 1000, "foobared", 2, "twice-failover-client"); -// -// forceFailover(pool); -// // after failover sentinel needs a bit of time to stabilize before a new failover -// Thread.sleep(1000); -// forceFailover(pool); -// -// // you can test failover as much as possible -// } -// -// private void forceFailover(JedisSentinelPool pool) throws InterruptedException { -// HostAndPort oldMaster = pool.getCurrentHostMaster(); -// -// // jedis connection should be master -// Jedis beforeFailoverJedis = pool.getResource(); -// assertEquals("PONG", beforeFailoverJedis.ping()); -// -// waitForFailover(pool, oldMaster); -// -// Jedis afterFailoverJedis = pool.getResource(); -// assertEquals("PONG", afterFailoverJedis.ping()); -// assertEquals(2, afterFailoverJedis.getDB()); -// assertEquals("twice-failover-client", afterFailoverJedis.clientGetname()); -// -// // returning both connections to the pool should not throw -// beforeFailoverJedis.close(); -// afterFailoverJedis.close(); -// } -// -// private void waitForFailover(JedisSentinelPool pool, HostAndPort oldMaster) -// throws InterruptedException { -// HostAndPort newMaster = JedisSentinelTestUtil.waitForNewPromotedMaster(MASTER_NAME, -// sentinelJedis1, sentinelJedis2); -// -// waitForJedisSentinelPoolRecognizeNewMaster(pool, newMaster); -// } -// -// private void waitForJedisSentinelPoolRecognizeNewMaster(JedisSentinelPool pool, -// HostAndPort newMaster) throws InterruptedException { -// -// while (true) { -// HostAndPort currentHostMaster = pool.getCurrentHostMaster(); -// -// if (newMaster.equals(currentHostMaster)) break; -// -// // System.out.println("JedisSentinelPool's master is not yet changed, sleep..."); -// Thread.sleep(100); -// } -// } - } diff --git a/src/test/java/redis/clients/jedis/SentineledConnectionProviderTest.java b/src/test/java/redis/clients/jedis/SentineledConnectionProviderTest.java new file mode 100644 index 0000000000..97b2d96932 --- /dev/null +++ b/src/test/java/redis/clients/jedis/SentineledConnectionProviderTest.java @@ -0,0 +1,146 @@ +package redis.clients.jedis; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertSame; + +import java.util.HashSet; +import java.util.Set; + +import org.apache.commons.pool2.impl.GenericObjectPoolConfig; +import org.junit.Before; +import org.junit.Test; + +import redis.clients.jedis.exceptions.JedisConnectionException; +import redis.clients.jedis.exceptions.JedisException; +import redis.clients.jedis.providers.SentineledConnectionProvider; + +/** + * @see JedisSentinelPoolTest + */ +public class SentineledConnectionProviderTest { + + private static final String MASTER_NAME = "mymaster"; + + //protected static HostAndPort master = HostAndPorts.getRedisServers().get(2); + //protected static HostAndPort slave1 = HostAndPorts.getRedisServers().get(3); + + protected static final HostAndPort sentinel1 = HostAndPorts.getSentinelServers().get(1); + protected static final HostAndPort sentinel2 = HostAndPorts.getSentinelServers().get(3); + + protected Set sentinels = new HashSet<>(); + + @Before + public void setUp() throws Exception { + sentinels.clear(); + + sentinels.add(sentinel1); + sentinels.add(sentinel2); + } + + @Test + 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(), + sentinels, DefaultJedisClientConfig.builder().build())) { + + provider.getConnection().close(); + } + } + } + + @Test(expected = JedisConnectionException.class) + public void initializeWithNotAvailableSentinelsShouldThrowException() { + Set wrongSentinels = new HashSet<>(); + wrongSentinels.add(new HostAndPort("localhost", 65432)); + wrongSentinels.add(new HostAndPort("localhost", 65431)); + + try (SentineledConnectionProvider provider = new SentineledConnectionProvider(MASTER_NAME, + DefaultJedisClientConfig.builder().build(), wrongSentinels, DefaultJedisClientConfig.builder().build())) { + } + } + + @Test(expected = JedisException.class) + public void initializeWithNotMonitoredMasterNameShouldThrowException() { + final String wrongMasterName = "wrongMasterName"; + try (SentineledConnectionProvider provider = new SentineledConnectionProvider(wrongMasterName, + DefaultJedisClientConfig.builder().build(), sentinels, DefaultJedisClientConfig.builder().build())) { + } + } + + @Test + 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(), + config, sentinels, DefaultJedisClientConfig.builder().build())) { + assertSame(SentineledConnectionProvider.class, jedis.provider.getClass()); + jedis.set("foo", "bar"); + assertEquals("bar", jedis.get("foo")); + } + } + + @Test + public void checkResourceIsCloseable() { + GenericObjectPoolConfig config = new GenericObjectPoolConfig<>(); + config.setMaxTotal(1); + config.setBlockWhenExhausted(false); + JedisSentineled jedis = new JedisSentineled(MASTER_NAME, + DefaultJedisClientConfig.builder().timeoutMillis(1000).password("foobared").database(2).build(), + config, sentinels, DefaultJedisClientConfig.builder().build()); + + Connection conn = jedis.provider.getConnection(); + try { + conn.ping(); + } finally { + conn.close(); + } + + Connection conn2 = jedis.provider.getConnection(); + try { + assertEquals(conn, conn2); + } finally { + conn2.close(); + } + } +// +// @Test +// public void testResetInvalidPassword() { +// JedisFactory factory = new JedisFactory(null, 0, 2000, 2000, "foobared", 0, "my_shiny_client_name") { }; +// +// try (JedisSentinelPool pool = new JedisSentinelPool(MASTER_NAME, sentinels, new JedisPoolConfig(), factory)) { +// Jedis obj1_ref; +// try (Jedis obj1_1 = pool.getResource()) { +// obj1_ref = obj1_1; +// obj1_1.set("foo", "bar"); +// assertEquals("bar", obj1_1.get("foo")); +// } +// try (Jedis obj1_2 = pool.getResource()) { +// assertSame(obj1_ref, obj1_2); +// factory.setPassword("wrong password"); +// try (Jedis obj2 = pool.getResource()) { +// fail("Should not get resource from pool"); +// } catch (JedisException e) { } +// } +// } +// } +// +// @Test +// public void testResetValidPassword() { +// JedisFactory factory = new JedisFactory(null, 0, 2000, 2000, "wrong password", 0, "my_shiny_client_name") { }; +// +// try (JedisSentinelPool pool = new JedisSentinelPool(MASTER_NAME, sentinels, new JedisPoolConfig(), factory)) { +// try (Jedis obj1 = pool.getResource()) { +// fail("Should not get resource from pool"); +// } catch (JedisException e) { } +// +// factory.setPassword("foobared"); +// try (Jedis obj2 = pool.getResource()) { +// obj2.set("foo", "bar"); +// assertEquals("bar", obj2.get("foo")); +// } +// } +// } +}