@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 testGetPartitionInfoOrdering()
    throws Exception
  {
    String serviceName = "articles";
    String clusterName = "cluster";
    String path = "path";
    String strategyName = "degrader";

    // setup 3 partitions. Partition 1 and Partition 2 both have server1 - server3. Partition 3 only has server1.
    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));
    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));
    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 can only map keys from 1 - 3.
    PartitionAccessor accessor = new TestPartitionAccessor();

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

    List<Integer> keys = new ArrayList<>();
    keys.add(1);
    keys.add(2);
    keys.add(3);
    keys.add(123);

    HostToKeyMapper<Integer> result = balancer.getPartitionInformation(serviceURI, keys, 3, 123);

    Assert.assertEquals(result.getLimitHostPerPartition(), 3);

    Assert.assertEquals(1, result.getUnmappedKeys().size());
    Assert.assertEquals(123, (int)result.getUnmappedKeys().iterator().next().getKey());

    //partition 0 should be null
    Assert.assertNull(result.getPartitionInfoMap().get(0));
    // results for partition 1 should contain server1, server2 and server3
    KeysAndHosts<Integer> keysAndHosts1 = result.getPartitionInfoMap().get(1);
    assertEquals(keysAndHosts1.getKeys().size(), 1);
    assertEquals((int) keysAndHosts1.getKeys().iterator().next(), 1);
    List<URI> ordering1 = keysAndHosts1.getHosts();
    // results for partition 2 should be the same as partition1.
    KeysAndHosts<Integer> keysAndHosts2 = result.getPartitionInfoMap().get(2);
    assertEquals(keysAndHosts2.getKeys().size(), 1);
    assertEquals((int) keysAndHosts2.getKeys().iterator().next(), 2);
    List<URI> ordering2 = keysAndHosts2.getHosts();
    //for partition 3
    KeysAndHosts<Integer> keysAndHosts3 = result.getPartitionInfoMap().get(3);
    assertEquals(keysAndHosts3.getKeys().size(), 1);
    assertEquals((int) keysAndHosts3.getKeys().iterator().next(), 3);
    List<URI> ordering3 = keysAndHosts3.getHosts();

    // Just compare the size and contents of the list, not the ordering.
    assertEquals(ordering1.size(), 3);
    List<URI> allServers = new ArrayList<>();
    allServers.add(server1);
    allServers.add(server2);
    allServers.add(server3);
    Assert.assertTrue(ordering1.containsAll(allServers));
    Assert.assertTrue(ordering2.containsAll(allServers));
    Assert.assertEquals(ordering1, ordering2);
    Assert.assertEquals(ordering3.get(0), server1);

    Assert.assertTrue(result.getPartitionsWithoutEnoughHosts().containsKey(3));
    Assert.assertEquals((int)result.getPartitionsWithoutEnoughHosts().get(3), 2);
  }