DefaultXpipeMetaManager extends AbstractMetaManager implements XpipeMetaManager { @Override public boolean updateRedisMaster(String dc, String clusterId, String shardId, RedisMeta redisMaster) throws MetaException { String activeDc = getActiveDc(clusterId, shardId); if(!activeDc.equals(dc)){ throw new MetaException("active dc:" + activeDc + ", but given:" + dc + ", clusterID:" + clusterId); } ShardMeta shardMeta = getDirectShardMeta(dc, clusterId, shardId); if(shardMeta == null){ throw new MetaException(String.format("unfound shard %s,%s,%s", dc, clusterId, shardId)); } boolean found = false, changed = false; String newMaster = String.format("%s:%d", redisMaster.getIp(), redisMaster.getPort()); String oldRedisMaster = null; for(RedisMeta redisMeta : shardMeta.getRedises()){ if(redisMeta.getIp().equals(redisMaster.getIp()) && redisMeta.getPort().equals(redisMaster.getPort())){ found = true; if(!redisMeta.isMaster()){ logger.info("[updateRedisMaster][change redis to master]{}", redisMeta); redisMeta.setMaster(null); changed = true; }else{ logger.info("[updateRedisMaster][redis already master]{}", redisMeta); } }else{ if(redisMeta.isMaster()){ logger.info("[updateRedisMaster][change redis to slave]{}", redisMeta); oldRedisMaster = String.format("%s:%d", redisMeta.getIp(), redisMeta.getPort()); redisMeta.setMaster(getNewMasterofOldMaster(redisMeta, newMaster)); changed = true; } } } if(oldRedisMaster != null){ for(RedisMeta redisMeta : shardMeta.getRedises()){ if(oldRedisMaster.equalsIgnoreCase(redisMeta.getMaster())){ redisMeta.setMaster(newMaster); changed = true; } } for(KeeperMeta keeperMeta : shardMeta.getKeepers()){ if(oldRedisMaster.equalsIgnoreCase(keeperMeta.getMaster())){ keeperMeta.setMaster(newMaster); changed = true; } } } if(!found){ redisMaster.setParent(shardMeta); shardMeta.getRedises().add(redisMaster); changed = true; } return changed; } DefaultXpipeMetaManager(XpipeMeta xpipeMeta); private  DefaultXpipeMetaManager(String fileName); static XpipeMetaManager buildFromFile(String fileName); static XpipeMetaManager buildFromMeta(XpipeMeta xpipeMeta); XpipeMeta load(String fileName); @Override String getActiveDc(String clusterId, String shardId); @Override Set<String> getBackupDcs(String clusterId, String shardId); @Override Set<String> getRelatedDcs(String clusterId, String shardId); @Override Set<String> getDcClusters(String dc); @Override ClusterMeta getClusterMeta(String dc, String clusterId); @Override ClusterType getClusterType(String clusterId); ClusterMeta getDirectClusterMeta(String dc, String clusterId); @Override ShardMeta getShardMeta(String dc, String clusterId, String shardId); @SuppressWarnings("unchecked") @Override List<KeeperMeta> getKeepers(String dc, String clusterId, String shardId); @SuppressWarnings("unchecked") @Override List<RedisMeta> getRedises(String dc, String clusterId, String shardId); @Override KeeperMeta getKeeperActive(String dc, String clusterId, String shardId); @Override List<KeeperMeta> getKeeperBackup(String dc, String clusterId, String shardId); @Override MetaDesc findMetaDesc(HostPort hostPort); @Override Pair<String, RedisMeta> getRedisMaster(String clusterId, String shardId); @Override boolean noneKeeperActive(String dc, String clusterId, String shardId); @Override void setSurviveKeepers(String dcId, String clusterId, String shardId, List<KeeperMeta> surviveKeepers); @Override boolean updateKeeperActive(String dc, String clusterId, String shardId, KeeperMeta activeKeeper); String getFileName(); @Override List<MetaServerMeta> getMetaServers(String dc); @Override ZkServerMeta getZkServerMeta(String dc); @Override Set<String> getDcs(); @Override boolean updateRedisMaster(String dc, String clusterId, String shardId, RedisMeta redisMaster); @Override boolean dcExists(String dc); @Override KeeperContainerMeta getKeeperContainer(String dc, KeeperMeta keeperMeta); @Override void update(DcMeta dcMeta); @Override void update(String dcId, ClusterMeta clusterMeta); @Override ClusterMeta removeCluster(String dcId, String clusterId); @Override DcMeta getDcMeta(String dcId); @Override String getDcZone(String dcId); @Override List<KeeperMeta> getAllSurviceKeepers(String dcId, String clusterId, String shardId); @Override boolean hasCluster(String dcId, String clusterId); @Override boolean hasShard(String dcId, String clusterId, String shardId); @Override SentinelMeta getSentinel(String dc, String clusterId, String shardId); @Override String getSentinelMonitorName(String dc, String clusterId, String shardId); @Override void primaryDcChanged(String dc, String clusterId, String shardId, String newPrimaryDc); @Override List<RouteMeta> routes(String currentDc, String tag); @Override RouteMeta randomRoute(String currentDc, String tag, Integer orgId, String dstDc); @Override List<ClusterMeta> getSpecificActiveDcClusters(String currentDc, String clusterActiveDc); @Override String toString();  }
@Test public void testUpdateRedisMaster() throws MetaException { Pair<String, RedisMeta> redisMaster = metaManager.getRedisMaster(clusterId1, shardId); Assert.assertEquals(redisMaster.getKey(), "jq"); boolean result = metaManager.updateRedisMaster(redisMaster.getKey(), clusterId1, shardId, redisMaster.getValue()); Assert.assertTrue(!result); KeeperMeta activeKeeper = null; for (KeeperMeta keeperMeta : metaManager.getKeepers(dc, clusterId1, shardId)) { if (keeperMeta.getMaster() .equals(String.format("%s:%d", redisMaster.getValue().getIp(), redisMaster.getValue().getPort()))) { activeKeeper = keeperMeta; } } Assert.assertNotNull(activeKeeper); for (RedisMeta redis : metaManager.getRedises(dc, clusterId1, shardId)) { if (!redis.equals(redisMaster.getValue())) { String master = String.format("%s:%d", redis.getIp(), redis.getPort()); Assert.assertNotEquals(activeKeeper.getMaster(), master); result = metaManager.updateRedisMaster(redisMaster.getKey(), clusterId1, shardId, redis); Assert.assertTrue(result); KeeperMeta active = metaManager.getKeeperActive(redisMaster.getKey(), clusterId1, shardId); Assert.assertEquals(active.getMaster(), master); } } }