DefaultKeeperElectorManager extends AbstractCurrentMetaObserver implements KeeperElectorManager, Observer, TopElement { protected void updateShardLeader(String leaderLatchPath, List<ChildData> childrenData, String clusterId, String shardId){ List<String> childrenPaths = new LinkedList<>(); childrenData.forEach(childData -> childrenPaths.add(childData.getPath())); logger.info("[updateShardLeader]{}, {}, {}", clusterId, shardId, childrenPaths); List<String> sortedChildren = LockInternals.getSortedChildren("latch-", sorter, childrenPaths); List<KeeperMeta> survivalKeepers = new ArrayList<>(childrenData.size()); for(String path : sortedChildren){ for(ChildData childData : childrenData){ if(path.equals(childData.getPath())){ String data = new String(childData.getData()); KeeperMeta keeper = JsonCodec.INSTANCE.decode(data, KeeperMeta.class); survivalKeepers.add(keeper); break; } } } if(survivalKeepers.size() != childrenData.size()){ throw new IllegalStateException(String.format("[children data not equal with survival keepers]%s, %s", childrenData, survivalKeepers)); } KeeperActiveElectAlgorithm klea = keeperActiveElectAlgorithmManager.get(clusterId, shardId); KeeperMeta activeKeeper = klea.select(clusterId, shardId, survivalKeepers); currentMetaManager.setSurviveKeepers(clusterId, shardId, survivalKeepers, activeKeeper); }  @Override Set<ClusterType> getSupportClusterTypes(); void setKeeperActiveElectAlgorithmManager(
			KeeperActiveElectAlgorithmManager keeperActiveElectAlgorithmManager); static final int FIRST_PATH_CHILDREN_CACHE_SLEEP_MILLI; }
@Test public void testUpdateShardLeader(){ String prefix = "/path"; List<ChildData> dataList = new LinkedList<>(); final int portBegin = 4000; final int count = 3; dataList.add(new ChildData(prefix + "/" + randomString(10) +"-latch-02", null, JsonCodec.INSTANCE.encodeAsBytes(new KeeperMeta().setId("127.0.0.1").setPort(portBegin + 1)))); dataList.add(new ChildData(prefix + "/"+ randomString(10) + "-latch-03", null, JsonCodec.INSTANCE.encodeAsBytes(new KeeperMeta().setId("127.0.0.1").setPort(portBegin + 2)))); dataList.add(new ChildData(prefix + "/"+ randomString(10) + "-latch-01", null, JsonCodec.INSTANCE.encodeAsBytes(new KeeperMeta().setId("127.0.0.1").setPort(portBegin)))); keeperElectorManager.updateShardLeader(prefix, dataList, clusterMeta.getId(), shardMeta.getId()); verify(keeperActiveElectAlgorithm).select(eq(clusterMeta.getId()), eq(shardMeta.getId()), argThat(new BaseMatcher<List<KeeperMeta>>() { @Override public boolean matches(Object item) { List<KeeperMeta> keepers = (List<KeeperMeta>) item; if(keepers.size() != count){ return false; } KeeperMeta prefix = null; for(KeeperMeta keeperMeta : keepers){ if(prefix != null){ if(keeperMeta.getPort() < prefix.getPort()){ return false; } } prefix = keeperMeta; } return true; } @Override public void describeTo(Description description) { } })); }