@Override
  public DeterministicSubsettingMetadata getSubsettingMetadata(LoadBalancerState state)
  {
    if (_clusterName == null)
    {
      _log.debug("Peer cluster name not provided.");
      return null;
    }

    FutureCallback<DeterministicSubsettingMetadata> metadataFutureCallback = new FutureCallback<>();

    state.listenToCluster(_clusterName, (type, name) ->
    {
      LoadBalancerStateItem<UriProperties> uriItem = state.getUriProperties(_clusterName);

      synchronized (_lock)
      {
        if (uriItem.getVersion() != _peerClusterVersion)
        {
          _peerClusterVersion = uriItem.getVersion();
          UriProperties uriProperties = uriItem.getProperty();
          if (uriProperties != null)
          {
            // Sort the URIs so each client sees the same ordering
            List<String> sortedHosts = uriProperties.getPartitionDesc().keySet().stream()
                .map(URI::getHost)
                .sorted()
                .distinct()
                .collect(Collectors.toList());

            int instanceId = sortedHosts.indexOf(_hostName);

            if (instanceId >= 0)
            {
              _subsettingMetadata = new DeterministicSubsettingMetadata(instanceId, sortedHosts.size(),
                  _peerClusterVersion);
            }
            else
            {
              _subsettingMetadata = null;
            }
          }
          else
          {
            _subsettingMetadata = null;
          }

          _log.debug("Got deterministic subsetting metadata for cluster {}: {}", _clusterName, _subsettingMetadata);
        }
      }
      metadataFutureCallback.onSuccess(_subsettingMetadata);
    });

    try
    {
      return metadataFutureCallback.get(_timeout, _unit);
    } catch (InterruptedException | ExecutionException | TimeoutException e) {
      _log.warn("Failed to fetch deterministic subsetting metadata from ZooKeeper for cluster " + _clusterName, e);
      return null;
    }
  }
@Test
  public void testGetSubsettingMetadata()
  {
    List<String> schemes = new ArrayList<>();
    Map<Integer, PartitionData> partitionData = new HashMap<>(1);
    partitionData.put(DefaultPartitionAccessor.DEFAULT_PARTITION_ID, new PartitionData(1d));
    Map<URI, Map<Integer, PartitionData>> uriData = new HashMap<>();
    for (int i = 0; i < 10; i++)
    {
      uriData.put(URI.create("http://test" + i + ".linkedin.com:8888/test"), partitionData);
    }
    schemes.add("http");

    _state.listenToCluster("cluster-1", new LoadBalancerState.NullStateListenerCallback());
    _state.listenToService("service-1", new LoadBalancerState.NullStateListenerCallback());
    _clusterRegistry.put("cluster-1", new ClusterProperties("cluster-1", schemes));
    _uriRegistry.put("cluster-1", new UriProperties("cluster-1", uriData));
    _serviceRegistry.put("service-1", new ServiceProperties("service-1",
        "cluster-1",
        "/test", Collections.singletonList("random")));

    DeterministicSubsettingMetadata metadata = _metadataProvider.getSubsettingMetadata(_state);

    assertEquals(metadata.getInstanceId(), 2);
    assertEquals(metadata.getTotalInstanceCount(), 10);
    assertEquals(metadata.getPeerClusterVersion(), 5);

    uriData.remove(URI.create("http://test0.linkedin.com:8888/test"));
    _uriRegistry.put("cluster-1", new UriProperties("cluster-1", uriData));

    metadata = _metadataProvider.getSubsettingMetadata(_state);
    assertEquals(metadata.getInstanceId(), 1);
    assertEquals(metadata.getTotalInstanceCount(), 9);
    assertEquals(metadata.getPeerClusterVersion(), 7);

    uriData.remove(URI.create("http://test2.linkedin.com:8888/test"));
    _uriRegistry.put("cluster-1", new UriProperties("cluster-1", uriData));

    metadata = _metadataProvider.getSubsettingMetadata(_state);
    assertNull(metadata);
  }