private synchronized void topologyProbing() {
    // 1. get the latest datanode list
    final List<TDataNodeLocation> dataNodeLocations = new ArrayList<>();
    final Set<Integer> dataNodeIds = new HashSet<>();
    for (final TDataNodeConfiguration dataNodeConf :
        configManager.getNodeManager().getRegisteredDataNodes()) {
      final TDataNodeLocation location = dataNodeConf.getLocation();
      if (startingDataNodes.contains(location.getDataNodeId())) {
        continue; // we shall wait for internal endpoint to be ready
      }
      dataNodeLocations.add(location);
      dataNodeIds.add(location.getDataNodeId());
    }

    // 2. send the verify connection RPC to all datanodes
    final TNodeLocations nodeLocations = new TNodeLocations();
    nodeLocations.setDataNodeLocations(dataNodeLocations);
    nodeLocations.setConfigNodeLocations(Collections.emptyList());
    final Map<Integer, TDataNodeLocation> dataNodeLocationMap =
        configManager.getNodeManager().getRegisteredDataNodes().stream()
            .map(TDataNodeConfiguration::getLocation)
            .collect(Collectors.toMap(TDataNodeLocation::getDataNodeId, location -> location));
    final DataNodeAsyncRequestContext<TNodeLocations, TTestConnectionResp>
        dataNodeAsyncRequestContext =
            new DataNodeAsyncRequestContext<>(
                CnToDnAsyncRequestType.SUBMIT_TEST_DN_INTERNAL_CONNECTION_TASK,
                nodeLocations,
                dataNodeLocationMap);
    CnToDnInternalServiceAsyncRequestManager.getInstance()
        .sendAsyncRequestWithTimeoutInMs(dataNodeAsyncRequestContext, PROBING_TIMEOUT_MS);
    final List<TTestConnectionResult> results = new ArrayList<>();
    dataNodeAsyncRequestContext
        .getResponseMap()
        .forEach(
            (nodeId, resp) -> {
              if (resp.isSetResultList()) {
                results.addAll(resp.getResultList());
              }
            });

    // 3. collect results and update the heartbeat timestamps
    for (final TTestConnectionResult result : results) {
      final int fromDataNodeId =
          Optional.ofNullable(result.getSender().getDataNodeLocation())
              .map(TDataNodeLocation::getDataNodeId)
              .orElse(-1);
      final int toDataNodeId = result.getServiceProvider().getNodeId();
      if (result.isSuccess()
          && dataNodeIds.contains(fromDataNodeId)
          && dataNodeIds.contains(toDataNodeId)) {
        // testAllDataNodeConnectionWithTimeout ensures the heartbeats are Dn-Dn internally. Here we
        // just double-check.
        final List<AbstractHeartbeatSample> heartbeatHistory =
            heartbeats.computeIfAbsent(
                new Pair<>(fromDataNodeId, toDataNodeId), p -> new LinkedList<>());
        heartbeatHistory.add(new NodeHeartbeatSample(NodeStatus.Running));
        if (heartbeatHistory.size() > SAMPLING_WINDOW_SIZE) {
          heartbeatHistory.remove(0);
        }
      }
    }

    // 4. use failure detector to identify potential network partitions
    final Map<Integer, Set<Integer>> latestTopology =
        dataNodeLocations.stream()
            .collect(Collectors.toMap(TDataNodeLocation::getDataNodeId, k -> new HashSet<>()));
    for (final Map.Entry<Pair<Integer, Integer>, List<AbstractHeartbeatSample>> entry :
        heartbeats.entrySet()) {
      final int fromId = entry.getKey().getLeft();
      final int toId = entry.getKey().getRight();

      if (!entry.getValue().isEmpty()
          && !failureDetector.isAvailable(entry.getKey(), entry.getValue())) {
        LOGGER.debug("Connection from DataNode {} to DataNode {} is broken", fromId, toId);
      } else {
        Optional.ofNullable(latestTopology.get(fromId)).ifPresent(s -> s.add(toId));
      }
    }

    logAsymmetricPartition(latestTopology);

    // 5. notify the listeners on topology change
    if (shouldRun.get()) {
      topologyChangeListener.accept(latestTopology);
    }
  }
Schedule the {@link #topologyProbing} task either: 1. every PROBING_INTERVAL_MS interval. 2.
Manually triggered by outside events (node restart / register, etc.).
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/TopologyService.java