RebalanceOperationPerformer implements OperationPerformer<RebalanceOperation, RebalanceResult> { @VisibleForTesting RebalanceResult executeRebalanceOnDS(ManagementService managementService, InternalCache cache, String simulate, List<String> excludeRegionsList, FunctionExecutor functionExecutor) { RebalanceResultImpl rebalanceResult = new RebalanceResultImpl(); rebalanceResult.setSuccess(false); List<String> errors = new ArrayList<>(); List<MemberPRInfo> listMemberRegion = getMemberRegionList(managementService, cache, excludeRegionsList); if (listMemberRegion.isEmpty()) { rebalanceResult.setStatusMessage(CliStrings.REBALANCE__MSG__NO_REBALANCING_REGIONS_ON_DS); rebalanceResult.setSuccess(true); return rebalanceResult; } Iterator<MemberPRInfo> iterator = listMemberRegion.iterator(); boolean flagToContinueWithRebalance = false; while (iterator.hasNext()) { if (iterator.next().dsMemberList.size() > 1) { flagToContinueWithRebalance = true; break; } } if (!flagToContinueWithRebalance) { rebalanceResult.setStatusMessage(CliStrings.REBALANCE__MSG__NO_REBALANCING_REGIONS_ON_DS); rebalanceResult.setSuccess(true); return rebalanceResult; } List<RebalanceRegionResult> rebalanceRegionResults = new ArrayList<>(); for (MemberPRInfo memberPR : listMemberRegion) { try { if (memberPR.dsMemberList.size() > 1) { for (int i = 0; i < memberPR.dsMemberList.size(); i++) { DistributedMember dsMember = memberPR.dsMemberList.get(i); RebalanceFunction rebalanceFunction = new RebalanceFunction(); Object[] functionArgs = new Object[3]; functionArgs[0] = simulate; Set<String> regionSet = new HashSet<>(); regionSet.add(memberPR.region); functionArgs[1] = regionSet; Set<String> excludeRegionSet = new HashSet<>(); functionArgs[2] = excludeRegionSet; List<Object> resultList = new ArrayList<>(); try { if (checkMemberPresence(cache, dsMember)) { resultList = functionExecutor.execute(rebalanceFunction, functionArgs, dsMember); if (checkResultList(errors, resultList, dsMember)) { continue; } List<String> rstList = Arrays.asList(((String) resultList.get(0)).split(",")); rebalanceRegionResults.add(toRebalanceRegionResult(rstList)); rebalanceResult.setSuccess(true); break; } else { if (i == memberPR.dsMemberList.size() - 1) { break; } else { continue; } } } catch (Exception ex) { if (i == memberPR.dsMemberList.size() - 1) { errors.add( MessageFormat.format( CliStrings.REBALANCE__MSG__NO_EXECUTION_FOR_REGION_0_ON_MEMBERS_1, memberPR.region, listOfAllMembers(memberPR.dsMemberList)) + ", " + CliStrings.REBALANCE__MSG__REASON + ex.getMessage()); } else { continue; } } if (checkResultList(errors, resultList, dsMember)) { continue; } List<String> rstList = Arrays.asList(((String) resultList.get(0)).split(",")); rebalanceRegionResults.add(toRebalanceRegionResult(rstList)); rebalanceResult.setSuccess(true); } } } catch (NoMembersException e) { rebalanceResult.setStatusMessage(e.getMessage()); rebalanceResult.setRebalanceSummary(rebalanceRegionResults); return rebalanceResult; } } rebalanceResult.setRebalanceSummary(rebalanceRegionResults); if (rebalanceRegionResults.isEmpty()) { rebalanceResult.setSuccess(false); } return rebalanceResult; }  RebalanceResult perform(Cache cache, RebalanceOperation parameters); static DistributedMember getAssociatedMembers(String region, InternalCache cache); static List<MemberPRInfo> getMemberRegionList(ManagementService managementService,
      InternalCache cache,
      List<String> listExcludedRegion);  }
@Test public void executeRebalanceOnDSWithNoRegionsReturnsSuccessAndNoRegionMessage() { ManagementService managementService = mock(ManagementService.class); DistributedSystemMXBean distributedSystemMXBean = mock(DistributedSystemMXBean.class); when(distributedSystemMXBean.listRegions()).thenReturn(new String[] {}); when(managementService.getDistributedSystemMXBean()).thenReturn(distributedSystemMXBean); InternalDistributedSystem internalDistributedSystem = mock(InternalDistributedSystem.class); InternalCache cache = mock(InternalCache.class); when(cache.getInternalDistributedSystem()).thenReturn(internalDistributedSystem); when(internalDistributedSystem.getDistributionManager()) .thenReturn(mock(DistributionManager.class)); RebalanceOperationPerformer.FunctionExecutor functionExecutor = mock(RebalanceOperationPerformer.FunctionExecutor.class); RebalanceResult result = performer.executeRebalanceOnDS(managementService, cache, "true", null, functionExecutor); assertThat(result.getSuccess()).isTrue(); assertThat(result.getStatusMessage()) .isEqualTo("Distributed system has no regions that can be rebalanced."); }
@Test public void executeRebalanceOnDSWithOneRegionReturnsSuccessAndNoRegionMessage() { ManagementService managementService = mock(ManagementService.class); DistributedRegionMXBean regionMXBean = mock(DistributedRegionMXBean.class); when(regionMXBean.getRegionType()).thenReturn("PARTITION"); when(regionMXBean.getMembers()).thenReturn(new String[] {"member1"}); when(managementService.getDistributedRegionMXBean(SEPARATOR + "region1")) .thenReturn(regionMXBean); DistributedSystemMXBean distributedSystemMXBean = mock(DistributedSystemMXBean.class); when(distributedSystemMXBean.listRegions()).thenReturn(new String[] {"region1"}); when(managementService.getDistributedSystemMXBean()).thenReturn(distributedSystemMXBean); InternalDistributedSystem internalDistributedSystem = mock(InternalDistributedSystem.class); InternalCache cache = mock(InternalCache.class); when(cache.getInternalDistributedSystem()).thenReturn(internalDistributedSystem); DistributionManager distributionManager = mock(DistributionManager.class); InternalDistributedMember distributedMember = mock(InternalDistributedMember.class); when(distributedMember.getUniqueId()).thenReturn("member1"); Set<InternalDistributedMember> members = Collections.singleton(distributedMember); when(distributionManager.getDistributionManagerIds()).thenReturn(members); when(internalDistributedSystem.getDistributionManager()).thenReturn(distributionManager); RebalanceOperationPerformer.FunctionExecutor functionExecutor = mock(RebalanceOperationPerformer.FunctionExecutor.class); RebalanceResult result = performer.executeRebalanceOnDS(managementService, cache, "true", Collections.emptyList(), functionExecutor); assertThat(result.getSuccess()).isTrue(); assertThat(result.getStatusMessage()) .isEqualTo("Distributed system has no regions that can be rebalanced."); }
@Test public void executeRebalanceOnDSWithOneRegionOnTwoMembersReturnsSuccessAndRebalanceRegionResult() { ManagementService managementService = mock(ManagementService.class); DistributedRegionMXBean regionMXBean = mock(DistributedRegionMXBean.class); when(regionMXBean.getRegionType()).thenReturn("PARTITION"); when(regionMXBean.getMembers()).thenReturn(new String[] {"member1", "member2"}); when(managementService.getDistributedRegionMXBean(SEPARATOR + "region1")) .thenReturn(regionMXBean); DistributedSystemMXBean distributedSystemMXBean = mock(DistributedSystemMXBean.class); when(distributedSystemMXBean.listRegions()).thenReturn(new String[] {"region1"}); when(managementService.getDistributedSystemMXBean()).thenReturn(distributedSystemMXBean); InternalDistributedSystem internalDistributedSystem = mock(InternalDistributedSystem.class); InternalCache cache = mock(InternalCache.class); when(cache.getInternalDistributedSystem()).thenReturn(internalDistributedSystem); DistributionManager distributionManager = mock(DistributionManager.class); when(cache.getDistributionManager()).thenReturn(distributionManager); InternalDistributedMember distributedMember1 = mock(InternalDistributedMember.class); when(distributedMember1.getUniqueId()).thenReturn("member1"); InternalDistributedMember distributedMember2 = mock(InternalDistributedMember.class); when(distributedMember2.getUniqueId()).thenReturn("member2"); Set<InternalDistributedMember> members = new HashSet<>(); members.add(distributedMember1); members.add(distributedMember2); when(distributionManager.getDistributionManagerIds()).thenReturn(members); when(distributionManager.getNormalDistributionManagerIds()).thenReturn(members); when(internalDistributedSystem.getDistributionManager()).thenReturn(distributionManager); RebalanceOperationPerformer.FunctionExecutor functionExecutor = mock(RebalanceOperationPerformer.FunctionExecutor.class); List<Object> resultList = new ArrayList<>(); resultList.add("0,1,2,3,4,5,6,7,8,9," + SEPARATOR + "region1"); when(functionExecutor.execute(any(), any(), any())).thenReturn(resultList); RebalanceResult result = performer.executeRebalanceOnDS(managementService, cache, "true", Collections.emptyList(), functionExecutor); assertThat(result.getSuccess()).isTrue(); assertThat(result.getRebalanceRegionResults()).isNotNull(); assertThat(result.getRebalanceRegionResults()).hasSize(1); RebalanceRegionResult regionResult = result.getRebalanceRegionResults().get(0); assertThat(regionResult.getRegionName()).isEqualTo("region1"); assertThat(regionResult.getBucketCreateBytes()).isEqualTo(0); assertThat(regionResult.getBucketCreateTimeInMilliseconds()).isEqualTo(1); assertThat(regionResult.getBucketCreatesCompleted()).isEqualTo(2); assertThat(regionResult.getBucketTransferBytes()).isEqualTo(3); assertThat(regionResult.getBucketTransferTimeInMilliseconds()).isEqualTo(4); assertThat(regionResult.getBucketTransfersCompleted()).isEqualTo(5); assertThat(regionResult.getPrimaryTransferTimeInMilliseconds()).isEqualTo(6); assertThat(regionResult.getPrimaryTransfersCompleted()).isEqualTo(7); assertThat(regionResult.getTimeInMilliseconds()).isEqualTo(8); }