public void listenToCluster(String clusterName)
  {
    // if cluster name is a symlink, watch for D2SymlinkNode instead
    String resourceName = D2_CLUSTER_NODE_PREFIX + clusterName;
    if (SymlinkUtil.isSymlinkNodeOrPath(clusterName))
    {
      listenToSymlink(clusterName, resourceName);
    }
    else
    {
      _watchedClusterResources.computeIfAbsent(clusterName, k ->
      {
        XdsClient.NodeResourceWatcher watcher = getClusterResourceWatcher(clusterName);
        _xdsClient.watchXdsResource(resourceName, watcher);
        return watcher;
      });
    }
  }
@Test
  public void testListenToClusterSymlink()
  {
    XdsToD2PropertiesAdaptorFixture fixture = new XdsToD2PropertiesAdaptorFixture();
    fixture.getSpiedAdaptor().listenToCluster(SYMLINK_NAME);

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

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

    // verify both cluster and uri data of the actual cluster is watched
    verify(fixture._xdsClient).watchXdsResource(eq(PRIMARY_CLUSTER_RESOURCE_NAME), anyNodeWatcher());
    verify(fixture._xdsClient).watchXdsResource(eq(PRIMARY_URI_RESOURCE_NAME), anyNodeWatcher());

    // update cluster data
    NodeResourceWatcher clusterNodeWatcher = fixture._nodeWatcher;
    clusterNodeWatcher.onChanged(getClusterNodeUpdate(PRIMARY_CLUSTER_NAME));

    // verify cluster data is published under symlink name and actual cluster name
    verify(fixture._clusterEventBus).publishInitialize(SYMLINK_NAME, PRIMARY_CLUSTER_PROPERTIES);
    verify(fixture._clusterEventBus).publishInitialize(PRIMARY_CLUSTER_NAME, PRIMARY_CLUSTER_PROPERTIES);

    // test update symlink to a new primary cluster
    String primaryClusterResourceName2 = CLUSTER_NODE_PREFIX + PRIMARY_CLUSTER_NAME_2;
    ClusterStoreProperties primaryClusterProperties2 = new ClusterStoreProperties(PRIMARY_CLUSTER_NAME_2);

    symlinkNodeWatcher.onChanged(getSymlinkNodeUpdate(primaryClusterResourceName2));

    verify(fixture._xdsClient).watchXdsResource(eq(primaryClusterResourceName2), anyNodeWatcher());
    verify(fixture._xdsClient).watchXdsResource(eq(URI_NODE_PREFIX + PRIMARY_CLUSTER_NAME_2), anyMapWatcher());
    verifyClusterNodeUpdate(fixture, PRIMARY_CLUSTER_NAME_2, SYMLINK_NAME, primaryClusterProperties2);

    // 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.
    clusterNodeWatcher.onChanged(getClusterNodeUpdate(PRIMARY_CLUSTER_NAME_2));

    verify(fixture._clusterEventBus).publishInitialize(PRIMARY_CLUSTER_NAME, primaryClusterProperties2);
    // verify symlink is published just once
    verify(fixture._clusterEventBus).publishInitialize(SYMLINK_NAME, primaryClusterProperties2);
  }