public synchronized void checkConsensusPipe(Map<ConsensusPipeName, PipeStatus> existedPipes) {
    final PipeStatus expectedStatus = isStarted.get() ? PipeStatus.RUNNING : PipeStatus.STOPPED;
    final Map<ConsensusPipeName, Peer> expectedPipes =
        peerManager.getOtherPeers(thisNode).stream()
            .collect(
                ImmutableMap.toImmutableMap(
                    peer -> new ConsensusPipeName(thisNode, peer), peer -> peer));

    existedPipes.forEach(
        (existedName, existedStatus) -> {
          if (!expectedPipes.containsKey(existedName)) {
            try {
              LOGGER.warn("{} drop consensus pipe [{}]", consensusGroupId, existedName);
              consensusPipeManager.updateConsensusPipe(existedName, PipeStatus.DROPPED);
            } catch (Exception e) {
              LOGGER.warn("{} cannot drop consensus pipe [{}]", consensusGroupId, existedName, e);
            }
          } else if (!expectedStatus.equals(existedStatus)) {
            try {
              LOGGER.warn(
                  "{} update consensus pipe [{}] to status {}",
                  consensusGroupId,
                  existedName,
                  expectedStatus);
              if (expectedStatus.equals(PipeStatus.RUNNING)) {
                // Do nothing. Because Pipe framework's metaSync will do that.
                return;
              }
              consensusPipeManager.updateConsensusPipe(existedName, expectedStatus);
            } catch (Exception e) {
              LOGGER.warn(
                  "{} cannot update consensus pipe [{}] to status {}",
                  consensusGroupId,
                  existedName,
                  expectedStatus,
                  e);
            }
          }
        });

    expectedPipes.forEach(
        (expectedName, expectedPeer) -> {
          if (!existedPipes.containsKey(expectedName)) {
            try {
              LOGGER.warn(
                  "{} create and update consensus pipe [{}] to status {}",
                  consensusGroupId,
                  expectedName,
                  expectedStatus);
              consensusPipeManager.createConsensusPipe(thisNode, expectedPeer);
              consensusPipeManager.updateConsensusPipe(expectedName, expectedStatus);
            } catch (Exception e) {
              LOGGER.warn(
                  "{} cannot create and update consensus pipe [{}] to status {}",
                  consensusGroupId,
                  expectedName,
                  expectedStatus,
                  e);
            }
          }
        });
  }
update given consensus pipes' status, returns the peer corresponding to the pipe that failed to
update
iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/PipeConsensusServerImpl.java