DefaultCurrentMetaManager extends AbstractLifecycleObservable implements CurrentMetaManager, Observer { @VisibleForTesting protected void refreshKeeperMaster(ClusterMeta clusterMeta) { Set<String> shards = clusterMeta.getShards().keySet(); String clusterId = clusterMeta.getId(); for (String shardId : shards) { notifyKeeperMasterChanged(clusterId, shardId, getKeeperMaster(clusterId, shardId)); } } DefaultCurrentMetaManager(); @Override synchronized void addObserver(Observer observer); @Override Set<String> allClusters(); @Override void deleteSlot(int slotId); @Override void addSlot(int slotId); @Override void exportSlot(int slotId); @Override void importSlot(int slotId); @Override void update(Object args, Observable observable); @Override boolean hasCluster(String clusterId); @Override boolean hasShard(String clusterId, String shardId); @Override RedisMeta getRedisMaster(String clusterId, String shardId); @Override ClusterMeta getClusterMeta(String clusterId); @Override RouteMeta randomRoute(String clusterId); @Override Pair<String, Integer> getKeeperMaster(String clusterId, String shardId); @Override List<KeeperMeta> getSurviveKeepers(String clusterId, String shardId); @Override KeeperMeta getKeeperActive(String clusterId, String shardId); @Override String getCurrentMetaDesc(); @Override boolean updateKeeperActive(String clusterId, String shardId, KeeperMeta activeKeeper); @Override void addResource(String clusterId, String shardId, Releasable releasable); @Override void setSurviveKeepers(String clusterId, String shardId, List<KeeperMeta> surviceKeepers, KeeperMeta activeKeeper); @Override void setKeeperMaster(String clusterId, String shardId, String ip, int port); @Override void setKeeperMaster(String clusterId, String shardId, String addr); @Override boolean watchIfNotWatched(String clusterId, String shardId); @Override void setCurrentCRDTMaster(String clusterId, String shardId, long gid, String ip, int port); @Override RedisMeta getCurrentCRDTMaster(String clusterId, String shardId); @Override RedisMeta getCurrentMaster(String clusterId, String shardId); @Override void setPeerMaster(String dcId, String clusterId, String shardId, long gid, String ip, int port); @Override RedisMeta getPeerMaster(String dcId, String clusterId, String shardId); @Override Set<String> getUpstreamPeerDcs(String clusterId, String shardId); List<RedisMeta> getAllPeerMasters(String clusterId, String shardId); @Override void removePeerMaster(String dcId, String clusterId, String shardId); void setSlotManager(SlotManager slotManager); void setDcMetaCache(DcMetaCache dcMetaCache);  }
@Test public void testRefreshKeeperMaster() { currentMetaServerMetaManager = spy(new DefaultCurrentMetaManager()); currentMetaServerMetaManager.setDcMetaCache(dcMetaCache); String clusterId = getClusterId(); when(currentMetaServerMetaManager.allClusters()).thenReturn(Sets.newHashSet(clusterId)); Assert.assertFalse(currentMetaServerMetaManager.allClusters().isEmpty()); doReturn(Pair.from("127.0.0.1", randomPort())).when(currentMetaServerMetaManager).getKeeperMaster(anyString(), anyString()); when(dcMetaCache.randomRoute(clusterId)).thenReturn(new RouteMeta(1000).setTag(Route.TAG_META)); when(dcMetaCache.getClusterMeta(clusterId)).thenReturn(getCluster(getDcs()[0], clusterId)); int times = getCluster(getDcs()[0], clusterId).getShards().size(); currentMetaServerMetaManager.addMetaServerStateChangeHandler(handler); currentMetaServerMetaManager.routeChanges(); verify(handler, times(times)).keeperMasterChanged(eq(clusterId), anyString(), any()); }