public SubsetItem getClientsSubset(String serviceName,
      int minClusterSubsetSize,
      int partitionId,
      Map<URI, Double> possibleUris,
      long version,
      SimpleLoadBalancerState state)
  {
    SubsettingStrategy<URI> subsettingStrategy = _subsettingStrategyFactory.get(serviceName, minClusterSubsetSize, partitionId);

    if (subsettingStrategy == null)
    {
      return new SubsetItem(false, false, possibleUris, Collections.emptySet());
    }

    DeterministicSubsettingMetadata metadata = _subsettingMetadataProvider.getSubsettingMetadata(state);

    if (metadata == null)
    {
      return new SubsetItem(false, false, possibleUris, Collections.emptySet());
    }

    synchronized (_lockMap.computeIfAbsent(serviceName, name -> new Object()))
    {
      SubsetCache subsetCache = _subsetCache.get(serviceName);
      if (isCacheValid(version, metadata.getPeerClusterVersion(), minClusterSubsetSize, subsetCache))
      {
        if (subsetCache.getWeightedSubsets().containsKey(partitionId))
        {
          return new SubsetItem(true, false, subsetCache.getWeightedSubsets().get(partitionId), Collections.emptySet());
        }
      }

      Map<URI, Double> subsetMap = subsettingStrategy.getWeightedSubset(possibleUris, metadata);

      if (subsetMap == null)
      {
        return new SubsetItem(false, false, possibleUris, Collections.emptySet());
      }
      else
      {
        LOG.debug("Force updating subset cache for service " + serviceName);
        Set<URI> doNotSlowStartUris = new HashSet<>();

        if (subsetCache != null)
        {
          Set<URI> oldPossibleUris = subsetCache.getPossibleUris().getOrDefault(partitionId, Collections.emptySet());
          for (URI uri : subsetMap.keySet())
          {
            if (oldPossibleUris.contains(uri))
            {
              doNotSlowStartUris.add(uri);
            }
          }
          subsetCache.setVersion(version);
          subsetCache.setPeerClusterVersion(metadata.getPeerClusterVersion());
          subsetCache.setMinClusterSubsetSize(minClusterSubsetSize);
          subsetCache.getPossibleUris().put(partitionId, possibleUris.keySet());
          subsetCache.getWeightedSubsets().put(partitionId, subsetMap);
        }
        else
        {
          LOG.info("Cluster subsetting enabled for service: " + serviceName);
          Map<Integer, Set<URI>> servicePossibleUris = new HashMap<>();
          Map<Integer, Map<URI, Double>> serviceWeightedSubset = new HashMap<>();
          servicePossibleUris.put(partitionId, possibleUris.keySet());
          serviceWeightedSubset.put(partitionId, subsetMap);
          subsetCache = new SubsetCache(version, metadata.getPeerClusterVersion(),
              minClusterSubsetSize, servicePossibleUris, serviceWeightedSubset);

          _subsetCache.put(serviceName, subsetCache);
        }

        LOG.debug("Subset cache updated for service " + serviceName + ": " + subsetCache);

        return new SubsetItem(true, true, subsetMap, doNotSlowStartUris);
      }
    }
  }
@Test
  public void testDoNotSlowStart()
  {
    Mockito.when(_subsettingMetadataProvider.getSubsettingMetadata(_state))
        .thenReturn(new DeterministicSubsettingMetadata(0, 5, 0));

    Map<URI, Double> weightedUris = createUris(20);
    SubsettingState.SubsetItem subsetItem = _subsettingState.getClientsSubset(SERVICE_NAME, 4, 0,
        weightedUris, 0, _state);

    Map<URI, Double> weightedUris1 = createUris(40);
    SubsettingState.SubsetItem subsetItem1 = _subsettingState.getClientsSubset(SERVICE_NAME, 4, 0,
        weightedUris1, 1, _state);

    verifyDoNotSlowStart(subsetItem1.getWeightedUriSubset(), subsetItem1.getDoNotSlowStartUris(), weightedUris);
  }