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 testListenToNormalUri() throws PropertySerializationException
  {
    XdsToD2PropertiesAdaptorFixture fixture = new XdsToD2PropertiesAdaptorFixture();
    fixture.getSpiedAdaptor().listenToUris(PRIMARY_CLUSTER_NAME);

    verify(fixture._xdsClient).watchXdsResource(eq(PRIMARY_URI_RESOURCE_NAME), anyMapWatcher());
    XdsD2.D2URI protoUri = getD2URI(PRIMARY_CLUSTER_NAME, URI_NAME, VERSION);
    Map<String, XdsD2.D2URI> uriMap = new HashMap<>(Collections.singletonMap(URI_NAME, protoUri));
    fixture._uriMapWatcher.onChanged(new XdsClient.D2URIMapUpdate(uriMap));
    verify(fixture._uriEventBus).publishInitialize(PRIMARY_CLUSTER_NAME, _uriSerializer.fromProto(protoUri));
    verify(fixture._eventEmitter).emitSDStatusInitialRequestEvent(
        eq(PRIMARY_CLUSTER_NAME), eq(true), anyLong(), eq(true));
    // no status update receipt event emitted for initial update
    verify(fixture._eventEmitter, never()).emitSDStatusUpdateReceiptEvent(
        any(), any(), anyInt(), any(), anyBoolean(), any(), any(), any(), any(), any(), anyLong());

    // add uri 2
    uriMap.put(URI_NAME_2, getD2URI(PRIMARY_CLUSTER_NAME, URI_NAME_2, VERSION));
    fixture._uriMapWatcher.onChanged(new XdsClient.D2URIMapUpdate(uriMap));
    verify(fixture._eventEmitter).emitSDStatusInitialRequestEvent(
        eq(PRIMARY_CLUSTER_NAME), eq(true), anyLong(), eq(true)); // no more initial request event emitted
    verify(fixture._eventEmitter).emitSDStatusUpdateReceiptEvent( // status update receipt event emitted for added uri
        any(), eq(HOST_2), anyInt(), eq(ServiceDiscoveryEventEmitter.StatusUpdateActionType.MARK_READY), anyBoolean(),
        any(), any(), any(), eq((int) VERSION), any(), anyLong());

    // update uri 1, remove uri2, add uri3
    uriMap.clear();
    uriMap.put(URI_NAME, getD2URI(PRIMARY_CLUSTER_NAME, URI_NAME, VERSION_2));
    uriMap.put(URI_NAME_3, getD2URI(PRIMARY_CLUSTER_NAME, URI_NAME_3, VERSION));
    fixture._uriMapWatcher.onChanged(new XdsClient.D2URIMapUpdate(uriMap));
    // events should be emitted only for remove/add, but not update
    verify(fixture._eventEmitter, never()).emitSDStatusUpdateReceiptEvent(
        any(), eq(HOST_1), anyInt(), eq(ServiceDiscoveryEventEmitter.StatusUpdateActionType.MARK_READY), anyBoolean(),
        any(), any(), any(), eq((int) VERSION_2), any(), anyLong());
    verify(fixture._eventEmitter).emitSDStatusUpdateReceiptEvent(
        any(), eq(HOST_2), anyInt(), eq(StatusUpdateActionType.MARK_DOWN), anyBoolean(),
        any(), any(), any(), eq((int) VERSION), any(), anyLong());
    verify(fixture._eventEmitter).emitSDStatusUpdateReceiptEvent(
        any(), eq(HOST_3), anyInt(), eq(ServiceDiscoveryEventEmitter.StatusUpdateActionType.MARK_READY), anyBoolean(),
        any(), any(), any(), eq((int) VERSION), any(), anyLong());
  }