@Override
  public <K> HostToKeyMapper<K> getPartitionInformation(URI serviceUri, Collection<K> keys,
                                                        int limitHostPerPartition,
                                                        int hash)
          throws ServiceUnavailableException
  {
    if (limitHostPerPartition <= 0)
    {
      throw new IllegalArgumentException("limitHostPartition cannot be 0 or less");
    }
    ServiceProperties service = listenToServiceAndCluster(serviceUri);
    String serviceName = service.getServiceName();
    String clusterName = service.getClusterName();
    ClusterProperties cluster = getClusterProperties(serviceName, clusterName);

    LoadBalancerStateItem<UriProperties> uriItem = getUriItem(serviceName, clusterName, cluster);
    UriProperties uris = uriItem.getProperty();

    List<LoadBalancerState.SchemeStrategyPair> orderedStrategies =
            _state.getStrategiesForService(serviceName, service.getPrioritizedSchemes());
    Map<Integer, Integer> partitionWithoutEnoughHost = new HashMap<>();

    if (! orderedStrategies.isEmpty())
    {
      // get the partitionId -> keys mapping
      final PartitionAccessor accessor = getPartitionAccessor(serviceName, clusterName);
      int maxPartitionId = accessor.getMaxPartitionId();
      List<K> unmappedKeys = new ArrayList<>();
      Map<Integer, Set<K>> partitionSet = getPartitionSet(keys, accessor, unmappedKeys);

      // get the partitionId -> host URIs list
      Map<Integer, KeysAndHosts<K>> partitionDataMap = new HashMap<>();
      for (Integer partitionId : partitionSet.keySet())
      {
        for (LoadBalancerState.SchemeStrategyPair pair : orderedStrategies)
        {
          TrackerClientSubsetItem subsetItem = getPotentialClients(serviceName, service, cluster, uris,
              pair.getScheme(), partitionId, uriItem.getVersion());
          Map<URI, TrackerClient> trackerClients = subsetItem.getWeightedSubset();
          int size = Math.min(trackerClients.size(), limitHostPerPartition);
          List<URI> rankedUri = new ArrayList<>(size);

          Ring<URI> ring = pair.getStrategy().getRing(uriItem.getVersion(), partitionId, trackerClients,
              subsetItem.shouldForceUpdate());
          Iterator<URI> iterator = ring.getIterator(hash);

          while (iterator.hasNext() && rankedUri.size() < size)
          {
            URI uri = iterator.next();
            if (!rankedUri.contains(uri))
            {
              rankedUri.add(uri);
            }
          }
          if (rankedUri.size() < limitHostPerPartition)
          {
            partitionWithoutEnoughHost.put(partitionId, limitHostPerPartition - rankedUri.size());
          }

          KeysAndHosts<K> keysAndHosts = new KeysAndHosts<>(partitionSet.get(partitionId), rankedUri);
          partitionDataMap.put(partitionId, keysAndHosts);
          if (!rankedUri.isEmpty())
          {
            // don't go to the next strategy if there are already hosts in the current one
            break;
          }
        }
      }

      return new HostToKeyMapper<>(unmappedKeys, partitionDataMap, limitHostPerPartition, maxPartitionId + 1, partitionWithoutEnoughHost);
    }
    else
    {
      throw new ServiceUnavailableException(serviceName, "PEGA_1009. Unable to find a load balancer strategy" +
        "Server Schemes: [" + String.join(", ", service.getPrioritizedSchemes()) + ']');
    }
  }
@Test
  public void testGetAllPartitionMultipleHostsOrdering()
      throws Exception
  {
    String serviceName = "articles";
    String clusterName = "cluster";
    String path = "path";
    String strategyName = "degrader";

    //setup partition
    Map<URI,Map<Integer, PartitionData>> partitionDescriptions = new HashMap<>();

    final URI server1 = new URI("http://foo1.com");
    Map<Integer, PartitionData> server1Data = new HashMap<>();
    server1Data.put(1, new PartitionData(1.0));
    server1Data.put(2, new PartitionData(1.0));
    server1Data.put(3, new PartitionData(1.0));
    partitionDescriptions.put(server1, server1Data);

    final URI server2 = new URI("http://foo2.com");
    Map<Integer, PartitionData> server2Data = new HashMap<>();
    server2Data.put(1, new PartitionData(1.0));
    server2Data.put(2, new PartitionData(1.0));
    //server2Data.put(3, new PartitionData(1.0));
    partitionDescriptions.put(server2, server2Data);

    final URI server3 = new URI("http://foo3.com");
    Map<Integer, PartitionData> server3Data = new HashMap<>();
    server3Data.put(1, new PartitionData(1.0));
    server3Data.put(2, new PartitionData(1.0));
    //server3Data.put(3, new PartitionData(1.0));
    partitionDescriptions.put(server3, server3Data);

    //setup strategy which involves tweaking the hash ring to get partitionId -> URI host
    List<LoadBalancerState.SchemeStrategyPair> orderedStrategies = new ArrayList<>();
    LoadBalancerStrategy strategy = new TestLoadBalancerStrategy(partitionDescriptions);

    orderedStrategies.add(new LoadBalancerState.SchemeStrategyPair(PropertyKeys.HTTP_SCHEME, strategy));

    //setup the partition accessor which is used to get partitionId -> keys
    PartitionAccessor accessor = new TestPartitionAccessor();

    URI serviceURI = new URI("d2://" + serviceName);
    SimpleLoadBalancer balancer = new SimpleLoadBalancer(new PartitionedLoadBalancerTestState(
            clusterName, serviceName, path, strategyName, partitionDescriptions, orderedStrategies,
            accessor
    ), _d2Executor);

    HostToKeyMapper<URI> result = balancer.getPartitionInformation(serviceURI, null, 3, 123);

    Assert.assertEquals(result.getPartitionInfoMap().size(), 4);
    Assert.assertEquals(4, result.getPartitionCount());
    // partition 0 should be empty
    Assert.assertTrue(result.getPartitionInfoMap().get(0).getHosts().isEmpty());
    // partition 1 should have server1, server2 and server3.
    List<URI> ordering1 = result.getPartitionInfoMap().get(1).getHosts();

    List<URI> allServers = new ArrayList<>();
    allServers.add(server1);
    allServers.add(server2);
    allServers.add(server3);

    assertEquals(ordering1.size(), 3);
    Assert.assertTrue(ordering1.containsAll(allServers));

    // partition 2 should be the same as partition 1
    List<URI> ordering2 = result.getPartitionInfoMap().get(2).getHosts();
    Assert.assertEquals(ordering1, ordering2);
    // partition 3 should only contain server1
    List<URI> ordering3 = result.getPartitionInfoMap().get(3).getHosts();
    Assert.assertEquals(ordering3.get(0), server1);

    // partition 0 and partition 3 should not have enough hosts: lacking 3 and 2 respectively.
    Assert.assertTrue(result.getPartitionsWithoutEnoughHosts().containsKey(3));
    Assert.assertTrue(result.getPartitionsWithoutEnoughHosts().containsKey(0));
    Assert.assertEquals((int)result.getPartitionsWithoutEnoughHosts().get(3), 2);
    Assert.assertEquals((int)result.getPartitionsWithoutEnoughHosts().get(0), 3);
  }