DefaultKeeperElectorManager extends AbstractCurrentMetaObserver implements KeeperElectorManager, Observer, TopElement { protected void observerShardLeader(final String clusterId, final String shardId) { logger.info("[observerShardLeader]{}, {}", clusterId, shardId); final String leaderLatchPath = MetaZkConfig.getKeeperLeaderLatchPath(clusterId, shardId); final CuratorFramework client = zkClient.get(); if(currentMetaManager.watchIfNotWatched(clusterId, shardId)){ try { logger.info("[observerShardLeader][add PathChildrenCache]{}, {}", clusterId, shardId); PathChildrenCache pathChildrenCache = new PathChildrenCache(client, leaderLatchPath, true, XpipeThreadFactory.create(String.format("PathChildrenCache:%s-%s", clusterId, shardId))); pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() { private AtomicBoolean isFirst = new AtomicBoolean(true); @Override public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception { if(isFirst.get()){ isFirst.set(false); try { logger.info("[childEvent][first sleep]{}", FIRST_PATH_CHILDREN_CACHE_SLEEP_MILLI); TimeUnit.MILLISECONDS.sleep(FIRST_PATH_CHILDREN_CACHE_SLEEP_MILLI); }catch (Exception e){ logger.error("[childEvent]", e); } } logger.info("[childEvent]{}, {}, {}, {}", clusterId, shardId, event.getType(), ZkUtils.toString(event.getData())); updateShardLeader(leaderLatchPath, pathChildrenCache.getCurrentData(), clusterId, shardId); } }); pathChildrenCache.start(); registerJob(clusterId, shardId, new Releasable() { @Override public void release() throws Exception { try{ logger.info("[release][release children cache]{}, {}", clusterId, shardId); pathChildrenCache.close(); }catch (Exception e){ logger.error(String.format("[release][release children cache]%s, %s", clusterId, shardId), e); } } }); } catch (Exception e) { logger.error("[observerShardLeader]" + clusterId + " " + shardId, e); } }else{ logger.warn("[observerShardLeader][already watched]{}, {}", clusterId, shardId); } }  @Override Set<ClusterType> getSupportClusterTypes(); void setKeeperActiveElectAlgorithmManager(
			KeeperActiveElectAlgorithmManager keeperActiveElectAlgorithmManager); static final int FIRST_PATH_CHILDREN_CACHE_SLEEP_MILLI; }
@Test public void testObserverShardLeader() throws Exception { String clusterId = clusterMeta.getId(); String shardId = shardMeta.getId(); when(currentMetaManager.watchIfNotWatched(anyString(), anyString())).thenReturn(true); keeperElectorManager.observerShardLeader(clusterId, shardId); addKeeperZkNode(clusterId, shardId, getZkClient()); sleep(sleepForZkMilli); verify(currentMetaManager).setSurviveKeepers(anyString(), anyString(), anyList(), any(KeeperMeta.class)); verify(currentMetaManager).addResource(anyString(), anyString(), any(Releasable.class)); when(currentMetaManager.watchIfNotWatched(anyString(), anyString())).thenReturn(false); keeperElectorManager.observerShardLeader(clusterId, shardId); sleep(sleepForZkMilli); verify(currentMetaManager).setSurviveKeepers(anyString(), anyString(), anyList(), any(KeeperMeta.class)); verify(currentMetaManager).addResource(anyString(), anyString(), any(Releasable.class)); }