ClusterServiceImpl extends AbstractConsoleService<ClusterTblDao> implements ClusterService { @VisibleForTesting @DalTransaction protected void balanceCluster(String dcName, List<SetinelTbl> sentinels, final String cluster) { List<DcClusterShardTbl> dcClusterShards = dcClusterShardService.findAllByDcCluster(dcName, cluster); if(dcClusterShards == null || sentinels == null || sentinels.isEmpty()) { throw new XpipeRuntimeException("DcClusterShard | Sentinels should not be null"); } long randomlySelectedSentinelId = randomlyChoseSentinels(sentinels); dcClusterShards.forEach(dcClusterShard -> { dcClusterShard.setSetinelId(randomlySelectedSentinelId); try { dcClusterShardService.updateDcClusterShard(dcClusterShard); } catch (DalException e) { throw new XpipeRuntimeException(e.getMessage()); } }); }  @Override ClusterTbl find(final String clusterName); @Override List<ClusterTbl> findAllByNames(List<String> clusterNames); @Override ClusterStatus clusterStatus(String clusterName); @Override List<DcTbl> getClusterRelatedDcs(String clusterName); @Override ClusterTbl find(final long clusterId); @Override List<String> findAllClusterNames(); @Override Long getAllCount(); @Override @DalTransaction synchronized ClusterTbl createCluster(ClusterModel clusterModel); long getOrgIdFromClusterOrgName(ClusterTbl cluster); @Override ClusterTbl findClusterAndOrg(String clusterName); @Override List<ClusterTbl> findClustersWithOrgInfoByActiveDcId(long activeDc); @Override List<ClusterTbl> findAllClustersWithOrgInfo(); @Override List<ClusterTbl> findClustersWithOrgInfoByClusterType(String clusterType); @Override void updateCluster(String clusterName, ClusterTbl cluster); @Override void updateActivedcId(long id, long activeDcId); @Override void updateStatusById(long id, ClusterStatus clusterStatus); @Override void deleteCluster(String clusterName); @Override void bindDc(String clusterName, String dcName); @Override void unbindDc(String clusterName, String dcName); @Override void update(final ClusterTbl cluster); boolean checkEmails(String emails); @Override List<String> reBalanceSentinels(String dcName, final int numOfClusters, final boolean activeOnly); @Override void reBalanceClusterSentinels(String dcName, final List<String> clusters); @Override List<ClusterListUnhealthyClusterModel> findUnhealthyClusters(); @Override List<ClusterTbl> findAllClusterByDcNameBind(String dcName); @Override List<ClusterTbl> findActiveClustersByDcName(String dcName); @Override List<ClusterTbl> findAllClustersByDcName(String dcName); @Override List<ClusterTbl> findAllClusterByKeeperContainer(long keeperContainerId);  }
@Test public void testBalanceCluster() throws Exception { List<SetinelTbl> setinelTbls1 = Arrays.asList(new SetinelTbl().setSetinelId(999L), new SetinelTbl().setSetinelId(1000L), new SetinelTbl().setSetinelId(9999L)); List<SetinelTbl> setinelTbls2 = Arrays.asList(new SetinelTbl().setSetinelId(1999L), new SetinelTbl().setSetinelId(11000L), new SetinelTbl().setSetinelId(19999L)); Map<String, List<SetinelTbl>> map = Maps.newHashMap(); map.put(dcNames[0], setinelTbls1); map.put(dcNames[1], setinelTbls2); ClusterTbl clusterTbl = clusterService.find(clusterName); List<DcTbl> dcTbls = dcService.findAllDcs(); logger.info("dcTbls: {}", dcTbls); long []dcIds = new long[] {dcTbls.get(0).getId(), dcTbls.get(1).getId()}; Map<Long, SetinelTbl> dcToSentinel = Maps.newHashMap(); dcToSentinel.put(dcIds[0], setinelTbls1.get(0)); dcToSentinel.put(dcIds[1], setinelTbls2.get(0)); ShardTbl shard1 = new ShardTbl().setShardName("cluster1shard1test").setSetinelMonitorName("cluster1shard1monitor") .setClusterId(clusterTbl.getId()).setDeleted(false).setClusterInfo(clusterTbl); ShardTbl shard2 = new ShardTbl().setShardName("cluster1shard2test").setSetinelMonitorName("cluster1shard2monitor") .setClusterId(clusterTbl.getId()).setDeleted(false).setClusterInfo(clusterTbl); shardService.createShard(clusterName, shard1, dcToSentinel); dcToSentinel.put(dcIds[0], setinelTbls1.get(1)); dcToSentinel.put(dcIds[1], setinelTbls2.get(1)); shardService.createShard(clusterName, shard2, dcToSentinel); int checkTimes = 10; while(checkTimes -- > 0) { clusterService.reBalanceSentinels("jq", 10, true); DcClusterShardTbl dcClusterShardTbl1 = dcClusterShardService.find(dcNames[0], clusterName, shard1.getShardName()); DcClusterShardTbl dcClusterShardTbl2 = dcClusterShardService.find(dcNames[0], clusterName, shard2.getShardName()); Assert.assertEquals(dcClusterShardTbl1.getSetinelId(), dcClusterShardTbl2.getSetinelId()); } }