@Override
  public void onNodeStatisticsChanged(NodeStatisticsChangeEvent event) {
    final Set<Integer> datanodeIds =
        configManager.getNodeManager().getRegisteredDataNodeLocations().keySet();
    final Map<Integer, Pair<NodeStatistics, NodeStatistics>> changes =
        event.getDifferentNodeStatisticsMap();
    for (final Map.Entry<Integer, Pair<NodeStatistics, NodeStatistics>> entry :
        changes.entrySet()) {
      final Integer nodeId = entry.getKey();
      final Pair<NodeStatistics, NodeStatistics> changeEvent = entry.getValue();
      if (!datanodeIds.contains(nodeId)) {
        continue;
      }
      if (changeEvent.getLeft() == null) {
        // if a new datanode registered, DO NOT trigger probing immediately
        startingDataNodes.add(nodeId);
        continue;
      } else {
        startingDataNodes.remove(nodeId);
      }

      final Set<Pair<Integer, Integer>> affectedPairs =
          heartbeats.keySet().stream()
              .filter(
                  pair ->
                      Objects.equals(pair.getLeft(), nodeId)
                          || Objects.equals(pair.getRight(), nodeId))
              .collect(Collectors.toSet());

      if (changeEvent.getRight() == null) {
        // datanode removed from cluster, clean up probing history
        affectedPairs.forEach(heartbeats::remove);
      } else {
        // we only trigger probing immediately if node comes around from UNKNOWN to RUNNING
        if (NodeStatus.Unknown.equals(changeEvent.getLeft().getStatus())
            && NodeStatus.Running.equals(changeEvent.getRight().getStatus())) {
          // let's clear the history when a new node comes around
          affectedPairs.forEach(pair -> heartbeats.put(pair, new ArrayList<>()));
          awaitForSignal.signal();
        }
      }
    }
  }
We only listen to datanode remove / restart / register events
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/TopologyService.java