Merge with master
This commit is contained in:
@@ -23,7 +23,6 @@ public class HostAndPortUtil {
|
||||
sentinelHostAndPortList.add(new HostAndPort("localhost", Protocol.DEFAULT_SENTINEL_PORT));
|
||||
sentinelHostAndPortList.add(new HostAndPort("localhost", Protocol.DEFAULT_SENTINEL_PORT + 1));
|
||||
sentinelHostAndPortList.add(new HostAndPort("localhost", Protocol.DEFAULT_SENTINEL_PORT + 2));
|
||||
sentinelHostAndPortList.add(new HostAndPort("localhost", Protocol.DEFAULT_SENTINEL_PORT + 3));
|
||||
|
||||
clusterHostAndPortList.add(new HostAndPort("localhost", 7379));
|
||||
clusterHostAndPortList.add(new HostAndPort("localhost", 7380));
|
||||
|
||||
@@ -2,18 +2,17 @@ package redis.clients.jedis.tests;
|
||||
|
||||
import java.util.HashSet;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
|
||||
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
|
||||
import redis.clients.jedis.DebugParams;
|
||||
import redis.clients.jedis.HostAndPort;
|
||||
import redis.clients.jedis.Jedis;
|
||||
import redis.clients.jedis.JedisPubSub;
|
||||
import redis.clients.jedis.JedisSentinelPool;
|
||||
import redis.clients.jedis.Transaction;
|
||||
import redis.clients.jedis.exceptions.JedisConnectionException;
|
||||
import redis.clients.jedis.tests.utils.JedisSentinelTestUtil;
|
||||
|
||||
public class JedisSentinelPoolTest extends JedisTestBase {
|
||||
private static final String MASTER_NAME = "mymaster";
|
||||
@@ -22,12 +21,8 @@ public class JedisSentinelPoolTest extends JedisTestBase {
|
||||
.get(2);
|
||||
protected static HostAndPort slave1 = HostAndPortUtil.getRedisServers()
|
||||
.get(3);
|
||||
protected static HostAndPort slave2 = HostAndPortUtil.getRedisServers()
|
||||
.get(4);
|
||||
protected static HostAndPort sentinel1 = HostAndPortUtil
|
||||
.getSentinelServers().get(1);
|
||||
protected static HostAndPort sentinel2 = HostAndPortUtil
|
||||
.getSentinelServers().get(2);
|
||||
|
||||
protected static Jedis sentinelJedis1;
|
||||
|
||||
@@ -36,7 +31,6 @@ public class JedisSentinelPoolTest extends JedisTestBase {
|
||||
@Before
|
||||
public void setUp() throws Exception {
|
||||
sentinels.add(sentinel1.toString());
|
||||
sentinels.add(sentinel2.toString());
|
||||
|
||||
sentinelJedis1 = new Jedis(sentinel1.getHost(), sentinel1.getPort());
|
||||
}
|
||||
@@ -46,17 +40,47 @@ public class JedisSentinelPoolTest extends JedisTestBase {
|
||||
JedisSentinelPool pool = new JedisSentinelPool(MASTER_NAME, sentinels,
|
||||
new GenericObjectPoolConfig(), 1000, "foobared", 2);
|
||||
|
||||
// perform failover
|
||||
doSegFaultMaster(pool);
|
||||
|
||||
// perform failover once again
|
||||
doSegFaultMaster(pool);
|
||||
forceFailover(pool);
|
||||
forceFailover(pool);
|
||||
|
||||
// you can test failover as much as possible
|
||||
// but you need to prepare additional slave per failover
|
||||
}
|
||||
|
||||
@Test
|
||||
public void returnResourceShouldResetState() {
|
||||
GenericObjectPoolConfig config = new GenericObjectPoolConfig();
|
||||
config.setMaxTotal(1);
|
||||
config.setBlockWhenExhausted(false);
|
||||
JedisSentinelPool pool = new JedisSentinelPool(MASTER_NAME, sentinels,
|
||||
config, 1000, "foobared", 2);
|
||||
|
||||
Jedis jedis = pool.getResource();
|
||||
Jedis jedis2 = null;
|
||||
|
||||
try {
|
||||
jedis.set("hello", "jedis");
|
||||
Transaction t = jedis.multi();
|
||||
t.set("hello", "world");
|
||||
pool.returnResource(jedis);
|
||||
|
||||
jedis2 = pool.getResource();
|
||||
|
||||
assertTrue(jedis == jedis2);
|
||||
assertEquals("jedis", jedis2.get("hello"));
|
||||
} catch (JedisConnectionException e) {
|
||||
if (jedis2 != null) {
|
||||
pool.returnBrokenResource(jedis2);
|
||||
jedis2 = null;
|
||||
}
|
||||
} finally {
|
||||
if (jedis2 != null)
|
||||
pool.returnResource(jedis2);
|
||||
|
||||
pool.destroy();
|
||||
}
|
||||
}
|
||||
|
||||
private void doSegFaultMaster(JedisSentinelPool pool)
|
||||
private void forceFailover(JedisSentinelPool pool)
|
||||
throws InterruptedException {
|
||||
HostAndPort oldMaster = pool.getCurrentHostMaster();
|
||||
|
||||
@@ -64,14 +88,16 @@ public class JedisSentinelPoolTest extends JedisTestBase {
|
||||
Jedis jedis = pool.getResource();
|
||||
assertEquals("PONG", jedis.ping());
|
||||
|
||||
try {
|
||||
jedis.debug(DebugParams.SEGFAULT());
|
||||
} catch (Exception e) {
|
||||
}
|
||||
|
||||
// It can throw JedisDataException while there's no slave to promote
|
||||
// There's nothing we can do, so we just pass Exception to make test
|
||||
// fail fast
|
||||
sentinelJedis1.sentinelFailover(MASTER_NAME);
|
||||
|
||||
waitForFailover(pool, oldMaster);
|
||||
// JedisSentinelPool recognize master but may not changed internal pool
|
||||
// yet
|
||||
Thread.sleep(100);
|
||||
|
||||
|
||||
jedis = pool.getResource();
|
||||
assertEquals("PONG", jedis.ping());
|
||||
assertEquals("foobared", jedis.configGet("requirepass").get(1));
|
||||
@@ -80,62 +106,15 @@ public class JedisSentinelPoolTest extends JedisTestBase {
|
||||
|
||||
private void waitForFailover(JedisSentinelPool pool, HostAndPort oldMaster)
|
||||
throws InterruptedException {
|
||||
waitForJedisSentinelPoolRecognizeNewMaster(pool);
|
||||
HostAndPort newMaster = JedisSentinelTestUtil
|
||||
.waitForNewPromotedMaster(sentinelJedis1);
|
||||
|
||||
waitForJedisSentinelPoolRecognizeNewMaster(pool, newMaster);
|
||||
}
|
||||
|
||||
private void waitForJedisSentinelPoolRecognizeNewMaster(
|
||||
JedisSentinelPool pool) throws InterruptedException {
|
||||
|
||||
final AtomicReference<String> newmaster = new AtomicReference<String>(
|
||||
"");
|
||||
|
||||
sentinelJedis1.psubscribe(new JedisPubSub() {
|
||||
|
||||
@Override
|
||||
public void onMessage(String channel, String message) {
|
||||
// TODO Auto-generated method stub
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onPMessage(String pattern, String channel,
|
||||
String message) {
|
||||
if (channel.equals("+switch-master")) {
|
||||
newmaster.set(message);
|
||||
punsubscribe();
|
||||
}
|
||||
// TODO Auto-generated method stub
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onSubscribe(String channel, int subscribedChannels) {
|
||||
// TODO Auto-generated method stub
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onUnsubscribe(String channel, int subscribedChannels) {
|
||||
// TODO Auto-generated method stub
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onPUnsubscribe(String pattern, int subscribedChannels) {
|
||||
// TODO Auto-generated method stub
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onPSubscribe(String pattern, int subscribedChannels) {
|
||||
// TODO Auto-generated method stub
|
||||
|
||||
}
|
||||
}, "*");
|
||||
|
||||
String[] chunks = newmaster.get().split(" ");
|
||||
HostAndPort newMaster = new HostAndPort(chunks[3],
|
||||
Integer.parseInt(chunks[4]));
|
||||
JedisSentinelPool pool, HostAndPort newMaster)
|
||||
throws InterruptedException {
|
||||
|
||||
while (true) {
|
||||
String host = pool.getCurrentHostMaster().getHost();
|
||||
@@ -150,25 +129,5 @@ public class JedisSentinelPoolTest extends JedisTestBase {
|
||||
Thread.sleep(100);
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void returnResourceShouldResetState() {
|
||||
GenericObjectPoolConfig config = new GenericObjectPoolConfig();
|
||||
config.setMaxTotal(1);
|
||||
config.setBlockWhenExhausted(false);
|
||||
JedisSentinelPool pool = new JedisSentinelPool(MASTER_NAME, sentinels,
|
||||
config, 1000, "foobared", 2);
|
||||
|
||||
Jedis jedis = pool.getResource();
|
||||
jedis.set("hello", "jedis");
|
||||
Transaction t = jedis.multi();
|
||||
t.set("hello", "world");
|
||||
pool.returnResource(jedis);
|
||||
|
||||
Jedis jedis2 = pool.getResource();
|
||||
assertTrue(jedis == jedis2);
|
||||
assertEquals("jedis", jedis2.get("hello"));
|
||||
pool.returnResource(jedis2);
|
||||
pool.destroy();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -22,15 +22,15 @@ public class JedisSentinelTest extends JedisTestBase {
|
||||
|
||||
protected static HostAndPort master = HostAndPortUtil.getRedisServers()
|
||||
.get(0);
|
||||
protected static HostAndPort slave = HostAndPortUtil.getRedisServers().get(
|
||||
5);
|
||||
protected static HostAndPort slave = HostAndPortUtil.getRedisServers()
|
||||
.get(4);
|
||||
protected static HostAndPort sentinel = HostAndPortUtil
|
||||
.getSentinelServers().get(0);
|
||||
|
||||
protected static HostAndPort sentinelForFailover = HostAndPortUtil
|
||||
.getSentinelServers().get(3);
|
||||
.getSentinelServers().get(2);
|
||||
protected static HostAndPort masterForFailover = HostAndPortUtil
|
||||
.getRedisServers().get(6);
|
||||
.getRedisServers().get(5);
|
||||
|
||||
@Before
|
||||
public void setup() throws InterruptedException {
|
||||
@@ -50,30 +50,35 @@ public class JedisSentinelTest extends JedisTestBase {
|
||||
@Test
|
||||
public void sentinel() {
|
||||
Jedis j = new Jedis(sentinel.getHost(), sentinel.getPort());
|
||||
List<Map<String, String>> masters = j.sentinelMasters();
|
||||
|
||||
boolean inMasters = false;
|
||||
for (Map<String, String> master : masters)
|
||||
if (MASTER_NAME.equals(master.get("name")))
|
||||
inMasters = true;
|
||||
try {
|
||||
List<Map<String, String>> masters = j.sentinelMasters();
|
||||
|
||||
assertTrue(inMasters);
|
||||
boolean inMasters = false;
|
||||
for (Map<String, String> master : masters)
|
||||
if (MASTER_NAME.equals(master.get("name")))
|
||||
inMasters = true;
|
||||
|
||||
List<String> masterHostAndPort = j
|
||||
.sentinelGetMasterAddrByName(MASTER_NAME);
|
||||
HostAndPort masterFromSentinel = new HostAndPort(
|
||||
masterHostAndPort.get(0), Integer.parseInt(masterHostAndPort
|
||||
.get(1)));
|
||||
assertEquals(master, masterFromSentinel);
|
||||
assertTrue(inMasters);
|
||||
|
||||
List<Map<String, String>> slaves = j.sentinelSlaves(MASTER_NAME);
|
||||
assertTrue(slaves.size() > 0);
|
||||
assertEquals(master.getPort(),
|
||||
Integer.parseInt(slaves.get(0).get("master-port")));
|
||||
List<String> masterHostAndPort = j
|
||||
.sentinelGetMasterAddrByName(MASTER_NAME);
|
||||
HostAndPort masterFromSentinel = new HostAndPort(
|
||||
masterHostAndPort.get(0),
|
||||
Integer.parseInt(masterHostAndPort.get(1)));
|
||||
assertEquals(master, masterFromSentinel);
|
||||
|
||||
// DO NOT RE-RUN TEST TOO FAST, RESET TAKES SOME TIME TO... RESET
|
||||
assertEquals(Long.valueOf(1), j.sentinelReset(MASTER_NAME));
|
||||
assertEquals(Long.valueOf(0), j.sentinelReset("woof" + MASTER_NAME));
|
||||
List<Map<String, String>> slaves = j.sentinelSlaves(MASTER_NAME);
|
||||
assertTrue(slaves.size() > 0);
|
||||
assertEquals(master.getPort(),
|
||||
Integer.parseInt(slaves.get(0).get("master-port")));
|
||||
|
||||
// DO NOT RE-RUN TEST TOO FAST, RESET TAKES SOME TIME TO... RESET
|
||||
assertEquals(Long.valueOf(1), j.sentinelReset(MASTER_NAME));
|
||||
assertEquals(Long.valueOf(0), j.sentinelReset("woof" + MASTER_NAME));
|
||||
} finally {
|
||||
j.close();
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
@@ -81,40 +86,48 @@ public class JedisSentinelTest extends JedisTestBase {
|
||||
Jedis j = new Jedis(sentinelForFailover.getHost(),
|
||||
sentinelForFailover.getPort());
|
||||
|
||||
HostAndPort currentMaster = new HostAndPort(
|
||||
masterForFailover.getHost(), masterForFailover.getPort());
|
||||
try {
|
||||
List<String> masterHostAndPort = j
|
||||
.sentinelGetMasterAddrByName(FAILOVER_MASTER_NAME);
|
||||
HostAndPort currentMaster = new HostAndPort(masterHostAndPort.get(0),
|
||||
Integer.parseInt(masterHostAndPort.get(1)));
|
||||
String result = j.sentinelFailover(FAILOVER_MASTER_NAME);
|
||||
assertEquals("OK", result);
|
||||
|
||||
List<String> masterHostAndPort = j
|
||||
.sentinelGetMasterAddrByName(FAILOVER_MASTER_NAME);
|
||||
String result = j.sentinelFailover(FAILOVER_MASTER_NAME);
|
||||
assertEquals("OK", result);
|
||||
JedisSentinelTestUtil.waitForNewPromotedMaster(j);
|
||||
|
||||
JedisSentinelTestUtil.waitForNewPromotedMaster(sentinelForFailover,
|
||||
FAILOVER_MASTER_NAME, currentMaster);
|
||||
masterHostAndPort = j
|
||||
.sentinelGetMasterAddrByName(FAILOVER_MASTER_NAME);
|
||||
HostAndPort newMaster = new HostAndPort(masterHostAndPort.get(0),
|
||||
Integer.parseInt(masterHostAndPort.get(1)));
|
||||
|
||||
masterHostAndPort = j.sentinelGetMasterAddrByName(FAILOVER_MASTER_NAME);
|
||||
HostAndPort newMaster = new HostAndPort(masterHostAndPort.get(0),
|
||||
Integer.parseInt(masterHostAndPort.get(1)));
|
||||
assertNotEquals(newMaster, currentMaster);
|
||||
} finally {
|
||||
j.close();
|
||||
}
|
||||
|
||||
assertNotEquals(newMaster, currentMaster);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void sentinelMonitor() {
|
||||
Jedis j = new Jedis(sentinel.getHost(), sentinel.getPort());
|
||||
|
||||
// monitor new master
|
||||
String result = j.sentinelMonitor(MONITOR_MASTER_NAME, MASTER_IP,
|
||||
master.getPort(), 1);
|
||||
assertEquals("OK", result);
|
||||
|
||||
// already monitored
|
||||
try {
|
||||
j.sentinelMonitor(MONITOR_MASTER_NAME, MASTER_IP, master.getPort(),
|
||||
1);
|
||||
fail();
|
||||
} catch (JedisDataException e) {
|
||||
// pass
|
||||
// monitor new master
|
||||
String result = j.sentinelMonitor(MONITOR_MASTER_NAME, MASTER_IP,
|
||||
master.getPort(), 1);
|
||||
assertEquals("OK", result);
|
||||
|
||||
// already monitored
|
||||
try {
|
||||
j.sentinelMonitor(MONITOR_MASTER_NAME, MASTER_IP,
|
||||
master.getPort(), 1);
|
||||
fail();
|
||||
} catch (JedisDataException e) {
|
||||
// pass
|
||||
}
|
||||
} finally {
|
||||
j.close();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -122,19 +135,23 @@ public class JedisSentinelTest extends JedisTestBase {
|
||||
public void sentinelRemove() {
|
||||
Jedis j = new Jedis(sentinel.getHost(), sentinel.getPort());
|
||||
|
||||
ensureMonitored(sentinel, REMOVE_MASTER_NAME, MASTER_IP,
|
||||
master.getPort(), 1);
|
||||
|
||||
String result = j.sentinelRemove(REMOVE_MASTER_NAME);
|
||||
assertEquals("OK", result);
|
||||
|
||||
// not exist
|
||||
try {
|
||||
result = j.sentinelRemove(REMOVE_MASTER_NAME);
|
||||
assertNotEquals("OK", result);
|
||||
fail();
|
||||
} catch (JedisDataException e) {
|
||||
// pass
|
||||
ensureMonitored(sentinel, REMOVE_MASTER_NAME, MASTER_IP,
|
||||
master.getPort(), 1);
|
||||
|
||||
String result = j.sentinelRemove(REMOVE_MASTER_NAME);
|
||||
assertEquals("OK", result);
|
||||
|
||||
// not exist
|
||||
try {
|
||||
result = j.sentinelRemove(REMOVE_MASTER_NAME);
|
||||
assertNotEquals("OK", result);
|
||||
fail();
|
||||
} catch (JedisDataException e) {
|
||||
// pass
|
||||
}
|
||||
} finally {
|
||||
j.close();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -142,24 +159,29 @@ public class JedisSentinelTest extends JedisTestBase {
|
||||
public void sentinelSet() {
|
||||
Jedis j = new Jedis(sentinel.getHost(), sentinel.getPort());
|
||||
|
||||
Map<String, String> parameterMap = new HashMap<String, String>();
|
||||
parameterMap.put("down-after-milliseconds", String.valueOf(1234));
|
||||
parameterMap.put("parallel-syncs", String.valueOf(3));
|
||||
parameterMap.put("quorum", String.valueOf(2));
|
||||
j.sentinelSet(MASTER_NAME, parameterMap);
|
||||
try {
|
||||
Map<String, String> parameterMap = new HashMap<String, String>();
|
||||
parameterMap.put("down-after-milliseconds", String.valueOf(1234));
|
||||
parameterMap.put("parallel-syncs", String.valueOf(3));
|
||||
parameterMap.put("quorum", String.valueOf(2));
|
||||
j.sentinelSet(MASTER_NAME, parameterMap);
|
||||
|
||||
List<Map<String, String>> masters = j.sentinelMasters();
|
||||
for (Map<String, String> master : masters) {
|
||||
if (master.get("name").equals(MASTER_NAME)) {
|
||||
assertEquals(1234,
|
||||
Integer.parseInt(master.get("down-after-milliseconds")));
|
||||
assertEquals(3, Integer.parseInt(master.get("parallel-syncs")));
|
||||
assertEquals(2, Integer.parseInt(master.get("quorum")));
|
||||
List<Map<String, String>> masters = j.sentinelMasters();
|
||||
for (Map<String, String> master : masters) {
|
||||
if (master.get("name").equals(MASTER_NAME)) {
|
||||
assertEquals(1234, Integer.parseInt(master
|
||||
.get("down-after-milliseconds")));
|
||||
assertEquals(3,
|
||||
Integer.parseInt(master.get("parallel-syncs")));
|
||||
assertEquals(2, Integer.parseInt(master.get("quorum")));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
parameterMap.put("quorum", String.valueOf(1));
|
||||
j.sentinelSet(MASTER_NAME, parameterMap);
|
||||
parameterMap.put("quorum", String.valueOf(1));
|
||||
j.sentinelSet(MASTER_NAME, parameterMap);
|
||||
} finally {
|
||||
j.close();
|
||||
}
|
||||
}
|
||||
|
||||
private void ensureMonitored(HostAndPort sentinel, String masterName,
|
||||
@@ -168,6 +190,8 @@ public class JedisSentinelTest extends JedisTestBase {
|
||||
try {
|
||||
j.sentinelMonitor(masterName, ip, port, quorum);
|
||||
} catch (JedisDataException e) {
|
||||
} finally {
|
||||
j.close();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -176,6 +200,9 @@ public class JedisSentinelTest extends JedisTestBase {
|
||||
try {
|
||||
j.sentinelRemove(masterName);
|
||||
} catch (JedisDataException e) {
|
||||
} finally {
|
||||
j.close();
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -155,7 +155,7 @@ public class ShardedJedisPoolTest extends Assert {
|
||||
// items on one shard
|
||||
// alter shard 1 and recreate pool
|
||||
pool.destroy();
|
||||
shards.set(1, new JedisShardInfo("nohost", 1234));
|
||||
shards.set(1, new JedisShardInfo("localhost", 1234));
|
||||
pool = new ShardedJedisPool(redisConfig, shards);
|
||||
jedis = pool.getResource();
|
||||
Long actual = Long.valueOf(0);
|
||||
|
||||
@@ -0,0 +1,17 @@
|
||||
package redis.clients.jedis.tests.utils;
|
||||
|
||||
public class FailoverAbortedException extends RuntimeException {
|
||||
private static final long serialVersionUID = 1925110762858409954L;
|
||||
|
||||
public FailoverAbortedException(String message) {
|
||||
super(message);
|
||||
}
|
||||
|
||||
public FailoverAbortedException(Throwable cause) {
|
||||
super(cause);
|
||||
}
|
||||
|
||||
public FailoverAbortedException(String message, Throwable cause) {
|
||||
super(message, cause);
|
||||
}
|
||||
}
|
||||
@@ -1,131 +1,60 @@
|
||||
package redis.clients.jedis.tests.utils;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
|
||||
import redis.clients.jedis.HostAndPort;
|
||||
import redis.clients.jedis.Jedis;
|
||||
import redis.clients.jedis.JedisPubSub;
|
||||
import redis.clients.jedis.tests.utils.FailoverAbortedException;
|
||||
|
||||
public class JedisSentinelTestUtil {
|
||||
|
||||
public static void waitForSentinelRecognizeRedisReplication(
|
||||
HostAndPort sentinel, String masterName, HostAndPort master,
|
||||
List<HostAndPort> slaves) throws InterruptedException {
|
||||
Jedis sentinelJedis = new Jedis(sentinel.getHost(), sentinel.getPort());
|
||||
while (true) {
|
||||
Thread.sleep(1000);
|
||||
|
||||
if (!isMasterRecognized(sentinelJedis, masterName, master)) {
|
||||
System.out.println("Master not recognized by Sentinel "
|
||||
+ sentinel.getHost() + ":" + sentinel.getPort()
|
||||
+ ", sleep...");
|
||||
continue;
|
||||
}
|
||||
|
||||
if (!isSlavesRecognized(sentinelJedis, masterName, slaves)) {
|
||||
System.out.println("Slaves not recognized by Sentinel "
|
||||
+ sentinel.getHost() + ":" + sentinel.getPort()
|
||||
+ ", sleep...");
|
||||
continue;
|
||||
}
|
||||
|
||||
// all recognized
|
||||
break;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
public static HostAndPort waitForNewPromotedMaster(HostAndPort sentinel,
|
||||
String masterName, HostAndPort oldMaster)
|
||||
public static HostAndPort waitForNewPromotedMaster(Jedis sentinelJedis)
|
||||
throws InterruptedException {
|
||||
Jedis sentinelJedis = new Jedis(sentinel.getHost(), sentinel.getPort());
|
||||
|
||||
final AtomicReference<String> newmaster = new AtomicReference<String>(
|
||||
"");
|
||||
|
||||
HostAndPort newMaster = null;
|
||||
while (true) {
|
||||
Thread.sleep(1000);
|
||||
sentinelJedis.psubscribe(new JedisPubSub() {
|
||||
|
||||
List<String> sentinelMasterInfos = sentinelJedis
|
||||
.sentinelGetMasterAddrByName(masterName);
|
||||
if (sentinelMasterInfos == null) {
|
||||
System.out
|
||||
.println("Cannot retrieve Sentinel's master address info, sleep...");
|
||||
continue;
|
||||
@Override
|
||||
public void onMessage(String channel, String message) {
|
||||
}
|
||||
|
||||
newMaster = new HostAndPort(sentinelMasterInfos.get(0),
|
||||
Integer.parseInt(sentinelMasterInfos.get(1)));
|
||||
@Override
|
||||
public void onPMessage(String pattern, String channel,
|
||||
String message) {
|
||||
if (channel.equals("+switch-master")) {
|
||||
newmaster.set(message);
|
||||
punsubscribe();
|
||||
} else if (channel.startsWith("-failover-abort")) {
|
||||
punsubscribe();
|
||||
throw new FailoverAbortedException("Unfortunately sentinel cannot failover... reason(channel) : " +
|
||||
channel + " / message : " + message);
|
||||
}
|
||||
}
|
||||
|
||||
if (!newMaster.equals(oldMaster))
|
||||
break;
|
||||
@Override
|
||||
public void onSubscribe(String channel, int subscribedChannels) {
|
||||
}
|
||||
|
||||
System.out
|
||||
.println("Sentinel's master is not yet changed, sleep...");
|
||||
}
|
||||
@Override
|
||||
public void onUnsubscribe(String channel, int subscribedChannels) {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onPUnsubscribe(String pattern, int subscribedChannels) {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onPSubscribe(String pattern, int subscribedChannels) {
|
||||
}
|
||||
}, "*");
|
||||
|
||||
String[] chunks = newmaster.get().split(" ");
|
||||
HostAndPort newMaster = new HostAndPort(chunks[3],
|
||||
Integer.parseInt(chunks[4]));
|
||||
|
||||
return newMaster;
|
||||
}
|
||||
|
||||
public static void waitForSentinelsRecognizeEachOthers()
|
||||
throws InterruptedException {
|
||||
// During failover, master has been changed
|
||||
// It means that sentinels need to recognize other sentinels from new
|
||||
// master's hello channel
|
||||
// Without recognizing, Sentinels cannot run failover
|
||||
|
||||
// Sentinels need to take some time to recognize each other...
|
||||
// http://redis.io/topics/sentinel
|
||||
// Sentinel Rule #8: Every Sentinel publishes a message to every
|
||||
// monitored master
|
||||
// Pub/Sub channel __sentinel__:hello, every five seconds, blabla...
|
||||
|
||||
// FIXME There're no command for sentinel to list recognized sentinels
|
||||
// so sleep wisely (channel's hello message interval + margin)
|
||||
Thread.sleep(5000 + 500);
|
||||
}
|
||||
|
||||
private static boolean isMasterRecognized(Jedis sentinelJedis,
|
||||
String masterName, HostAndPort master) {
|
||||
List<String> sentinelMasterInfos = sentinelJedis
|
||||
.sentinelGetMasterAddrByName(masterName);
|
||||
if (sentinelMasterInfos == null)
|
||||
return false;
|
||||
|
||||
HostAndPort sentinelMaster = new HostAndPort(
|
||||
sentinelMasterInfos.get(0),
|
||||
Integer.parseInt(sentinelMasterInfos.get(1)));
|
||||
|
||||
return sentinelMaster.equals(master);
|
||||
}
|
||||
|
||||
private static boolean isSlavesRecognized(Jedis sentinelJedis,
|
||||
String masterName, List<HostAndPort> slaves) {
|
||||
List<Map<String, String>> slavesMap = sentinelJedis
|
||||
.sentinelSlaves(masterName);
|
||||
|
||||
if (slavesMap.size() != slaves.size())
|
||||
return false;
|
||||
|
||||
int slavesRecognized = 0;
|
||||
|
||||
for (HostAndPort slave : slaves) {
|
||||
if (isSlaveFoundInSlavesMap(slavesMap, slave))
|
||||
slavesRecognized++;
|
||||
}
|
||||
|
||||
return slavesRecognized == slaves.size();
|
||||
}
|
||||
|
||||
private static boolean isSlaveFoundInSlavesMap(
|
||||
List<Map<String, String>> slavesMap, HostAndPort slave) {
|
||||
for (Map<String, String> slaveMap : slavesMap) {
|
||||
HostAndPort sentinelSlave = new HostAndPort(slaveMap.get("ip"),
|
||||
Integer.parseInt(slaveMap.get("port")));
|
||||
|
||||
if (sentinelSlave.equals(slave))
|
||||
return true;
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user