update current meta when keeper master switch

ctripcorp · Jan 17, 2025 · c1fd4f9 · c1fd4f9
1 parent a1edbe9
commit c1fd4f9
Show file tree

Hide file tree

Showing 5 changed files with 18 additions and 8 deletions.
diff --git a/...n/java/com/ctrip/xpipe/redis/meta/server/dcchange/impl/AbstractChangePrimaryDcAction.java b/...n/java/com/ctrip/xpipe/redis/meta/server/dcchange/impl/AbstractChangePrimaryDcAction.java
@@ -95,7 +95,10 @@ protected void makeKeepersOk(Long clusterDbId, Long shardDbId, Pair<String, Inte
 				currentMetaManager.getClusterRouteByDcId(currentMetaManager.getClusterMeta(clusterDbId).getActiveDc(), clusterDbId),
 				keyedObjectPool, 1000, 1, scheduled, executors);
 		try {
+			currentMetaManager.setKeeperMaster(clusterDbId, shardDbId, newMaster.getKey(), newMaster.getValue(), null);
+			// 必须先改 meta, 再修改 keepr, 不然可能被 KeeperStateAlignChecker reset 回去。
 			job.execute().get(waitTimeoutSeconds/2, TimeUnit.SECONDS);
+			logger.debug("[doRun][set]cluster_{}, shard_{}, {}", clusterDbId, shardDbId, newMaster);
 			executionLog.info("[makeKeepersOk]success");
 		} catch (InterruptedException | ExecutionException | TimeoutException e) {
 			logger.error("[makeKeepersOk]" + e.getMessage());

diff --git a/redis/redis-meta/src/main/java/com/ctrip/xpipe/redis/meta/server/impl/DefaultMetaServer.java b/redis/redis-meta/src/main/java/com/ctrip/xpipe/redis/meta/server/impl/DefaultMetaServer.java
@@ -243,13 +243,13 @@ public void updateUpstream(String clusterId, String shardId, String ip, int port
 						port);
 			} else {
 				logger.info("[updateUpstream]{},{},{},{}", clusterId, shardId, ip, port);
-				currentMetaManager.setKeeperMaster(clusterShard.getKey(), clusterShard.getValue(), ip, port);
+				currentMetaManager.setKeeperMaster(clusterShard.getKey(), clusterShard.getValue(), ip, port, null);
 			}
 		} else {
 			List<KeeperMeta> keepers = dcMetaCache.getShardKeepers(clusterShard.getKey(), clusterShard.getValue());
 			if (!CollectionUtils.isEmpty(keepers)) {
 				logger.info("[hetero][keeper][updateUpstream]{},{},{},{}", clusterId, shardId, ip, port);
-				currentMetaManager.setKeeperMaster(clusterShard.getKey(), clusterShard.getValue(), ip, port);
+				currentMetaManager.setKeeperMaster(clusterShard.getKey(), clusterShard.getValue(), ip, port, null);
 				return;
 			}
 

diff --git a/...m/ctrip/xpipe/redis/meta/server/keeper/keepermaster/impl/AbstractKeeperMasterChooser.java b/...m/ctrip/xpipe/redis/meta/server/keeper/keepermaster/impl/AbstractKeeperMasterChooser.java
@@ -4,6 +4,7 @@
 import com.ctrip.xpipe.redis.meta.server.meta.CurrentMetaManager;
 import com.ctrip.xpipe.redis.meta.server.meta.DcMetaCache;
 import com.ctrip.xpipe.tuple.Pair;
+import com.ctrip.xpipe.utils.StringUtil;
 
 import java.util.concurrent.ScheduledExecutorService;
 
@@ -33,6 +34,7 @@ public AbstractKeeperMasterChooser(Long clusterDbId, Long shardDbId, DcMetaCache
 
 	@Override
 	protected void work() {
+		String dcName = dcMetaCache.getPrimaryDc(clusterDbId, shardDbId);
 		Pair<String, Integer> keeperMaster = chooseKeeperMaster();
 		logger.debug("[doRun]cluster_{}, shard_{}, {}", clusterDbId, shardDbId, keeperMaster);
 		Pair<String, Integer> currentMaster = currentMetaManager.getKeeperMaster(clusterDbId, shardDbId);
@@ -41,7 +43,7 @@ protected void work() {
 			return;
 		}
 		logger.debug("[doRun][set]cluster_{}, shard_{}, {}", clusterDbId, shardDbId, keeperMaster);
-		currentMetaManager.setKeeperMaster(clusterDbId, shardDbId, keeperMaster.getKey(), keeperMaster.getValue());
+		currentMetaManager.setKeeperMaster(clusterDbId, shardDbId, keeperMaster.getKey(), keeperMaster.getValue(), dcName);
 	}
 
 	@Override

diff --git a/...s/redis-meta/src/main/java/com/ctrip/xpipe/redis/meta/server/meta/CurrentMetaManager.java b/...s/redis-meta/src/main/java/com/ctrip/xpipe/redis/meta/server/meta/CurrentMetaManager.java
@@ -77,7 +77,7 @@ public interface CurrentMetaManager extends Observable {
 
 	void setKeeperMaster(Long clusterDbId, Long shardDbId, String addr);
 
-	void setKeeperMaster(Long clusterDbId, Long shardDbId, String ip, int port);
+	void setKeeperMaster(Long clusterDbId, Long shardDbId, String ip, int port, String expectedPrimaryDc);
 
 	void setApplierMasterAndNotify(Long clusterDbId, Long shardDbId, String ip, int port, String sids);
 

diff --git a/.../src/main/java/com/ctrip/xpipe/redis/meta/server/meta/impl/DefaultCurrentMetaManager.java b/.../src/main/java/com/ctrip/xpipe/redis/meta/server/meta/impl/DefaultCurrentMetaManager.java
@@ -27,6 +27,7 @@
 import com.ctrip.xpipe.tuple.Pair;
 import com.ctrip.xpipe.utils.IpUtils;
 import com.ctrip.xpipe.utils.ObjectUtils;
+import com.ctrip.xpipe.utils.StringUtil;
 import com.ctrip.xpipe.utils.VisibleForTesting;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Sets;
@@ -533,9 +534,13 @@ public String getSrcSids(Long clusterDbId, Long shardDbId) {
 	}
 
 	@Override
-	public void setKeeperMaster(Long clusterDbId, Long shardDbId, String ip, int port) {
-
-
+	public void setKeeperMaster(Long clusterDbId, Long shardDbId, String ip, int port, String expectedPrimaryDc) {
+		String dcName = dcMetaCache.getPrimaryDc(clusterDbId, shardDbId);
+		if(expectedPrimaryDc != null && !StringUtil.trimEquals(dcName, expectedPrimaryDc)) {
+			// 如果 expectedDc 为null, 不进行校验。发生了dr切换，禁止修改。
+			// 如果任务基于 PrimaryDc 来修改 keeper meta 就要校验检测过程是否 dc 切换。
+			return;
+		}
 		Pair<String, Integer> keeperMaster = new Pair<String, Integer>(ip, port);
 		if(currentMeta.setKeeperMaster(clusterDbId, shardDbId, keeperMaster)){
 			logger.info("[setKeeperMaster]cluster_{},shard_{},{}:{}", clusterDbId, shardDbId, ip, port);
@@ -574,7 +579,7 @@ public void setKeeperMaster(Long clusterDbId, Long shardDbId, String addr) {
 
 		logger.info("[setKeeperMaster]cluster_{},shard_{},{}", clusterDbId, shardDbId, addr);
 		Pair<String, Integer> inetAddr = IpUtils.parseSingleAsPair(addr);
-		setKeeperMaster(clusterDbId, shardDbId, inetAddr.getKey(), inetAddr.getValue());
+		setKeeperMaster(clusterDbId, shardDbId, inetAddr.getKey(), inetAddr.getValue(), null);
 	}
 
 	@Override