@Override
  public <KEY> URIMappingResult<KEY> mapUris(List<URIKeyPair<KEY>> requestUriKeyPairs)
      throws ServiceUnavailableException
  {
    if (requestUriKeyPairs == null || requestUriKeyPairs.isEmpty())
    {
      return new URIMappingResult<>(Collections.emptyMap(), Collections.emptyMap(), Collections.emptyMap());
    }

    // API assumes that all requests will be made to the same service, just use the first request to get the service name and act as sample uri
    URI sampleURI = requestUriKeyPairs.get(0).getRequestUri();
    String serviceName = LoadBalancerUtil.getServiceNameFromUri(sampleURI);

    // To achieve scatter-gather, we require the following information
    PartitionAccessor accessor = _partitionInfoProvider.getPartitionAccessor(serviceName);
    Map<Integer, Ring<URI>> rings = _hashRingProvider.getRings(sampleURI);
    HashFunction<Request> hashFunction = _hashRingProvider.getRequestHashFunction(serviceName);

    Map<Integer, Set<KEY>> unmapped = new HashMap<>();

    // Pass One
    Map<Integer, List<URIKeyPair<KEY>>> requestsByPartition =
        distributeToPartitions(requestUriKeyPairs, accessor, unmapped);

    // Pass Two
    Map<URI, Integer> hostToParitionId = new HashMap<>();
    Map<URI, Set<KEY>> hostToKeySet = distributeToHosts(requestsByPartition, rings, hashFunction, hostToParitionId, unmapped);

    return new URIMappingResult<>(hostToKeySet, unmapped, hostToParitionId);
  }
@Test
  public void testUniversalStickiness() throws ServiceUnavailableException, URISyntaxException
  {
    int partitionCount = 4;
    int totalHostCount = 200;

    HashRingProvider ringProvider = createStaticHashRingProvider(totalHostCount, partitionCount, getHashFunction(true));
    HashFunction<Request> hashFunction = ringProvider.getRequestHashFunction(TEST_SERVICE);
    PartitionInfoProvider infoProvider = createRangeBasedPartitionInfoProvider(partitionCount);
    URIMapper mapper = new RingBasedUriMapper(ringProvider, infoProvider);

    URIKeyPair<Integer> request1 = new URIKeyPair<>(1, new URI("d2://testService/1")); // no partition, will be unmapped
    URIKeyPair<Integer> request2 = new URIKeyPair<>(2, new URI("d2://testService/2?partition=0")); // partition 0
    URIKeyPair<Integer> request3 = new URIKeyPair<>(3, new URI("d2://testService/3?partition=1")); // partition 1
    URIKeyPair<Integer> request4 = new URIKeyPair<>(4, new URI("d2://testService/4?partition=2")); // partition 2
    URIKeyPair<Integer> request5 = new URIKeyPair<>(5, new URI("d2://testService/5?partition=3")); // partition 3
    URIKeyPair<Integer> request6 =
        new URIKeyPair<>(6, new URI("d2://testService/6?partition=0")); // partition 0 with different sticky key
    URIKeyPair<Integer> request7 =
        new URIKeyPair<>(7, new URI("d2://testService/7?partition=1")); // partition 1 with different sticky key
    URIKeyPair<Integer> request8 =
        new URIKeyPair<>(8, new URI("d2://testService/8?partition=2")); // partition 2 with different sticky key
    URIKeyPair<Integer> request9 =
        new URIKeyPair<>(9, new URI("d2://testService/9?partition=3")); // partition 3 with different sticky key
    URIKeyPair<Integer> request10 =
        new URIKeyPair<>(10, new URI("d2://testService/10?partition=0&uuid=1"));// with extra parameters

    List<URIKeyPair<Integer>> requests =
        Arrays.asList(request1, request2, request3, request4, request5, request6, request7, request8, request9,
            request10);

    // uriMapper mapping
    URIMappingResult<Integer> uriMapperResult = mapper.mapUris(requests);

    // normal mapping
    Map<Integer, Set<Integer>> normalUnmapped = new HashMap<>();
    Map<URI, Set<Integer>> normalHostToKeySet = new HashMap<>();
    for (URIKeyPair<Integer> request : requests)
    {
      int partitionId = 0;
      try
      {
        partitionId = infoProvider.getPartitionAccessor(TEST_SERVICE).getPartitionId(request.getRequestUri());
      }
      catch (PartitionAccessException e)
      {
        normalUnmapped.computeIfAbsent(-1, k -> new HashSet<>()).add(request.getKey());
      }
      Ring<URI> ring = ringProvider.getRings(request.getRequestUri()).get(partitionId);
      URI uri = ring.get(hashFunction.hash(new URIRequest(request.getRequestUri())));
      normalHostToKeySet.computeIfAbsent(uri, k -> new HashSet<>());
      normalHostToKeySet.get(uri).add(request.getKey());
    }

    // they should have the same results
    Assert.assertEquals(uriMapperResult.getUnmappedKeys(), normalUnmapped);
    for (Map.Entry<URI, Set<Integer>> resolvedKeys : uriMapperResult.getMappedKeys().entrySet())
    {
      Set<Integer> uriMapperKeySet = resolvedKeys.getValue();
      Assert.assertTrue(normalHostToKeySet.containsKey(resolvedKeys.getKey()));
      Set<Integer> normalKeySet = normalHostToKeySet.get(resolvedKeys.getKey());
      Assert.assertEquals(uriMapperKeySet, normalKeySet);
    }
  }