WeightedRandomRouterPolicy extends AbstractRouterPolicy { @Override public SubClusterId getHomeSubcluster( ApplicationSubmissionContext appSubmissionContext, List<SubClusterId> blacklist) throws YarnException { validate(appSubmissionContext); Map<SubClusterId, SubClusterInfo> activeSubclusters = getActiveSubclusters(); FederationPolicyUtils.validateSubClusterAvailability( new ArrayList<SubClusterId>(activeSubclusters.keySet()), blacklist); Map<SubClusterIdInfo, Float> weights = getPolicyInfo().getRouterPolicyWeights(); ArrayList<Float> weightList = new ArrayList<>(); ArrayList<SubClusterId> scIdList = new ArrayList<>(); for (Map.Entry<SubClusterIdInfo, Float> entry : weights.entrySet()) { if (blacklist != null && blacklist.contains(entry.getKey().toId())) { continue; } if (entry.getKey() != null && activeSubclusters.containsKey(entry.getKey().toId())) { weightList.add(entry.getValue()); scIdList.add(entry.getKey().toId()); } } int pickedIndex = FederationPolicyUtils.getWeightedRandom(weightList); if (pickedIndex == -1) { throw new FederationPolicyException( "No positive weight found on active subclusters"); } return scIdList.get(pickedIndex); }  @Override SubClusterId getHomeSubcluster(
      ApplicationSubmissionContext appSubmissionContext,
      List<SubClusterId> blacklist);  }
@Test public void testClusterChosenWithRightProbability() throws YarnException { ApplicationSubmissionContext context = mock(ApplicationSubmissionContext.class); when(context.getQueue()).thenReturn("queue1"); setApplicationSubmissionContext(context); Map<SubClusterId, AtomicLong> counter = new HashMap<>(); for (SubClusterIdInfo id : getPolicyInfo().getRouterPolicyWeights() .keySet()) { counter.put(id.toId(), new AtomicLong(0)); } float numberOfDraws = 10000; for (float i = 0; i < numberOfDraws; i++) { SubClusterId chosenId = ((FederationRouterPolicy) getPolicy()) .getHomeSubcluster(getApplicationSubmissionContext(), null); counter.get(chosenId).incrementAndGet(); } float totalActiveWeight = 0; for (SubClusterId id : getActiveSubclusters().keySet()) { SubClusterIdInfo idInfo = new SubClusterIdInfo(id); if (getPolicyInfo().getRouterPolicyWeights().containsKey(idInfo)) { totalActiveWeight += getPolicyInfo().getRouterPolicyWeights().get(idInfo); } } for (Map.Entry<SubClusterId, AtomicLong> counterEntry : counter .entrySet()) { float expectedWeight = getPolicyInfo().getRouterPolicyWeights() .get(new SubClusterIdInfo(counterEntry.getKey())) / totalActiveWeight; float actualWeight = counterEntry.getValue().floatValue() / numberOfDraws; if (getActiveSubclusters().containsKey(counterEntry.getKey())) { Assert.assertTrue( "Id " + counterEntry.getKey() + " Actual weight: " + actualWeight + " expected weight: " + expectedWeight, Math.abs(actualWeight - expectedWeight) < 0.01); } else { Assert .assertTrue( "Id " + counterEntry.getKey() + " Actual weight: " + actualWeight + " expected weight: " + expectedWeight, actualWeight == 0); } } }