public void listenToUris(String clusterName)
  {
    // if cluster name is a symlink, watch for D2SymlinkNode instead
    String resourceName = D2_URI_NODE_PREFIX + clusterName;
    if (SymlinkUtil.isSymlinkNodeOrPath(clusterName))
    {
      listenToSymlink(clusterName, resourceName);
    }
    else
    {
      _watchedUriResources.computeIfAbsent(clusterName, k ->
      {
        XdsClient.D2URIMapResourceWatcher watcher = getUriResourceWatcher(clusterName);
        _xdsClient.watchXdsResource(resourceName, watcher);
        return watcher;
      });
    }
  }
@Test
  public void testListenToUriSymlink() throws PropertySerializationException
  {
    XdsToD2PropertiesAdaptorFixture fixture = new XdsToD2PropertiesAdaptorFixture();
    fixture.getSpiedAdaptor().listenToUris(SYMLINK_NAME);

    // verify symlink is watched
    verify(fixture._xdsClient).watchXdsResource(eq(URI_SYMLINK_RESOURCE_NAME), anyNodeWatcher());

    // update symlink data
    NodeResourceWatcher symlinkNodeWatcher = fixture._nodeWatcher;
    symlinkNodeWatcher.onChanged(getSymlinkNodeUpdate(PRIMARY_URI_RESOURCE_NAME));

    // verify actual cluster of the uris is watched
    verify(fixture._xdsClient).watchXdsResource(eq(PRIMARY_URI_RESOURCE_NAME), anyMapWatcher());

    // update uri data
    D2URIMapResourceWatcher watcher = fixture._uriMapWatcher;
    watcher.onChanged(new XdsClient.D2URIMapUpdate(Collections.emptyMap()));

    // verify uri data is merged and published under symlink name and the actual cluster name
    verify(fixture._uriEventBus).publishInitialize(SYMLINK_NAME, getDefaultUriProperties(SYMLINK_NAME));
    verify(fixture._uriEventBus).publishInitialize(PRIMARY_CLUSTER_NAME, getDefaultUriProperties(PRIMARY_CLUSTER_NAME));

    // test update symlink to a new primary cluster
    String primaryUriResourceName2 = URI_NODE_PREFIX + PRIMARY_CLUSTER_NAME_2;
    symlinkNodeWatcher.onChanged(getSymlinkNodeUpdate(primaryUriResourceName2));

    verify(fixture._xdsClient).watchXdsResource(eq(primaryUriResourceName2), anyMapWatcher());
    verifyUriUpdate(fixture, PRIMARY_CLUSTER_NAME_2, SYMLINK_NAME);

    // if the old primary cluster gets an update, it will be published under its original cluster name
    // since the symlink points to the new primary cluster now.

    XdsD2.D2URI protoUri = getD2URI(PRIMARY_CLUSTER_NAME, LOCAL_HOST_URI.toString(), VERSION);
    UriProperties uriProps = new UriPropertiesJsonSerializer().fromProto(protoUri);

    watcher.onChanged(new XdsClient.D2URIMapUpdate(Collections.singletonMap(URI_NAME, protoUri)));

    verify(fixture._uriEventBus).publishInitialize(PRIMARY_CLUSTER_NAME, uriProps);
    // no status update receipt event emitted when data was empty before the update
    verify(fixture._eventEmitter, never()).emitSDStatusUpdateReceiptEvent(
        eq(PRIMARY_CLUSTER_NAME),
        eq(LOCAL_HOST),
        eq(PORT),
        eq(ServiceDiscoveryEventEmitter.StatusUpdateActionType.MARK_READY),
        eq(true),
        eq(XDS_SERVER),
        eq(URI_NODE_PREFIX + PRIMARY_CLUSTER_NAME + "/" + URI_NAME),
        eq(protoUri.toString()),
        eq((int) VERSION),
        eq(TRACING_ID),
        anyLong()
    );
  }