fix keeper integration test

ctripcorp · Nov 21, 2023 · 8165fc6 · 8165fc6
1 parent e73cc7e
commit 8165fc6
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 12 deletions.
diff --git a/...egration-test/src/test/java/com/ctrip/xpipe/redis/integratedtest/keeper/KeeperPsync2.java b/...egration-test/src/test/java/com/ctrip/xpipe/redis/integratedtest/keeper/KeeperPsync2.java
@@ -13,6 +13,8 @@
 import java.util.LinkedList;
 import java.util.List;
 
+import static com.ctrip.xpipe.redis.core.protocal.MASTER_STATE.REDIS_REPL_CONNECTED;
+
 /**
  * @author wenchao.meng
  *
@@ -31,10 +33,7 @@ public void testKeeperPsync2() throws Exception {
 
 		// init
 		initKeepers();
-
-		sleep(2000);
-
-		assertSyncCount(redisKeeperServers, 1);
+		assertSyncCount(redisKeeperServers, totalKeepers - 1);
 
 		for (int i = 0; i < testRound; i++) {
 
@@ -64,19 +63,26 @@ public void testKeeperPsync2() throws Exception {
 			sendMessageToMaster(redisMaster, 10);
 			redisKeeperServers = currentKeepers;
 
-			sleep(2000);
-			assertSyncCount(redisKeeperServers, 1);
-			assertCommandsEquals(redisKeeperServers);
+			waitKeepersConnected(redisKeeperServers);
+			assertSyncCount(redisKeeperServers, totalKeepers - 1);
+			waitConditionUntilTimeOut(() -> checkCommandsEquals(redisKeeperServers));
 		}
 	}
 
-	private void assertCommandsEquals(List<RedisKeeperServer> redisKeeperServers) {
+	private void waitKeepersConnected(List<RedisKeeperServer> keeperServers) throws Exception {
+		for (RedisKeeperServer keeperServer: redisKeeperServers) {
+			waitConditionUntilTimeOut(() -> keeperServer.getRedisMaster().getMasterState().equals(REDIS_REPL_CONNECTED));
+		}
+	}
+
+	private boolean checkCommandsEquals(List<RedisKeeperServer> redisKeeperServers) {
 
 		long end = redisKeeperServers.get(0).getReplicationStore().getEndOffset();
 		for (int i = 1; i < redisKeeperServers.size(); i++) {
-			Assert.assertEquals(end, redisKeeperServers.get(i).getReplicationStore().getEndOffset());
+			if (end != redisKeeperServers.get(i).getReplicationStore().getEndOffset()) return false;
 		}
 
+		return true;
 	}
 
 	private void assertSyncCount(List<RedisKeeperServer> redisKeeperServers, int expectedSyncCnt) {
@@ -89,7 +95,7 @@ private void assertSyncCount(List<RedisKeeperServer> redisKeeperServers, int exp
 			partialError += redisKeeperServer.getKeeperMonitor().getKeeperStats().getPartialSyncErrorCount();
 		}
 		Assert.assertEquals(0, partialError);
-		Assert.assertEquals(redisKeeperServers.size() - 1, full);
+		Assert.assertEquals(expectedSyncCnt, full);
 	}
 
 	private void initKeepers() throws Exception {
@@ -121,6 +127,8 @@ private void initKeepers() throws Exception {
 			redisKeeperServers.add(redisKeeperServer);
 			lastKeeper = keeperMeta;
 		}
+
+		waitKeepersConnected(redisKeeperServers);
 	}
 
 	@Override

diff --git a/...rc/test/java/com/ctrip/xpipe/redis/keeper/applier/xsync/DefaultCommandDispatcherTest.java b/...rc/test/java/com/ctrip/xpipe/redis/keeper/applier/xsync/DefaultCommandDispatcherTest.java
@@ -102,7 +102,7 @@ public void testFilterPublish() {
         RedisOp gtidOp1 = new RedisSingleKeyOpGtidWrapper(string2Bytes("GTID ggg:1 0"), "ggg", op1);
         Assert.assertFalse(dispatcher.shouldFilter(gtidOp1));
 
-        RedisSingleKeyOp op2 = new RedisOpSingleKey(RedisOpType.PUBLISH, string2Bytes("publish xpipe-hetero-ppp 222"), null, null);
+        RedisSingleKeyOp op2 = new RedisOpSingleKey(RedisOpType.PUBLISH, string2Bytes("publish xpipe-asymmetric-ppp 222"), null, null);
         RedisOp gtidOp2 = new RedisSingleKeyOpGtidWrapper(string2Bytes("GTID ggg:1 0"), "ggg", op2);
         Assert.assertFalse(dispatcher.shouldFilter(gtidOp2));
 
@@ -113,7 +113,7 @@ public void testFilterPublish() {
 
         //test estimated size by the way
         Assert.assertEquals(15, gtidOp1.estimatedSize());
-        Assert.assertEquals(36, gtidOp2.estimatedSize());
+        Assert.assertEquals(40, gtidOp2.estimatedSize());
         Assert.assertEquals(23, gtidOp3.estimatedSize());
     }