public void waitPeersToTargetPeerTransmissionCompleted(Peer targetPeer)
      throws ConsensusGroupModifyPeerException {
    boolean isTransmissionCompleted = false;
    boolean isFirstCheckForCurrentPeer = true;
    boolean isFirstCheckForOtherPeers = true;

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

        if (isConsensusPipesTransmissionCompleted(
            Collections.singletonList(new ConsensusPipeName(thisNode, targetPeer).toString()),
            isFirstCheckForCurrentPeer)) {
          final List<Peer> otherPeers = peerManager.getOtherPeers(thisNode);

          isTransmissionCompleted = true;
          for (Peer peer : otherPeers) {
            if (!peer.equals(targetPeer)) {
              isTransmissionCompleted &=
                  isRemotePeerConsensusPipesTransmissionCompleted(
                      peer,
                      Collections.singletonList(new ConsensusPipeName(peer, targetPeer).toString()),
                      isFirstCheckForOtherPeers);
            }
          }
          isFirstCheckForOtherPeers = false;
        }
        isFirstCheckForCurrentPeer = 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