public void waitTargetPeerToPeersTransmissionCompleted(Peer targetPeer)
      throws ConsensusGroupModifyPeerException {
    boolean isTransmissionCompleted = false;
    boolean isFirstCheck = true;

    try {
      while (!isTransmissionCompleted) {
        Thread.sleep(CHECK_TRANSMISSION_COMPLETION_INTERVAL_IN_MILLISECONDS);

        final List<String> consensusPipeNames =
            peerManager.getPeers().stream()
                .filter(peer -> !peer.equals(targetPeer))
                .map(peer -> new ConsensusPipeName(targetPeer, peer).toString())
                .collect(Collectors.toList());
        isTransmissionCompleted =
            isRemotePeerConsensusPipesTransmissionCompleted(
                targetPeer, consensusPipeNames, isFirstCheck);

        isFirstCheck = false;
      }
    } catch (InterruptedException e) {
      LOGGER.warn("{} is interrupted when waiting for transfer completed", thisNode, e);
      Thread.currentThread().interrupt();
      throw new ConsensusGroupModifyPeerException(
          String.format("%s is interrupted when waiting for transfer completed", thisNode), e);
    }
  }
Wait for the user written data up to firstCheck to be replicated
iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/PipeConsensusServerImpl.java