@SuppressWarnings("java:S2276")
  public synchronized void start(boolean startConsensusPipes) throws IOException {
    stateMachine.start();
    MetricService.getInstance().addMetricSet(this.pipeConsensusServerMetrics);

    if (startConsensusPipes) {
      // start all consensus pipes
      final List<Peer> otherPeers = peerManager.getOtherPeers(thisNode);
      List<Peer> failedPipes =
          updateConsensusPipesStatus(new ArrayList<>(otherPeers), PipeStatus.RUNNING);
      // considering procedure can easily time out, keep trying updateConsensusPipesStatus until all
      // consensus pipes are started gracefully or exceed the maximum number of attempts.
      // NOTE: start pipe procedure is idempotent guaranteed.
      try {
        for (int i = 0; i < MAX_RETRY_TIMES && !failedPipes.isEmpty(); i++) {
          failedPipes = updateConsensusPipesStatus(failedPipes, PipeStatus.RUNNING);
          Thread.sleep(RETRY_WAIT_TIME_IN_MS);
        }
      } catch (InterruptedException e) {
        LOGGER.warn(
            "PipeConsensusImpl-peer{}: pipeConsensusImpl thread get interrupted when start consensus pipe. May because IoTDB process is killed.",
            thisNode);
        throw new IOException(String.format("%s cannot start all consensus pipes", thisNode));
      }
      // if there still are some consensus pipes failed to start, throw an exception.
      if (!failedPipes.isEmpty()) {
        // roll back
        List<Peer> successfulPipes = new ArrayList<>(otherPeers);
        successfulPipes.removeAll(failedPipes);
        updateConsensusPipesStatus(successfulPipes, PipeStatus.STOPPED);
        throw new IOException(String.format("%s cannot start all consensus pipes", thisNode));
      }
    }
    isStarted.set(true);
  }
PipeConsensusServerImpl is a consensus server implementation for pipe consensus.
iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/PipeConsensusServerImpl.java