DefaultKeeperStateChangeHandler extends AbstractLifecycle implements MetaServerStateChangeHandler, TopElement { @Override public void keeperActiveElected(String clusterId, String shardId, KeeperMeta activeKeeper) { logger.info("[keeperActiveElected]{},{},{}", clusterId, shardId, activeKeeper); List<KeeperMeta> keepers = currentMetaManager.getSurviveKeepers(clusterId, shardId); if (keepers == null || keepers.size() == 0) { logger.info("[keeperActiveElected][none keeper survive, do nothing]"); return; } Pair<String, Integer> activeKeeperMaster = currentMetaManager.getKeeperMaster(clusterId, shardId); KeeperStateChangeJob keeperStateChangeJob = createKeeperStateChangeJob(clusterId, keepers, activeKeeperMaster); if (!dcMetaCache.isCurrentDcPrimary(clusterId, shardId)) { List<RedisMeta> slaves = dcMetaCache.getShardRedises(clusterId, shardId); logger.info("[keeperActiveElected][current dc backup, set slave to new keeper]{},{},{}", clusterId, shardId, slaves); keeperStateChangeJob.setActiveSuccessCommand(new DefaultSlaveOfJob(slaves, activeKeeper.getIp(), activeKeeper.getPort(), clientPool, scheduled, executors)); } keyedOneThreadTaskExecutor.execute(new Pair<String, String>(clusterId, shardId), keeperStateChangeJob); }  @Override void keeperMasterChanged(String clusterId, String shardId, Pair<String, Integer> newMaster); @Override void keeperActiveElected(String clusterId, String shardId, KeeperMeta activeKeeper); void setcurrentMetaManager(CurrentMetaManager currentMetaManager); void setDcMetaCache(DcMetaCache dcMetaCache); void setClientPool(SimpleKeyedObjectPool<Endpoint, NettyClient> clientPool); void setScheduled(ScheduledExecutorService scheduled); void setExecutors(ExecutorService executors);  }
@Test public void testDifferentShardExecute() throws Exception{ String clusterId1 = clusterId + "1"; String shardId1 = shardId + "1"; when(currentMetaManager.getSurviveKeepers(clusterId1, shardId1)).thenReturn(keepers); when(currentMetaManager.getKeeperMaster(clusterId1, shardId1)).thenReturn(keeperMaster); when(dcMetaCache.isCurrentDcPrimary(clusterId1, shardId1)).thenReturn(true); handler.keeperActiveElected(clusterId, shardId, null); handler.keeperActiveElected(clusterId1, shardId1, null); sleep(setStateTimeMilli/2); Assert.assertEquals(2, calledCount.get()); }
@Test public void testSameShardExecute() throws Exception{ handler.keeperActiveElected(clusterId, shardId, null); handler.keeperActiveElected(clusterId, shardId, null); sleep(setStateTimeMilli/2); Assert.assertEquals(1, calledCount.get()); sleep(setStateTimeMilli*3/2); Assert.assertEquals(2, calledCount.get()); }